jmri_openlcb.beans

  1"""Event-bound OpenLCB beans.
  2
  3OpenLCB doesn't use addresses — it uses 64-bit event IDs. Each bean
  4binds two event IDs (one per state) at construction:
  5
  6* :class:`OlcbTurnout` — ``closed_event`` and ``thrown_event``.
  7* :class:`OlcbSensor` — ``active_event`` and ``inactive_event``.
  8
  9When a turnout's commanded state changes, the bean *produces* the
 10matching event on the bus. When an inbound event matches a bean's
 11configured event ID, that bean's *known* state moves to the corresponding
 12value.
 13
 14Because the event-ID mapping has to be provided at bean construction
 15time, the standard ``Manager.provide(system_name)`` shape doesn't fit:
 16there's no way to derive the events from a string alone. We expose a
 17``configure(system_name, closed_event, thrown_event)`` method on the
 18manager for that.
 19"""
 20from __future__ import annotations
 21
 22from jmri_core.beans.sensor import Sensor
 23from jmri_core.beans.turnout import Turnout
 24from jmri_core.manager import Manager
 25from jmri_core.state import SensorState, TurnoutState
 26
 27from jmri_openlcb.events import EventId, EventReport
 28from jmri_openlcb.traffic import OlcbTrafficController
 29
 30# ----- Turnout -------------------------------------------------------------
 31
 32
 33class OlcbTurnout(Turnout):
 34    """A turnout bound to two event IDs (closed/thrown)."""
 35
 36    __slots__ = ("_closed_event", "_tc", "_thrown_event")
 37
 38    def __init__(
 39        self,
 40        system_name: str,
 41        closed_event: EventId,
 42        thrown_event: EventId,
 43        tc: OlcbTrafficController,
 44    ) -> None:
 45        super().__init__(system_name)
 46        self._closed_event = closed_event
 47        self._thrown_event = thrown_event
 48        self._tc = tc
 49
 50    @property
 51    def closed_event(self) -> EventId:
 52        return self._closed_event
 53
 54    @property
 55    def thrown_event(self) -> EventId:
 56        return self._thrown_event
 57
 58    async def set_commanded_state(self, state: TurnoutState) -> None:
 59        if state == TurnoutState.CLOSED:
 60            await self._tc.produce_event(self._closed_event)
 61        elif state == TurnoutState.THROWN:
 62            await self._tc.produce_event(self._thrown_event)
 63        await super().set_commanded_state(state)
 64
 65
 66class OlcbTurnoutManager(Manager[OlcbTurnout]):
 67    """Manager that creates turnouts only via :meth:`configure`."""
 68
 69    type_letter = "T"
 70
 71    def __init__(self, system_prefix: str, *, tc: OlcbTrafficController) -> None:
 72        super().__init__(system_prefix)
 73        self._tc = tc
 74        self._by_event: dict[EventId, tuple[OlcbTurnout, TurnoutState]] = {}
 75        self._unsub = tc.events.connect(self._on_event)
 76
 77    def _create(self, system_name: str) -> OlcbTurnout:
 78        raise NotImplementedError(
 79            "OpenLCB turnouts require explicit event-ID configuration; "
 80            "use OlcbTurnoutManager.configure(...) instead of provide()"
 81        )
 82
 83    async def configure(
 84        self,
 85        system_name: str,
 86        closed_event: EventId,
 87        thrown_event: EventId,
 88    ) -> OlcbTurnout:
 89        """Create and register a turnout with the given event bindings."""
 90        bean = OlcbTurnout(system_name, closed_event, thrown_event, self._tc)
 91        await self.register(bean)
 92        self._by_event[closed_event] = (bean, TurnoutState.CLOSED)
 93        self._by_event[thrown_event] = (bean, TurnoutState.THROWN)
 94        return bean
 95
 96    async def _on_event(self, report: EventReport) -> None:
 97        binding = self._by_event.get(report.event)
 98        if binding is None:
 99            return
100        bean, target_state = binding
101        await bean._set_known_state(target_state)
102
103
104# ----- Sensor --------------------------------------------------------------
105
106
107class OlcbSensor(Sensor):
108    """A sensor bound to two event IDs (active/inactive)."""
109
110    __slots__ = ("_active_event", "_inactive_event")
111
112    def __init__(
113        self,
114        system_name: str,
115        active_event: EventId,
116        inactive_event: EventId,
117    ) -> None:
118        super().__init__(system_name)
119        self._active_event = active_event
120        self._inactive_event = inactive_event
121
122    @property
123    def active_event(self) -> EventId:
124        return self._active_event
125
126    @property
127    def inactive_event(self) -> EventId:
128        return self._inactive_event
129
130
131class OlcbSensorManager(Manager[OlcbSensor]):
132    type_letter = "S"
133
134    def __init__(self, system_prefix: str, *, tc: OlcbTrafficController) -> None:
135        super().__init__(system_prefix)
136        self._by_event: dict[EventId, tuple[OlcbSensor, SensorState]] = {}
137        self._unsub = tc.events.connect(self._on_event)
138
139    def _create(self, system_name: str) -> OlcbSensor:
140        raise NotImplementedError(
141            "OpenLCB sensors require explicit event-ID configuration; "
142            "use OlcbSensorManager.configure(...)"
143        )
144
145    async def configure(
146        self,
147        system_name: str,
148        active_event: EventId,
149        inactive_event: EventId,
150    ) -> OlcbSensor:
151        bean = OlcbSensor(system_name, active_event, inactive_event)
152        await self.register(bean)
153        self._by_event[active_event] = (bean, SensorState.ACTIVE)
154        self._by_event[inactive_event] = (bean, SensorState.INACTIVE)
155        return bean
156
157    async def _on_event(self, report: EventReport) -> None:
158        binding = self._by_event.get(report.event)
159        if binding is None:
160            return
161        bean, target_state = binding
162        await bean._set_known_state(target_state)