jmri_nce.beans

  1"""NCE-aware bean classes + managers + power manager."""
  2from __future__ import annotations
  3
  4from jmri_core.beans.power import PowerManager
  5from jmri_core.beans.sensor import Sensor
  6from jmri_core.beans.turnout import Turnout
  7from jmri_core.manager import Manager
  8from jmri_core.state import PowerState, TurnoutState
  9
 10from jmri_nce.messages import accessory_decoder_normal_ops, power_off, power_on
 11from jmri_nce.traffic import NceTrafficController
 12
 13# ----- Turnout -------------------------------------------------------------
 14
 15
 16class NceTurnout(Turnout):
 17    """Sends an accessory-decoder normal-ops packet on commanded-state change.
 18
 19    Known state arrives via :class:`~jmri_nce.monitors.NceTurnoutMonitor`
 20    polling, not in-line, so :meth:`set_commanded_state` does **not** mirror
 21    commanded into known. The monitor sets known state on the next poll
 22    cycle (default 1.5s).
 23    """
 24
 25    __slots__ = ("_address", "_tc")
 26
 27    def __init__(self, system_name: str, address: int, tc: NceTrafficController) -> None:
 28        super().__init__(system_name)
 29        self._address = address
 30        self._tc = tc
 31
 32    @property
 33    def address(self) -> int:
 34        return self._address
 35
 36    async def set_commanded_state(self, state: TurnoutState) -> None:
 37        if state in (TurnoutState.CLOSED, TurnoutState.THROWN):
 38            await self._tc.send_command(
 39                accessory_decoder_normal_ops(self._address, on=(state == TurnoutState.CLOSED))
 40            )
 41        prev = self._commanded_state
 42        if prev == state:
 43            return
 44        self._commanded_state = state
 45        # Mirror commanded → known for immediate UI feedback. JMRI Java
 46        # does the same. The next monitor poll will correct this back if
 47        # the real CS state disagrees (e.g., motor stall, no decoder).
 48        if state in (TurnoutState.CLOSED, TurnoutState.THROWN):
 49            await self._set_known_state(state)
 50
 51
 52class NceTurnoutManager(Manager[NceTurnout]):
 53    """Constructs and indexes NCE turnouts by accessory address."""
 54
 55    type_letter = "T"
 56
 57    def __init__(self, system_prefix: str, *, tc: NceTrafficController) -> None:
 58        super().__init__(system_prefix)
 59        self._tc = tc
 60        self._by_address: dict[int, NceTurnout] = {}
 61
 62    def _create(self, system_name: str) -> NceTurnout:
 63        address = _parse_address(system_name, self.full_prefix)
 64        bean = NceTurnout(system_name, address, self._tc)
 65        self._by_address[address] = bean
 66        return bean
 67
 68    async def deregister(self, bean: NceTurnout) -> None:
 69        self._by_address.pop(bean.address, None)
 70        await super().deregister(bean)
 71
 72    def get_by_address(self, address: int) -> NceTurnout | None:
 73        return self._by_address.get(address)
 74
 75
 76# ----- Sensor --------------------------------------------------------------
 77
 78
 79class NceSensor(Sensor):
 80    """Passive — its state is set by the AIU poller.
 81
 82    Sensor system names embed the AIU address + input number: ``NS<n>``
 83    where *n* = (aiu_addr-1) * 14 + input_bit + 1. The poller does the
 84    decoding; the bean is just storage.
 85    """
 86
 87    __slots__ = ("_address",)
 88
 89    def __init__(self, system_name: str, address: int) -> None:
 90        super().__init__(system_name)
 91        self._address = address
 92
 93    @property
 94    def address(self) -> int:
 95        return self._address
 96
 97
 98class NceSensorManager(Manager[NceSensor]):
 99    type_letter = "S"
100
101    def __init__(self, system_prefix: str) -> None:
102        super().__init__(system_prefix)
103        self._by_address: dict[int, NceSensor] = {}
104
105    def _create(self, system_name: str) -> NceSensor:
106        address = _parse_address(system_name, self.full_prefix)
107        bean = NceSensor(system_name, address)
108        self._by_address[address] = bean
109        return bean
110
111    def get_by_address(self, address: int) -> NceSensor | None:
112        return self._by_address.get(address)
113
114
115# ----- Power ---------------------------------------------------------------
116
117
118class NcePowerManager(PowerManager):
119    """Send OP_POWER_ON / OP_POWER_OFF; state is set on successful reply."""
120
121    def __init__(self, tc: NceTrafficController) -> None:
122        super().__init__()
123        self._tc = tc
124
125    async def set_power(self, state: PowerState) -> None:
126        if state == PowerState.ON:
127            await self._tc.send_command(power_on())
128        elif state == PowerState.OFF:
129            await self._tc.send_command(power_off())
130        # NCE has no IDLE wire command; ignore other targets here.
131        await self._set_state(state)
132
133
134# ----- helpers --------------------------------------------------------------
135
136
137def _parse_address(system_name: str, full_prefix: str) -> int:
138    rest = system_name.removeprefix(full_prefix)
139    try:
140        return int(rest)
141    except ValueError as e:
142        raise ValueError(
143            f"NCE system name {system_name!r} must end in a numeric address"
144        ) from e