jmri_ncemonitor.beans

  1"""Read-only NamedBean views for the passive Cab Bus monitor.
  2
  3The monitor doesn't *control* the layout — it only *observes* it. The
  4beans this module exposes follow that constraint:
  5
  6* :class:`PassiveSensor` — same shape as a normal Sensor but only the
  7  AIU poller (via :class:`~jmri_ncemonitor.monitor.CabBusMonitor`)
  8  updates its state.
  9* :class:`PassiveTurnout` — set_commanded_state raises; the monitor sets
 10  *known* state from accessory traffic it observes on the bus.
 11* :class:`PassivePowerManager` — set_power raises; power state is
 12  observed-only.
 13
 14All three demonstrate that a fully-passive driver fits inside the
 15existing :class:`SystemConnection` ABC without requiring API changes —
 16clients are free to query state but commands fail loudly.
 17"""
 18from __future__ import annotations
 19
 20from jmri_core.beans.power import PowerManager
 21from jmri_core.beans.sensor import Sensor
 22from jmri_core.beans.turnout import Turnout
 23from jmri_core.manager import Manager
 24from jmri_core.state import PowerState, SensorState, TurnoutState
 25
 26from jmri_ncemonitor.messages import (
 27    AccessoryObserved,
 28    AiuUpdate,
 29    CabBusMessage,
 30)
 31from jmri_ncemonitor.monitor import CabBusMonitor
 32
 33# --- helper ----------------------------------------------------------------
 34
 35
 36PASSIVE_REASON = "NCE Cab Bus monitor is passive — commands are not supported"
 37
 38
 39# --- Sensor ----------------------------------------------------------------
 40
 41
 42class PassiveSensor(Sensor):
 43    """Read-only sensor — its known state is observed off the bus."""
 44
 45    __slots__ = ("_address",)
 46
 47    def __init__(self, system_name: str, address: int) -> None:
 48        super().__init__(system_name)
 49        self._address = address
 50
 51    @property
 52    def address(self) -> int:
 53        return self._address
 54
 55
 56class PassiveSensorManager(Manager[PassiveSensor]):
 57    """Sensor manager driven by observed :class:`AiuUpdate` events."""
 58
 59    type_letter = "S"
 60
 61    def __init__(self, system_prefix: str, monitor: CabBusMonitor) -> None:
 62        super().__init__(system_prefix)
 63        self._by_address: dict[int, PassiveSensor] = {}
 64        self._unsub = monitor.inbound.connect(self._on_message)
 65
 66    def _create(self, system_name: str) -> PassiveSensor:
 67        address = _parse_address(system_name, self.full_prefix)
 68        bean = PassiveSensor(system_name, address)
 69        self._by_address[address] = bean
 70        return bean
 71
 72    def get_by_address(self, address: int) -> PassiveSensor | None:
 73        return self._by_address.get(address)
 74
 75    async def _on_message(self, msg: CabBusMessage) -> None:
 76        if not isinstance(msg, AiuUpdate):
 77            return
 78        for bit in range(14):
 79            sensor_addr = (msg.aiu_module - 1) * 14 + bit + 1
 80            sensor = self._by_address.get(sensor_addr)
 81            if sensor is None:
 82                continue
 83            state = (
 84                SensorState.ACTIVE
 85                if (msg.input_word & (1 << bit))
 86                else SensorState.INACTIVE
 87            )
 88            await sensor._set_known_state(state)
 89
 90
 91# --- Turnout ---------------------------------------------------------------
 92
 93
 94class PassiveTurnout(Turnout):
 95    """Read-only turnout — commanded state cannot be set from this side."""
 96
 97    __slots__ = ("_address",)
 98
 99    def __init__(self, system_name: str, address: int) -> None:
100        super().__init__(system_name)
101        self._address = address
102
103    @property
104    def address(self) -> int:
105        return self._address
106
107    async def set_commanded_state(self, state: TurnoutState) -> None:
108        raise RuntimeError(PASSIVE_REASON)
109
110
111class PassiveTurnoutManager(Manager[PassiveTurnout]):
112    """Turnout manager driven by observed :class:`AccessoryObserved` events."""
113
114    type_letter = "T"
115
116    def __init__(self, system_prefix: str, monitor: CabBusMonitor) -> None:
117        super().__init__(system_prefix)
118        self._by_address: dict[int, PassiveTurnout] = {}
119        self._unsub = monitor.inbound.connect(self._on_message)
120
121    def _create(self, system_name: str) -> PassiveTurnout:
122        address = _parse_address(system_name, self.full_prefix)
123        bean = PassiveTurnout(system_name, address)
124        self._by_address[address] = bean
125        return bean
126
127    def get_by_address(self, address: int) -> PassiveTurnout | None:
128        return self._by_address.get(address)
129
130    async def _on_message(self, msg: CabBusMessage) -> None:
131        if not isinstance(msg, AccessoryObserved):
132            return
133        bean = self._by_address.get(msg.accessory_address)
134        if bean is None:
135            return
136        state = TurnoutState.CLOSED if msg.closed else TurnoutState.THROWN
137        await bean._set_known_state(state)
138
139
140# --- Power -----------------------------------------------------------------
141
142
143class PassivePowerManager(PowerManager):
144    """Power state is observed-only; ``set_power`` raises."""
145
146    async def set_power(self, state: PowerState) -> None:
147        raise RuntimeError(PASSIVE_REASON)
148
149
150# --- helpers ---------------------------------------------------------------
151
152
153def _parse_address(system_name: str, full_prefix: str) -> int:
154    rest = system_name.removeprefix(full_prefix)
155    try:
156        return int(rest)
157    except ValueError as e:
158        raise ValueError(
159            f"NCE Cab Bus monitor system name {system_name!r} "
160            f"must end in a numeric address"
161        ) from e