jmri_dccex

 8class BytePort(ABC):
 9    @abstractmethod
10    async def write(self, data: bytes) -> None: ...
11    @abstractmethod
12    async def read_byte(self) -> int: ...
13    @abstractmethod
14    async def aclose(self) -> None: ...
15    @property
16    @abstractmethod
17    def is_open(self) -> bool: ...

 19class DccexConnection(SystemConnection):
 20    name: ClassVar[str] = "DCC-EX"
 21    prefix: ClassVar[str] = "D"
 22
 23    def __init__(self) -> None:
 24        super().__init__()
 25        self._port: BytePort | None = None
 26        self._tc: DccexTrafficController | None = None
 27        self._programmer: DccexProgrammer | None = None
 28        self._cs_monitor_task: asyncio.Task[None] | None = None
 29        self._cs_ping_task: asyncio.Task[None] | None = None
 30
 31    async def open(self, settings: ConnectionSettings | None = None) -> None:
 32        s = settings or {}
 33        self._port = await _build_port(s)
 34        self._tc = DccexTrafficController(self._port)
 35        await self._tc.start()
 36
 37        self.turnouts = DccexTurnoutManager(self.prefix, tc=self._tc)        # type: ignore[assignment]
 38        self.sensors = DccexSensorManager(self.prefix, tc=self._tc)          # type: ignore[assignment]
 39        self.throttles = DccexThrottleManager(self.prefix, tc=self._tc)
 40        # Rebind the consist manager onto the DCC-EX-aware throttle
 41        # manager. Without this the fan-out drives placeholder beans
 42        # and no ``<t>`` (throttle) frames reach the CS.
 43        self.consists.rebind_throttles(self.throttles)
 44        self.power = DccexPowerManager(self._tc)
 45        self._programmer = DccexProgrammer(self._tc)
 46        self.programmers = DccexProgrammerManager(self._programmer)
 47        # Wire advanced-consist CV19 writes onto the DCC-EX programmer.
 48        self.consists.rebind_programmers(self.programmers)
 49
 50        # CS reachability: DCC-EX doesn't chatter unsolicited, so we
 51        # ping ``<s>`` (status query) every 2 seconds and watch for
 52        # ANY inbound message within 3 seconds. Status ping response
 53        # is ``<p 0>``/``<p 1>`` plus other startup chatter — any of
 54        # which resets the monitor's silence timer. Start pessimistic
 55        # (base class default is optimistic True) so a home-page
 56        # render during the monitor's startup grace shows "Off" until
 57        # we've actually seen a reply, not a false "On".
 58        self._cs_reachable = False
 59        self._cs_ping_task = asyncio.create_task(
 60            self._cs_ping_loop(), name="dccex-cs-ping",
 61        )
 62        self._cs_monitor_task = asyncio.create_task(
 63            self._cs_reachability_loop(), name="dccex-cs-reachability",
 64        )
 65
 66        await super().open(settings)
 67
 68    async def _cs_ping_loop(self) -> None:
 69        """Send ``<s>`` every 2s so the CS keeps replying with status
 70        traffic — that's what the reachability monitor watches for."""
 71        import contextlib  # noqa: PLC0415
 72        try:
 73            while True:
 74                await asyncio.sleep(2.0)
 75                if self._tc is None:
 76                    return
 77                try:
 78                    await self._tc.send(DccexMessage(command="s", args=()))
 79                except Exception:  # noqa: BLE001
 80                    # Send may raise on a closed port; the monitor
 81                    # loop notices via the silence timer.
 82                    pass
 83        except asyncio.CancelledError:
 84            with contextlib.suppress(Exception):
 85                return
 86
 87    async def _cs_reachability_loop(self) -> None:
 88        """3-second silence rule against ``last_inbound_at``."""
 89        import contextlib  # noqa: PLC0415
 90        try:
 91            await asyncio.sleep(3.0)   # startup grace
 92            while True:
 93                await asyncio.sleep(1.0)
 94                if self._tc is None:
 95                    return
 96                last = self._tc.last_inbound_at
 97                if last is None:
 98                    await self._set_cs_reachable(False)
 99                    continue
100                age = time.monotonic() - last
101                await self._set_cs_reachable(age <= 3.0)
102        except asyncio.CancelledError:
103            with contextlib.suppress(Exception):
104                return
105
106    async def close(self) -> None:
107        for task in (self._cs_ping_task, self._cs_monitor_task):
108            if task is not None:
109                task.cancel()
110                import contextlib  # noqa: PLC0415
111                with contextlib.suppress(asyncio.CancelledError, Exception):
112                    await task
113        self._cs_ping_task = None
114        self._cs_monitor_task = None
115        if self._tc is not None:
116            await self._tc.aclose()
117            self._tc = None
118        self._port = None
119        await super().close()
120
121    @property
122    def traffic(self) -> DccexTrafficController:
123        if self._tc is None:
124            raise RuntimeError("connection not open")
125        return self._tc

31@dataclass(frozen=True, slots=True)
32class DccexMessage:
33    """A parsed message — command character + space-separated args."""
34
35    command: str          # first token (e.g. "1", "T", "Q")
36    args: tuple[str, ...] # remaining tokens
37
38    @property
39    def raw(self) -> str:
40        if not self.args:
41            return self.command
42        return f"{self.command} {' '.join(self.args)}"

204class DccexOpsProgrammer(Programmer):
205    """Ops-mode adapter scoped to a specific DCC address."""
206
207    def __init__(self, inner: DccexProgrammer, address: int, *, long_address: bool) -> None:
208        self._inner = inner
209        self._address = address
210        self._long = long_address
211
212    @property
213    def mode(self) -> ProgrammingMode:
214        return ProgrammingMode.OPS_MODE
215
216    @property
217    def address(self) -> int:
218        return self._address
219
220    async def read_cv(self, cv: int) -> int:
221        """RailCom-backed ops-mode CV read for this address.
222
223        Raises :class:`RuntimeError` when the CS times out (usually
224        because RailCom isn't enabled in the firmware config) or
225        returns ``-1`` (decoder didn't answer via RailCom). Callers
226        that want a "try RailCom, fall back to write-and-read-back"
227        strategy should catch that exception at their layer.
228        """
229        return await self._inner.read_cv_ops(self._address, cv)
230
231    async def write_cv(self, cv: int, value: int) -> None:
232        await self._inner.write_cv_ops(self._address, cv, value)

127class DccexPowerManager(PowerManager):
128    """``<1>`` on, ``<0>`` off."""
129
130    def __init__(self, tc: DccexTrafficController) -> None:
131        super().__init__()
132        self._tc = tc
133        self._unsub = tc.inbound.connect(self._on_message)
134
135    async def set_power(self, state: PowerState) -> None:
136        cmd = "1" if state == PowerState.ON else "0"
137        await self._tc.send(DccexMessage(command=cmd, args=()))
138        await self._set_state(state)
139
140    async def _on_message(self, msg: DccexMessage) -> None:
141        # <p state> — power state echo. state=1 means on, 0 off.
142        if msg.command != "p" or not msg.args:
143            return
144        with contextlib.suppress(ValueError):
145            wire = int(msg.args[0])
146            await self._set_state(PowerState.ON if wire else PowerState.OFF)

 42class DccexProgrammer:
 43    """Direct programmer driver — used by both service and ops adapters."""
 44
 45    def __init__(self, tc: DccexTrafficController) -> None:
 46        self._tc = tc
 47        self._busy = asyncio.Lock()
 48
 49    async def write_cv_service(
 50        self, cv: int, value: int, *, timeout: float = 6.0
 51    ) -> None:
 52        async with self._busy:
 53            await self._send_and_wait_r(
 54                DccexMessage(command="W", args=(str(cv), str(value))),
 55                expected_cv=cv,
 56                timeout=timeout,
 57            )
 58
 59    async def read_cv_service(self, cv: int, *, timeout: float = 6.0) -> int:
 60        async with self._busy:
 61            return await self._send_and_wait_r(
 62                DccexMessage(command="R", args=(str(cv),)),
 63                expected_cv=cv,
 64                timeout=timeout,
 65            )
 66
 67    async def write_cv_ops(
 68        self, address: int, cv: int, value: int, *, timeout: float = 2.0
 69    ) -> None:
 70        """Fire-and-forget — ops-mode writes don't ack in DCC-EX's text protocol."""
 71        async with self._busy:
 72            await self._tc.send(DccexMessage(
 73                command="w", args=(str(address), str(cv), str(value)),
 74            ))
 75
 76    async def read_cv_ops(
 77        self, address: int, cv: int, *, timeout: float = 6.0,
 78    ) -> int:
 79        """RailCom-backed ops-mode CV read on the main track.
 80
 81        Wire: ``<r cab cv>`` request → ``<r cab cv value>`` reply. The
 82        CS's ``r`` handler explicitly requires ``DCCWaveform::isRailcom()``
 83        to return true; if RailCom is disabled in the firmware config
 84        the request is silently dropped, so a caller sees this as a
 85        timeout rather than a distinctive error. The timeout message
 86        surfaces that possibility explicitly.
 87
 88        Failure (``value == -1``) surfaces as :class:`RuntimeError`
 89        matching the same-shaped failure on the service-mode path so
 90        callers can use one exception filter for both.
 91        """
 92        async with self._busy:
 93            return await self._send_and_wait_ops_read(
 94                address=address, cv=cv, timeout=timeout,
 95            )
 96
 97    async def _send_and_wait_r(
 98        self, request: DccexMessage, *, expected_cv: int, timeout: float
 99    ) -> int:
100        """Await a 2-arg service-mode reply matching ``expected_cv``.
101
102        Accepts either ``<r cv value>`` (write ack + some firmwares'
103        read replies) or ``<v cv value>`` (canonical read reply per
104        DCC-EX upstream). Callers pass the CV they're expecting; a
105        reply for a different CV is ignored so an unrelated CS
106        message doesn't complete our future prematurely.
107        """
108        future: asyncio.Future[int] = asyncio.get_event_loop().create_future()
109
110        async def listener(msg: DccexMessage) -> None:
111            if future.done():
112                return
113            # Accept both letters — see the module docstring.
114            if msg.command not in ("r", "v") or len(msg.args) != 2:
115                return
116            try:
117                cv = int(msg.args[0])
118                value = int(msg.args[1])
119            except ValueError:
120                return
121            if cv != expected_cv:
122                return
123            if value < 0:
124                future.set_exception(
125                    RuntimeError(f"DCC-EX programming op failed on CV {cv}")
126                )
127            else:
128                future.set_result(value)
129
130        disconnect = self._tc.inbound.connect(listener)
131        try:
132            await self._tc.send(request)
133            return await asyncio.wait_for(future, timeout=timeout)
134        finally:
135            disconnect()
136
137    async def _send_and_wait_ops_read(
138        self, *, address: int, cv: int, timeout: float,
139    ) -> int:
140        """Send ``<r cab cv>`` and await ``<r cab cv value>`` (3 args).
141
142        The 3-arg shape lets the listener distinguish ops-mode
143        replies from service-mode replies unambiguously — a 2-arg
144        ``<r ...>`` is a write ack, a 3-arg ``<r ...>`` is a RailCom
145        read result.
146        """
147        future: asyncio.Future[int] = asyncio.get_event_loop().create_future()
148
149        async def listener(msg: DccexMessage) -> None:
150            if future.done():
151                return
152            if msg.command != "r" or len(msg.args) != 3:
153                return
154            try:
155                cab = int(msg.args[0])
156                got_cv = int(msg.args[1])
157                value = int(msg.args[2])
158            except ValueError:
159                return
160            if cab != address or got_cv != cv:
161                return
162            if value < 0:
163                future.set_exception(RuntimeError(
164                    f"DCC-EX ops-mode read failed on cab={address} CV={cv} "
165                    "(no RailCom response)"
166                ))
167            else:
168                future.set_result(value)
169
170        disconnect = self._tc.inbound.connect(listener)
171        try:
172            await self._tc.send(DccexMessage(
173                command="r", args=(str(address), str(cv)),
174            ))
175            return await asyncio.wait_for(future, timeout=timeout)
176        except asyncio.TimeoutError as e:
177            raise RuntimeError(
178                f"DCC-EX ops-mode read timed out on cab={address} CV={cv}. "
179                "The CS silently drops <r ...> if RailCom is disabled — "
180                "verify the firmware was built with RAILCOM enabled and "
181                "that the decoder supports RailCom cutout."
182            ) from e
183        finally:
184            disconnect()

235class DccexProgrammerManager(ProgrammerManager):
236    def __init__(self, inner: DccexProgrammer) -> None:
237        self._inner = inner
238        self._service_singleton: DccexServiceProgrammer | None = None
239
240    def global_programmer(self) -> DccexServiceProgrammer:
241        if self._service_singleton is None:
242            self._service_singleton = DccexServiceProgrammer(self._inner)
243        return self._service_singleton
244
245    def ops_programmer(self, address: int, *, long_address: bool = False) -> DccexOpsProgrammer:
246        return DccexOpsProgrammer(self._inner, address, long_address=long_address)
247
248    @property
249    def inner(self) -> DccexProgrammer:
250        return self._inner

83class DccexSensor(Sensor):
84    __slots__ = ("_address",)
85
86    def __init__(self, system_name: str, address: int) -> None:
87        super().__init__(system_name)
88        self._address = address
89
90    @property
91    def address(self) -> int:
92        return self._address

 95class DccexSensorManager(Manager[DccexSensor]):
 96    type_letter = "S"
 97
 98    def __init__(self, system_prefix: str, *, tc: DccexTrafficController) -> None:
 99        super().__init__(system_prefix)
100        self._by_address: dict[int, DccexSensor] = {}
101        self._unsub = tc.inbound.connect(self._on_message)
102
103    def _create(self, system_name: str) -> DccexSensor:
104        address = _parse_address(system_name, self.full_prefix)
105        bean = DccexSensor(system_name, address)
106        self._by_address[address] = bean
107        return bean
108
109    def get_by_address(self, address: int) -> DccexSensor | None:
110        return self._by_address.get(address)
111
112    async def _on_message(self, msg: DccexMessage) -> None:
113        # <Q id> active, <q id> inactive.
114        if msg.command not in ("Q", "q") or not msg.args:
115            return
116        with contextlib.suppress(ValueError):
117            addr = int(msg.args[0])
118            bean = self._by_address.get(addr)
119            if bean is not None:
120                state = SensorState.ACTIVE if msg.command == "Q" else SensorState.INACTIVE
121                await bean._set_known_state(state)

187class DccexServiceProgrammer(Programmer):
188    """Service-mode adapter satisfying the core ABC."""
189
190    def __init__(self, inner: DccexProgrammer) -> None:
191        self._inner = inner
192
193    @property
194    def mode(self) -> ProgrammingMode:
195        return ProgrammingMode.SERVICE_DIRECT
196
197    async def read_cv(self, cv: int) -> int:
198        return await self._inner.read_cv_service(cv)
199
200    async def write_cv(self, cv: int, value: int) -> None:
201        await self._inner.write_cv_service(cv, value)

40class DccexThrottle(Throttle):
41    """DCC-EX-aware throttle. Sends ``<t cab speed dir>`` on speed/direction."""
42
43    def __init__(
44        self, address: int, tc: DccexTrafficController, *, long_address: bool = False
45    ) -> None:
46        super().__init__(address=address, long_address=long_address)
47        self._tc = tc
48
49    async def set_speed(self, speed: float) -> None:
50        await super().set_speed(speed)
51        await self._push_speed()
52
53    async def set_direction(self, direction: ThrottleDirection) -> None:
54        await super().set_direction(direction)
55        await self._push_speed()
56
57    async def set_function(self, index: int, on: bool) -> None:
58        await super().set_function(index, on)
59        # DCC-EX <F cab function 1|0>.
60        await self._tc.send(DccexMessage(
61            command="F", args=(str(self.address), str(index), "1" if on else "0"),
62        ))
63
64    async def _push_speed(self) -> None:
65        direction = "1" if self.direction == ThrottleDirection.FORWARD else "0"
66        wire = speed_to_wire_128(self.speed)
67        await self._tc.send(DccexMessage(
68            command="t", args=(str(self.address), str(wire), direction),
69        ))

72class DccexThrottleManager(ThrottleManager):
73    """DCC-EX throttle manager — uses the core ThrottleManager hook."""
74
75    def __init__(self, system_prefix: str, *, tc: DccexTrafficController) -> None:
76        super().__init__(system_prefix)
77        self._tc = tc
78
79    async def _construct(self, address: int, *, long_address: bool) -> DccexThrottle:
80        return DccexThrottle(address=address, tc=self._tc, long_address=long_address)

24class DccexTrafficController:
25    def __init__(self, port: BytePort) -> None:
26        self._port = port
27        self._send_lock = asyncio.Lock()
28        self._read_task: asyncio.Task[None] | None = None
29        self._closed = False
30        self.inbound: Signal[DccexMessage] = Signal()
31        # Timestamp of the most recent inbound message. Read by
32        # :class:`DccexConnection`'s reachability monitor. DCC-EX
33        # doesn't chatter without solicitation, so the connection also
34        # runs a periodic ``<s>`` (status) ping to guarantee traffic.
35        self.last_inbound_at: float | None = None
36
37    async def start(self) -> None:
38        if self._read_task is not None:
39            return
40        self._read_task = asyncio.create_task(self._read_loop(), name="dccex-read")
41
42    async def aclose(self) -> None:
43        self._closed = True
44        if self._read_task is not None:
45            self._read_task.cancel()
46            import contextlib  # noqa: PLC0415
47            with contextlib.suppress(asyncio.CancelledError, Exception):
48                await self._read_task
49            self._read_task = None
50        await self._port.aclose()
51
52    async def send(self, msg: DccexMessage) -> None:
53        async with self._send_lock:
54            await self._port.write(encode(msg))
55
56    async def _read_loop(self) -> None:
57        try:
58            while not self._closed:
59                try:
60                    text = await read_message(self._port.read_byte)
61                except asyncio.CancelledError:
62                    return
63                except Exception as e:  # noqa: BLE001
64                    _log.warning("DCC-EX read error: %s", e)
65                    continue
66                try:
67                    msg = decode(text)
68                except ValueError as e:
69                    _log.warning("DCC-EX bad message %r: %s", text, e)
70                    continue
71                self.last_inbound_at = time.monotonic()
72                await self.inbound.emit(msg)
73        except asyncio.CancelledError:
74            return