jmri_dccex.connection

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