jmri_openlcb.connection

 1"""``OlcbConnection`` — SystemConnection subclass for OpenLCB/LCC.
 2
 3Settings:
 4    {
 5        "kind": "loopback" | "serial" | "tcp",
 6        "source_alias": 0x123,    # 12-bit node alias for outbound PCERs
 7        # ...transport-specific...
 8    }
 9"""
10from __future__ import annotations
11
12from typing import Any, ClassVar
13
14from jmri_core.hal.connection import ConnectionSettings, SystemConnection
15
16from jmri_openlcb.beans import OlcbSensorManager, OlcbTurnoutManager
17from jmri_openlcb.power import OlcbPowerManager
18from jmri_openlcb.traffic import OlcbTrafficController
19from jmri_openlcb.transport import BytePort, LoopbackPort, SerialPort, TcpPort
20
21
22class OlcbConnection(SystemConnection):
23    name: ClassVar[str] = "OpenLCB"
24    prefix: ClassVar[str] = "X"
25
26    def __init__(self) -> None:
27        super().__init__()
28        self._port: BytePort | None = None
29        self._tc: OlcbTrafficController | 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 = OlcbTrafficController(
35            self._port, source_alias=int(s.get("source_alias", 0x123))
36        )
37        await self._tc.start()
38
39        self.turnouts = OlcbTurnoutManager(self.prefix, tc=self._tc)  # type: ignore[assignment]
40        self.sensors = OlcbSensorManager(self.prefix, tc=self._tc)    # type: ignore[assignment]
41        # Well-known event power manager — produces + consumes the
42        # reserved 01.00.00.00.00.00.FF.FE / .FF event IDs so a
43        # single set_power() call converges every OpenLCB node
44        # listening to the bus. See :mod:`jmri_openlcb.power`.
45        self.power = OlcbPowerManager(self._tc)
46
47        await super().open(settings)
48
49    async def close(self) -> None:
50        # Drop the power manager's listener before we tear down the
51        # traffic controller so the manager doesn't fire against a
52        # closed signal.
53        if isinstance(self.power, OlcbPowerManager):
54            self.power.close()
55        if self._tc is not None:
56            await self._tc.aclose()
57            self._tc = None
58        self._port = None
59        await super().close()
60
61    @property
62    def traffic(self) -> OlcbTrafficController:
63        if self._tc is None:
64            raise RuntimeError("connection not open")
65        return self._tc
66
67
68async def _build_port(settings: dict[str, Any]) -> BytePort:
69    kind = settings.get("kind", "loopback")
70    if kind == "serial":
71        port = settings.get("port")
72        if not port:
73            raise ValueError("serial transport requires 'port'")
74        baud = int(settings.get("baud", 333333))
75        return await SerialPort.open(port=port, baud=baud)
76    if kind == "tcp":
77        host = settings.get("host")
78        tcp_port = settings.get("tcp_port")
79        if not host or tcp_port is None:
80            raise ValueError("tcp transport requires 'host' and 'tcp_port'")
81        return await TcpPort.open(host=host, port=int(tcp_port))
82    if kind == "loopback":
83        lb = settings.get("loopback_port")
84        if not isinstance(lb, LoopbackPort):
85            raise ValueError("loopback transport requires 'loopback_port'")
86        return lb
87    raise ValueError(f"unknown OpenLCB transport kind: {kind!r}")