jmri_dccex.beans

  1"""DCC-EX-aware beans + their managers + power manager.
  2
  3Wire-protocol summary (CommandStation-EX docs):
  4
  5* Turnouts: ``<T id state>`` where state ∈ {0,1}. Response: ``<H id state>``.
  6* Sensors:  ``<Q id>`` = active, ``<q id>`` = inactive (server → client).
  7* Power:    ``<1>`` = on, ``<0>`` = off. Response: ``<p state>``.
  8"""
  9from __future__ import annotations
 10
 11import contextlib
 12
 13from jmri_core.beans.power import PowerManager
 14from jmri_core.beans.sensor import Sensor
 15from jmri_core.beans.turnout import Turnout
 16from jmri_core.manager import Manager
 17from jmri_core.state import PowerState, SensorState, TurnoutState
 18
 19from jmri_dccex.codec import DccexMessage
 20from jmri_dccex.traffic import DccexTrafficController
 21
 22# ----- Turnout -------------------------------------------------------------
 23
 24
 25class DccexTurnout(Turnout):
 26    """Sends ``<T id state>`` on commanded-state change."""
 27
 28    __slots__ = ("_address", "_tc")
 29
 30    def __init__(self, system_name: str, address: int, tc: DccexTrafficController) -> None:
 31        super().__init__(system_name)
 32        self._address = address
 33        self._tc = tc
 34
 35    @property
 36    def address(self) -> int:
 37        return self._address
 38
 39    async def set_commanded_state(self, state: TurnoutState) -> None:
 40        wire = 1 if state == TurnoutState.CLOSED else 0
 41        await self._tc.send(DccexMessage(command="T", args=(str(self._address), str(wire))))
 42        await super().set_commanded_state(state)
 43
 44
 45class DccexTurnoutManager(Manager[DccexTurnout]):
 46    type_letter = "T"
 47
 48    def __init__(self, system_prefix: str, *, tc: DccexTrafficController) -> None:
 49        super().__init__(system_prefix)
 50        self._tc = tc
 51        self._by_address: dict[int, DccexTurnout] = {}
 52        self._unsub = tc.inbound.connect(self._on_message)
 53
 54    def _create(self, system_name: str) -> DccexTurnout:
 55        address = _parse_address(system_name, self.full_prefix)
 56        bean = DccexTurnout(system_name, address, self._tc)
 57        self._by_address[address] = bean
 58        return bean
 59
 60    def get_by_address(self, address: int) -> DccexTurnout | None:
 61        return self._by_address.get(address)
 62
 63    async def _on_message(self, msg: DccexMessage) -> None:
 64        # <H id state> — turnout state echo from CS.
 65        if msg.command != "H" or len(msg.args) < 2:
 66            return
 67        try:
 68            addr = int(msg.args[0])
 69            state_int = int(msg.args[1])
 70        except ValueError:
 71            return
 72        bean = self._by_address.get(addr)
 73        if bean is None:
 74            return
 75        state = TurnoutState.CLOSED if state_int else TurnoutState.THROWN
 76        await bean._set_known_state(state)
 77
 78
 79# ----- Sensor --------------------------------------------------------------
 80
 81
 82class DccexSensor(Sensor):
 83    __slots__ = ("_address",)
 84
 85    def __init__(self, system_name: str, address: int) -> None:
 86        super().__init__(system_name)
 87        self._address = address
 88
 89    @property
 90    def address(self) -> int:
 91        return self._address
 92
 93
 94class DccexSensorManager(Manager[DccexSensor]):
 95    type_letter = "S"
 96
 97    def __init__(self, system_prefix: str, *, tc: DccexTrafficController) -> None:
 98        super().__init__(system_prefix)
 99        self._by_address: dict[int, DccexSensor] = {}
100        self._unsub = tc.inbound.connect(self._on_message)
101
102    def _create(self, system_name: str) -> DccexSensor:
103        address = _parse_address(system_name, self.full_prefix)
104        bean = DccexSensor(system_name, address)
105        self._by_address[address] = bean
106        return bean
107
108    def get_by_address(self, address: int) -> DccexSensor | None:
109        return self._by_address.get(address)
110
111    async def _on_message(self, msg: DccexMessage) -> None:
112        # <Q id> active, <q id> inactive.
113        if msg.command not in ("Q", "q") or not msg.args:
114            return
115        with contextlib.suppress(ValueError):
116            addr = int(msg.args[0])
117            bean = self._by_address.get(addr)
118            if bean is not None:
119                state = SensorState.ACTIVE if msg.command == "Q" else SensorState.INACTIVE
120                await bean._set_known_state(state)
121
122
123# ----- Power ---------------------------------------------------------------
124
125
126class DccexPowerManager(PowerManager):
127    """``<1>`` on, ``<0>`` off."""
128
129    def __init__(self, tc: DccexTrafficController) -> None:
130        super().__init__()
131        self._tc = tc
132        self._unsub = tc.inbound.connect(self._on_message)
133
134    async def set_power(self, state: PowerState) -> None:
135        cmd = "1" if state == PowerState.ON else "0"
136        await self._tc.send(DccexMessage(command=cmd, args=()))
137        await self._set_state(state)
138
139    async def _on_message(self, msg: DccexMessage) -> None:
140        # <p state> — power state echo. state=1 means on, 0 off.
141        if msg.command != "p" or not msg.args:
142            return
143        with contextlib.suppress(ValueError):
144            wire = int(msg.args[0])
145            await self._set_state(PowerState.ON if wire else PowerState.OFF)
146
147
148# ----- helpers --------------------------------------------------------------
149
150
151def _parse_address(system_name: str, full_prefix: str) -> int:
152    rest = system_name.removeprefix(full_prefix)
153    try:
154        return int(rest)
155    except ValueError as e:
156        raise ValueError(
157            f"DCC-EX system name {system_name!r} must end in a numeric address"
158        ) from e