jmri_ncemonitor.connection
NceMonitorConnection — the passive SystemConnection subclass.
This is the abstraction stress test: a fully read-only driver dropping
into the same SystemConnection ABC as LocoNet and NCE. The contract
holds because the ABC only requires lifecycle (open/close) and
shaped manager attributes; whether the managers' beans are commanded or
purely observed is a per-bean concern.
Connection settings:
{
"kind": "loopback" | "serial",
# for serial:
"port": "COM7",
"baud": 9600,
# for loopback (tests):
"loopback_port": LoopbackPort,
}
1"""``NceMonitorConnection`` — the passive ``SystemConnection`` subclass. 2 3This is the abstraction stress test: a fully read-only driver dropping 4into the same ``SystemConnection`` ABC as LocoNet and NCE. The contract 5holds because the ABC only requires lifecycle (``open``/``close``) and 6shaped manager attributes; whether the managers' beans are commanded or 7purely observed is a per-bean concern. 8 9Connection settings: 10 11 { 12 "kind": "loopback" | "serial", 13 # for serial: 14 "port": "COM7", 15 "baud": 9600, 16 # for loopback (tests): 17 "loopback_port": LoopbackPort, 18 } 19""" 20from __future__ import annotations 21 22from typing import Any, ClassVar 23 24from jmri_core.hal.connection import ConnectionSettings, SystemConnection 25 26from jmri_ncemonitor.beans import ( 27 PassivePowerManager, 28 PassiveSensorManager, 29 PassiveTurnoutManager, 30) 31from jmri_ncemonitor.monitor import CabBusMonitor 32from jmri_ncemonitor.transport import BytePort, LoopbackPort, SerialPort 33 34 35class NceMonitorConnection(SystemConnection): 36 """Passive NCE Cab Bus monitor. Prefix ``M``, name ``NCE Cab Bus monitor``.""" 37 38 name: ClassVar[str] = "NCE Cab Bus monitor" 39 prefix: ClassVar[str] = "M" 40 41 def __init__(self) -> None: 42 super().__init__() 43 self._port: BytePort | None = None 44 self._monitor: CabBusMonitor | None = None 45 46 async def open(self, settings: ConnectionSettings | None = None) -> None: 47 s = settings or {} 48 self._port = await _build_port(s) 49 self._monitor = CabBusMonitor(self._port) 50 await self._monitor.start() 51 52 self.turnouts = PassiveTurnoutManager(self.prefix, self._monitor) # type: ignore[assignment] 53 self.sensors = PassiveSensorManager(self.prefix, self._monitor) # type: ignore[assignment] 54 self.power = PassivePowerManager() 55 56 await super().open(settings) 57 58 async def close(self) -> None: 59 if self._monitor is not None: 60 await self._monitor.aclose() 61 self._monitor = None 62 self._port = None 63 await super().close() 64 65 @property 66 def monitor(self) -> CabBusMonitor: 67 if self._monitor is None: 68 raise RuntimeError("connection not open") 69 return self._monitor 70 71 72async def _build_port(settings: dict[str, Any]) -> BytePort: 73 kind = settings.get("kind", "loopback") 74 if kind == "serial": 75 port = settings.get("port") 76 if not port: 77 raise ValueError("serial transport requires 'port'") 78 baud = int(settings.get("baud", 9600)) 79 return await SerialPort.open(port=port, baud=baud) 80 if kind == "loopback": 81 loop_port = settings.get("loopback_port") 82 if not isinstance(loop_port, LoopbackPort): 83 raise ValueError( 84 "loopback transport requires 'loopback_port' to be a LoopbackPort" 85 ) 86 return loop_port 87 raise ValueError(f"unknown NCEMonitor transport kind: {kind!r}")
36class NceMonitorConnection(SystemConnection): 37 """Passive NCE Cab Bus monitor. Prefix ``M``, name ``NCE Cab Bus monitor``.""" 38 39 name: ClassVar[str] = "NCE Cab Bus monitor" 40 prefix: ClassVar[str] = "M" 41 42 def __init__(self) -> None: 43 super().__init__() 44 self._port: BytePort | None = None 45 self._monitor: CabBusMonitor | None = None 46 47 async def open(self, settings: ConnectionSettings | None = None) -> None: 48 s = settings or {} 49 self._port = await _build_port(s) 50 self._monitor = CabBusMonitor(self._port) 51 await self._monitor.start() 52 53 self.turnouts = PassiveTurnoutManager(self.prefix, self._monitor) # type: ignore[assignment] 54 self.sensors = PassiveSensorManager(self.prefix, self._monitor) # type: ignore[assignment] 55 self.power = PassivePowerManager() 56 57 await super().open(settings) 58 59 async def close(self) -> None: 60 if self._monitor is not None: 61 await self._monitor.aclose() 62 self._monitor = None 63 self._port = None 64 await super().close() 65 66 @property 67 def monitor(self) -> CabBusMonitor: 68 if self._monitor is None: 69 raise RuntimeError("connection not open") 70 return self._monitor
Passive NCE Cab Bus monitor. Prefix M, name NCE Cab Bus monitor.
async def
open(self, settings: dict[str, typing.Any] | None = None) -> None:
47 async def open(self, settings: ConnectionSettings | None = None) -> None: 48 s = settings or {} 49 self._port = await _build_port(s) 50 self._monitor = CabBusMonitor(self._port) 51 await self._monitor.start() 52 53 self.turnouts = PassiveTurnoutManager(self.prefix, self._monitor) # type: ignore[assignment] 54 self.sensors = PassiveSensorManager(self.prefix, self._monitor) # type: ignore[assignment] 55 self.power = PassivePowerManager() 56 57 await super().open(settings)
Connect to hardware and start any IO loops.
Concrete drivers acquire serial/TCP ports here, install protocol
handlers, and start any required background tasks. The base
implementation should be called via await super().open(settings)
to mark the connection opened.
async def
close(self) -> None:
59 async def close(self) -> None: 60 if self._monitor is not None: 61 await self._monitor.aclose() 62 self._monitor = None 63 self._port = None 64 await super().close()
Disconnect from hardware. Idempotent. Base impl flips the flag.
monitor: jmri_ncemonitor.CabBusMonitor