jmri_loconet.beans

  1"""LocoNet-aware bean subclasses + their managers + LnPowerManager.
  2
  3* :class:`LnTurnout` overrides ``set_commanded_state`` to push
  4  ``OPC_SW_REQ`` and updates ``known_state`` from inbound ``OPC_SW_REP``.
  5* :class:`LnSensor` updates its state from inbound ``OPC_INPUT_REP``.
  6* :class:`LnPowerManager` sends ``OPC_GPON`` / ``OPC_GPOFF`` / ``OPC_IDLE``
  7  and tracks state from inbound copies.
  8"""
  9from __future__ import annotations
 10
 11from jmri_core.beans.power import PowerManager
 12from jmri_core.beans.sensor import Sensor
 13from jmri_core.beans.turnout import Turnout
 14from jmri_core.manager import Manager
 15from jmri_core.state import PowerState, SensorState, TurnoutState
 16
 17from jmri_loconet.messages import (
 18    LocoNetMessage,
 19    OpcGpOff,
 20    OpcGpOn,
 21    OpcIdle,
 22    OpcInputRep,
 23    OpcSwRep,
 24    OpcSwReq,
 25)
 26from jmri_loconet.traffic import LocoNetTrafficController
 27
 28
 29class LnTurnout(Turnout):
 30    """LocoNet-aware turnout: sends OPC_SW_REQ on commanded-state change."""
 31
 32    __slots__ = ("_address", "_tc")
 33
 34    def __init__(self, system_name: str, address: int, tc: LocoNetTrafficController) -> None:
 35        super().__init__(system_name)
 36        self._address = address
 37        self._tc = tc
 38
 39    @property
 40    def address(self) -> int:
 41        return self._address
 42
 43    async def set_commanded_state(self, state: TurnoutState) -> None:
 44        if state in (TurnoutState.CLOSED, TurnoutState.THROWN):
 45            await self._tc.send(OpcSwReq(
 46                address=self._address,
 47                closed=(state == TurnoutState.CLOSED),
 48                output_on=True,
 49            ))
 50        # Update the in-memory state via the base implementation; for
 51        # DIRECT-feedback turnouts known_state will mirror this until/unless
 52        # a layout report overrides it.
 53        await super().set_commanded_state(state)
 54
 55
 56class LnTurnoutManager(Manager[LnTurnout]):
 57    """Manager that constructs LocoNet turnouts and tracks them by address."""
 58
 59    type_letter = "T"
 60
 61    def __init__(self, system_prefix: str, *, tc: LocoNetTrafficController) -> None:
 62        super().__init__(system_prefix)
 63        self._tc = tc
 64        self._by_address: dict[int, LnTurnout] = {}
 65        self._unsub = tc.inbound.connect(self._on_message)
 66
 67    def _create(self, system_name: str) -> LnTurnout:
 68        address = _parse_address(system_name, self.full_prefix)
 69        bean = LnTurnout(system_name, address, self._tc)
 70        self._by_address[address] = bean
 71        return bean
 72
 73    async def deregister(self, bean: LnTurnout) -> None:
 74        self._by_address.pop(bean.address, None)
 75        await super().deregister(bean)
 76
 77    async def _on_message(self, msg: LocoNetMessage) -> None:
 78        if not isinstance(msg, OpcSwRep):
 79            return
 80        bean = self._by_address.get(msg.address)
 81        if bean is None:
 82            return
 83        state = TurnoutState.CLOSED if msg.closed else TurnoutState.THROWN
 84        await bean._set_known_state(state)
 85
 86
 87class LnSensor(Sensor):
 88    """LocoNet-aware sensor — read-only from the layout."""
 89
 90    __slots__ = ("_address",)
 91
 92    def __init__(self, system_name: str, address: int) -> None:
 93        super().__init__(system_name)
 94        self._address = address
 95
 96    @property
 97    def address(self) -> int:
 98        return self._address
 99
100
101class LnSensorManager(Manager[LnSensor]):
102    """Manager that constructs LocoNet sensors and listens for OPC_INPUT_REP."""
103
104    type_letter = "S"
105
106    def __init__(self, system_prefix: str, *, tc: LocoNetTrafficController) -> None:
107        super().__init__(system_prefix)
108        self._tc = tc
109        self._by_address: dict[int, LnSensor] = {}
110        self._unsub = tc.inbound.connect(self._on_message)
111
112    def _create(self, system_name: str) -> LnSensor:
113        address = _parse_address(system_name, self.full_prefix)
114        bean = LnSensor(system_name, address)
115        self._by_address[address] = bean
116        return bean
117
118    async def _on_message(self, msg: LocoNetMessage) -> None:
119        if not isinstance(msg, OpcInputRep):
120            return
121        bean = self._by_address.get(msg.address)
122        if bean is None:
123            # Sensors aren't always pre-created; clients can opt in via provide().
124            return
125        state = SensorState.ACTIVE if msg.active else SensorState.INACTIVE
126        await bean._set_known_state(state)
127
128
129class LnPowerManager(PowerManager):
130    """LocoNet track power. Sends OPC_GPON/OPC_GPOFF/OPC_IDLE."""
131
132    def __init__(self, tc: LocoNetTrafficController) -> None:
133        super().__init__()
134        self._tc = tc
135        self._unsub = tc.inbound.connect(self._on_message)
136
137    async def set_power(self, state: PowerState) -> None:
138        if state == PowerState.ON:
139            await self._tc.send(OpcGpOn())
140        elif state == PowerState.OFF:
141            await self._tc.send(OpcGpOff())
142        elif state == PowerState.IDLE:
143            await self._tc.send(OpcIdle())
144        await self._set_state(state)
145
146    async def _on_message(self, msg: LocoNetMessage) -> None:
147        if isinstance(msg, OpcGpOn):
148            await self._set_state(PowerState.ON)
149        elif isinstance(msg, OpcGpOff):
150            await self._set_state(PowerState.OFF)
151        elif isinstance(msg, OpcIdle):
152            await self._set_state(PowerState.IDLE)
153
154
155# ----- helpers --------------------------------------------------------------
156
157
158def _parse_address(system_name: str, full_prefix: str) -> int:
159    """Extract the numeric address from a LocoNet system name (``"LT12"`` → 12)."""
160    rest = system_name.removeprefix(full_prefix)
161    try:
162        return int(rest)
163    except ValueError as e:
164        raise ValueError(
165            f"LocoNet system name {system_name!r} must end in a numeric address"
166        ) from e