jmri_core.manager

  1"""``Manager[T]`` — the per-type bean registry.
  2
  3Every JMRI bean type (turnouts, sensors, signal heads, …) has a manager that
  4owns the set of beans of that type for one or more system connections. Two
  5lookup keys are supported:
  6
  7  * ``system_name`` — globally unique, e.g. ``"LT12"``;
  8  * ``user_name``   — optional human label, e.g. ``"Main Street"``.
  9
 10The canonical operation is :meth:`provide` — get-or-create by either key.
 11Drivers subclass :class:`Manager` and override :meth:`_create` to build
 12their concrete bean type.
 13"""
 14from __future__ import annotations
 15
 16from collections.abc import Iterator
 17from typing import Any
 18
 19from jmri_core.bean import NamedBean
 20from jmri_core.events import Signal
 21from jmri_core.names import SystemName
 22
 23
 24class Manager[T: NamedBean]:
 25    """Per-type bean registry. Subclass and override :meth:`_create`.
 26
 27    Attributes set on the class describe the bean type:
 28
 29    * ``type_letter`` — single uppercase letter the bean class owns
 30      (e.g. ``"T"`` for Turnout). Must be set by subclass.
 31    * ``system_prefix`` — connection prefix (e.g. ``"L"``). Set per-instance
 32      because one bean class can have many connection-specific managers.
 33    """
 34
 35    type_letter: str = ""  # subclass must override
 36
 37    def __init__(self, system_prefix: str) -> None:
 38        if not self.type_letter:
 39            raise TypeError(
 40                f"{type(self).__name__}.type_letter must be set on the class"
 41            )
 42        if not system_prefix:
 43            raise ValueError("system_prefix must be a non-empty string")
 44        self.system_prefix = system_prefix
 45        self._by_system: dict[str, T] = {}
 46        self._by_user: dict[str, T] = {}
 47        self.added: Signal[T] = Signal()
 48        self.removed: Signal[T] = Signal()
 49
 50    # ------------------------------------------------------------------ public
 51
 52    @property
 53    def full_prefix(self) -> str:
 54        """The connection prefix + type letter, e.g. ``"LT"``."""
 55        return f"{self.system_prefix}{self.type_letter}"
 56
 57    def __iter__(self) -> Iterator[T]:
 58        return iter(self._by_system.values())
 59
 60    def __len__(self) -> int:
 61        return len(self._by_system)
 62
 63    def __contains__(self, name: str) -> bool:
 64        return name in self._by_system or name in self._by_user
 65
 66    def get_by_system_name(self, name: str) -> T | None:
 67        return self._by_system.get(name)
 68
 69    def get_by_user_name(self, name: str) -> T | None:
 70        return self._by_user.get(name)
 71
 72    def get(self, name: str) -> T | None:
 73        """Look up by either system or user name; system first."""
 74        return self._by_system.get(name) or self._by_user.get(name)
 75
 76    def named_bean_set(self) -> list[T]:
 77        """Snapshot list, JMRI naming."""
 78        return list(self._by_system.values())
 79
 80    async def provide(self, name: str) -> T:
 81        """Get-or-create. Accepts either system name or address-only form.
 82
 83        Address-only example: passing ``"12"`` to a LocoNet turnout manager
 84        with ``system_prefix='L'`` resolves to ``"LT12"``. Passing the full
 85        system name ``"LT12"`` is also accepted.
 86        """
 87        if name in self._by_user:
 88            return self._by_user[name]
 89        sysname = self._normalize_name(name)
 90        if sysname in self._by_system:
 91            return self._by_system[sysname]
 92        bean = self._create(sysname)
 93        await self.register(bean)
 94        return bean
 95
 96    async def register(self, bean: T) -> None:
 97        """Add an externally constructed *bean* to the registry."""
 98        if bean.system_name in self._by_system:
 99            raise ValueError(f"bean with system_name {bean.system_name!r} already registered")
100        self._by_system[bean.system_name] = bean
101        if bean.user_name:
102            if bean.user_name in self._by_user:
103                raise ValueError(f"user_name {bean.user_name!r} already in use")
104            self._by_user[bean.user_name] = bean
105        await self.added.emit(bean)
106
107    async def deregister(self, bean: T) -> None:
108        """Remove *bean* from the registry. No-op if not present."""
109        if bean.system_name not in self._by_system:
110            return
111        del self._by_system[bean.system_name]
112        if bean.user_name and self._by_user.get(bean.user_name) is bean:
113            del self._by_user[bean.user_name]
114        await self.removed.emit(bean)
115
116    # ---------------------------------------------------------------- internal
117
118    def _normalize_name(self, name: str) -> str:
119        """Turn *name* into a full system name. Accepts either form."""
120        if name.startswith(self.full_prefix):
121            return name
122        # Treat as bare address. Validate it parses cleanly.
123        return SystemName.make(self.system_prefix, self.type_letter, name)
124
125    def _create(self, system_name: str) -> T:
126        """Subclasses must override to construct their concrete bean type."""
127        raise NotImplementedError(
128            f"{type(self).__name__}._create must be implemented by subclasses"
129        )
130
131    # Convenience for tests / persistence ------------------------------------
132
133    def _set_user_name(self, bean: T, new_name: str | None) -> None:
134        """Rename *bean* maintaining the user-name index. Internal."""
135        # Remove old binding if any
136        old = bean.user_name
137        if old and self._by_user.get(old) is bean:
138            del self._by_user[old]
139        bean.user_name = new_name
140        if new_name:
141            if new_name in self._by_user and self._by_user[new_name] is not bean:
142                raise ValueError(f"user_name {new_name!r} already in use")
143            self._by_user[new_name] = bean
144
145    def to_dict(self) -> dict[str, Any]:
146        """Serializable snapshot used by persistence layer."""
147        return {
148            "type_letter": self.type_letter,
149            "system_prefix": self.system_prefix,
150            "beans": [
151                {
152                    "system_name": b.system_name,
153                    "user_name": b.user_name,
154                    "comment": b.comment,
155                }
156                for b in self._by_system.values()
157            ],
158        }