jmri_loconet

256class LnOpsProgrammer(Programmer):
257    """Ops-mode adapter scoped to a specific DCC address."""
258
259    def __init__(self, inner: LnProgrammer, address: int, *, long_address: bool) -> None:
260        self._inner = inner
261        self._address = address
262        self._long = long_address
263
264    @property
265    def mode(self) -> ProgrammingMode:
266        return ProgrammingMode.OPS_MODE
267
268    @property
269    def address(self) -> int:
270        return self._address
271
272    async def read_cv(self, cv: int) -> int:
273        return await self._inner.read_cv_ops(self._address, cv)
274
275    async def write_cv(self, cv: int, value: int) -> None:
276        await self._inner.write_cv_ops(self._address, cv, value)

130class LnPowerManager(PowerManager):
131    """LocoNet track power. Sends OPC_GPON/OPC_GPOFF/OPC_IDLE."""
132
133    def __init__(self, tc: LocoNetTrafficController) -> None:
134        super().__init__()
135        self._tc = tc
136        self._unsub = tc.inbound.connect(self._on_message)
137
138    async def set_power(self, state: PowerState) -> None:
139        if state == PowerState.ON:
140            await self._tc.send(OpcGpOn())
141        elif state == PowerState.OFF:
142            await self._tc.send(OpcGpOff())
143        elif state == PowerState.IDLE:
144            await self._tc.send(OpcIdle())
145        await self._set_state(state)
146
147    async def _on_message(self, msg: LocoNetMessage) -> None:
148        if isinstance(msg, OpcGpOn):
149            await self._set_state(PowerState.ON)
150        elif isinstance(msg, OpcGpOff):
151            await self._set_state(PowerState.OFF)
152        elif isinstance(msg, OpcIdle):
153            await self._set_state(PowerState.IDLE)

139class LnProgrammer:
140    """Service- and ops-mode CV access against a LocoNet command station."""
141
142    def __init__(self, tc: LocoNetTrafficController) -> None:
143        self._tc = tc
144        # One waiter at a time (LocoNet programming is sequential).
145        self._busy = asyncio.Lock()
146
147    async def write_cv_direct(self, cv: int, value: int, *, timeout: float = 6.0) -> None:
148        """Service-mode-direct CV write. Blocks until OPC_LONG_ACK arrives."""
149        async with self._busy:
150            await self._send_and_wait_ack(
151                _build_prog_slot_message(PCMD_SERVICE_DIRECT_WRITE, cv, value),
152                timeout=timeout,
153            )
154
155    async def read_cv_direct(self, cv: int, *, timeout: float = 6.0) -> int:
156        """Service-mode-direct CV read. Returns the value byte."""
157        async with self._busy:
158            return await self._send_and_wait_read(
159                _build_prog_slot_message(PCMD_SERVICE_DIRECT_READ, cv, 0),
160                timeout=timeout,
161            )
162
163    async def write_cv_ops(
164        self, address: int, cv: int, value: int, *, timeout: float = 2.0,
165    ) -> None:
166        """Ops-mode CV write to *address*. No reply needed; we fire-and-forget."""
167        async with self._busy:
168            await self._tc.send(_build_prog_slot_message(
169                PCMD_OPS_MODE_WRITE, cv, value, ops_address=address,
170            ))
171
172    async def read_cv_ops(
173        self, address: int, cv: int, *, timeout: float = 6.0,
174    ) -> int:
175        """Ops-mode CV read — requires RailCom-capable decoder + readers."""
176        async with self._busy:
177            return await self._send_and_wait_read(
178                _build_prog_slot_message(
179                    PCMD_OPS_MODE_READ, cv, 0, ops_address=address,
180                ),
181                timeout=timeout,
182            )
183
184    # ----- internals -------------------------------------------------------
185
186    async def _send_and_wait_ack(self, msg: OpcWrSlData, *, timeout: float) -> None:
187        future: asyncio.Future[None] = asyncio.get_event_loop().create_future()
188
189        async def listener(inbound: LocoNetMessage) -> None:
190            if future.done():
191                return
192            if isinstance(inbound, OpcLongAck) and inbound.responding_opcode == OPC_WR_SL_DATA:
193                if inbound.ack_code == LONG_ACK_WRITE_DONE:
194                    future.set_result(None)
195                else:
196                    future.set_exception(
197                        RuntimeError(f"LocoNet programming write failed: ack={inbound.ack_code:#x}")
198                    )
199
200        disconnect = self._tc.inbound.connect(listener)
201        try:
202            await self._tc.send(msg)
203            await asyncio.wait_for(future, timeout=timeout)
204        finally:
205            disconnect()
206
207    async def _send_and_wait_read(self, msg: OpcWrSlData, *, timeout: float) -> int:
208        future: asyncio.Future[int] = asyncio.get_event_loop().create_future()
209
210        async def listener(inbound: LocoNetMessage) -> None:
211            if future.done():
212                return
213            if isinstance(inbound, OpcSlRdData) and inbound.slot == PROGRAMMING_SLOT:
214                # The slot-rd frame carries (CVH, DATA7) at the same byte
215                # positions our OpcSlRdData decoded as (address_high, snd).
216                cvh = (inbound.address >> 7) & 0x7F
217                value = _unpack_data(cvh, inbound.snd)
218                future.set_result(value)
219            elif (
220                isinstance(inbound, OpcLongAck)
221                and inbound.responding_opcode == OPC_WR_SL_DATA
222                and inbound.ack_code == LONG_ACK_READ_FAIL
223            ):
224                future.set_exception(
225                    RuntimeError("LocoNet programming read failed (decoder did not respond)")
226                )
227
228        disconnect = self._tc.inbound.connect(listener)
229        try:
230            await self._tc.send(msg)
231            return await asyncio.wait_for(future, timeout=timeout)
232        finally:
233            disconnect()

279class LnProgrammerManager(ProgrammerManager):
280    """Hands out LocoNet service/ops programmers backed by one :class:`LnProgrammer`."""
281
282    def __init__(self, inner: LnProgrammer) -> None:
283        self._inner = inner
284        self._service_singleton: LnServiceProgrammer | None = None
285
286    def global_programmer(self) -> LnServiceProgrammer:
287        if self._service_singleton is None:
288            self._service_singleton = LnServiceProgrammer(self._inner)
289        return self._service_singleton
290
291    def ops_programmer(self, address: int, *, long_address: bool = False) -> LnOpsProgrammer:
292        return LnOpsProgrammer(self._inner, address, long_address=long_address)
293
294    @property
295    def inner(self) -> LnProgrammer:
296        return self._inner

88class LnSensor(Sensor):
89    """LocoNet-aware sensor — read-only from the layout."""
90
91    __slots__ = ("_address",)
92
93    def __init__(self, system_name: str, address: int) -> None:
94        super().__init__(system_name)
95        self._address = address
96
97    @property
98    def address(self) -> int:
99        return self._address

102class LnSensorManager(Manager[LnSensor]):
103    """Manager that constructs LocoNet sensors and listens for OPC_INPUT_REP."""
104
105    type_letter = "S"
106
107    def __init__(self, system_prefix: str, *, tc: LocoNetTrafficController) -> None:
108        super().__init__(system_prefix)
109        self._tc = tc
110        self._by_address: dict[int, LnSensor] = {}
111        self._unsub = tc.inbound.connect(self._on_message)
112
113    def _create(self, system_name: str) -> LnSensor:
114        address = _parse_address(system_name, self.full_prefix)
115        bean = LnSensor(system_name, address)
116        self._by_address[address] = bean
117        return bean
118
119    async def _on_message(self, msg: LocoNetMessage) -> None:
120        if not isinstance(msg, OpcInputRep):
121            return
122        bean = self._by_address.get(msg.address)
123        if bean is None:
124            # Sensors aren't always pre-created; clients can opt in via provide().
125            return
126        state = SensorState.ACTIVE if msg.active else SensorState.INACTIVE
127        await bean._set_known_state(state)

239class LnServiceProgrammer(Programmer):
240    """Service-mode-direct adapter satisfying the core ABC."""
241
242    def __init__(self, inner: LnProgrammer) -> None:
243        self._inner = inner
244
245    @property
246    def mode(self) -> ProgrammingMode:
247        return ProgrammingMode.SERVICE_DIRECT
248
249    async def read_cv(self, cv: int) -> int:
250        return await self._inner.read_cv_direct(cv)
251
252    async def write_cv(self, cv: int, value: int) -> None:
253        await self._inner.write_cv_direct(cv, value)

 88class LnThrottle(Throttle):
 89    """A LocoNet throttle bound to one acquired slot."""
 90
 91    def __init__(self, address: int, slot: Slot, tc: LocoNetTrafficController,
 92                 long_address: bool = False) -> None:
 93        super().__init__(address=address, long_address=long_address)
 94        self._slot = slot
 95        self._tc = tc
 96
 97    @property
 98    def slot(self) -> Slot:
 99        return self._slot
100
101    async def set_speed(self, speed: float) -> None:
102        await super().set_speed(speed)
103        await self._tc.send(OpcLocoSpd(slot=self._slot.number, speed=speed_to_wire(speed)))
104
105    async def set_direction(self, direction: ThrottleDirection) -> None:
106        await super().set_direction(direction)
107        await self._push_dirf()
108
109    async def set_function(self, index: int, on: bool) -> None:
110        await super().set_function(index, on)
111        if index <= 4:
112            await self._push_dirf()
113        elif index <= 8:
114            await self._tc.send(OpcLocoSnd(slot=self._slot.number, snd=pack_snd(self.functions)))
115        elif index <= 20:
116            await self._push_func_group(group=3)
117        elif index <= 28:
118            await self._push_func_group(group=4)
119
120    async def _push_func_group(self, *, group: int) -> None:
121        """Build an NMRA function-high DCC packet and ship it via OPC_IMM_PACKET."""
122        dcc = function_high_packet_dcc(
123            address=self.address,
124            long_address=self.long_address,
125            functions=self.functions,
126            group=group,
127        )
128        body = wrap_in_imm_packet(dcc, repeat=2)
129        # Send as a raw OpcUnknown with the IMM_PACKET opcode — the
130        # traffic controller will compute and append the checksum.
131        # The first byte of body is already the opcode; strip it so
132        # OpcUnknown's payload contains just the rest.
133        assert body[0] == OPC_IMM_PACKET
134        await self._tc.send(OpcUnknown(opcode=OPC_IMM_PACKET, payload=body[1:]))
135
136    async def _push_dirf(self) -> None:
137        await self._tc.send(OpcLocoDirF(
138            slot=self._slot.number,
139            dirf=pack_dirf(self.direction, self.functions),
140        ))

143class LnThrottleManager(ThrottleManager):
144    """LocoNet throttle manager — extends the core ThrottleManager.
145
146    Override hook :meth:`_construct` acquires a slot via :class:`SlotManager`
147    before handing the throttle back. :meth:`release` is overridden to
148    return the slot to the command station as well as drop the bookkeeping.
149
150    Owns the slot-refresh task: periodically sends a null-move
151    (``OPC_MOVE_SLOTS src=N dest=N``) on every active slot so the CS
152    doesn't purge it during long operations.
153    """
154
155    def __init__(
156        self,
157        system_prefix: str,
158        *,
159        tc: LocoNetTrafficController,
160        slot_manager: SlotManager,
161        refresh_interval: float = SLOT_REFRESH_INTERVAL_S,
162    ) -> None:
163        super().__init__(system_prefix)
164        self._tc = tc
165        self._slots = slot_manager
166        self._refresh_interval = refresh_interval
167        self._refresh_task: asyncio.Task[None] | None = None
168        self._refresh_stop = asyncio.Event()
169
170    async def _construct(self, address: int, *, long_address: bool) -> LnThrottle:
171        slot = await self._slots.acquire(address)
172        await self._slots.claim(slot)
173        return LnThrottle(
174            address=address, slot=slot, tc=self._tc, long_address=long_address
175        )
176
177    async def release(self, throttle: Throttle) -> None:
178        if isinstance(throttle, LnThrottle):
179            await self._slots.release(throttle.slot)
180        await super().release(throttle)
181
182    # ----- slot refresh task ------------------------------------------------
183
184    def start_refresh(self) -> None:
185        """Start the periodic slot-refresh background task. Idempotent."""
186        if self._refresh_task is not None:
187            return
188        self._refresh_stop.clear()
189        self._refresh_task = asyncio.create_task(
190            self._refresh_loop(), name="loconet-slot-refresh",
191        )
192
193    async def stop_refresh(self) -> None:
194        """Stop the slot-refresh task. Idempotent."""
195        if self._refresh_task is None:
196            return
197        self._refresh_stop.set()
198        self._refresh_task.cancel()
199        import contextlib  # noqa: PLC0415
200        with contextlib.suppress(asyncio.CancelledError, Exception):
201            await self._refresh_task
202        self._refresh_task = None
203
204    async def _refresh_loop(self) -> None:
205        import contextlib  # noqa: PLC0415
206        try:
207            while not self._refresh_stop.is_set():
208                with contextlib.suppress(TimeoutError):
209                    await asyncio.wait_for(
210                        self._refresh_stop.wait(),
211                        timeout=self._refresh_interval,
212                    )
213                if self._refresh_stop.is_set():
214                    return
215                await self._refresh_all_slots()
216        except asyncio.CancelledError:
217            return
218
219    async def _refresh_all_slots(self) -> None:
220        """Send a null-move for every currently-active throttle's slot."""
221        for throttle in list(self):
222            if not isinstance(throttle, LnThrottle):
223                continue
224            try:
225                await self._tc.send(OpcMoveSlots(
226                    src=throttle.slot.number, dest=throttle.slot.number,
227                ))
228            except Exception as e:  # noqa: BLE001
229                _log.warning("slot refresh failed for slot %d: %s", throttle.slot.number, e)

30class LnTurnout(Turnout):
31    """LocoNet-aware turnout: sends OPC_SW_REQ on commanded-state change."""
32
33    __slots__ = ("_address", "_tc")
34
35    def __init__(self, system_name: str, address: int, tc: LocoNetTrafficController) -> None:
36        super().__init__(system_name)
37        self._address = address
38        self._tc = tc
39
40    @property
41    def address(self) -> int:
42        return self._address
43
44    async def set_commanded_state(self, state: TurnoutState) -> None:
45        if state in (TurnoutState.CLOSED, TurnoutState.THROWN):
46            await self._tc.send(OpcSwReq(
47                address=self._address,
48                closed=(state == TurnoutState.CLOSED),
49                output_on=True,
50            ))
51        # Update the in-memory state via the base implementation; for
52        # DIRECT-feedback turnouts known_state will mirror this until/unless
53        # a layout report overrides it.
54        await super().set_commanded_state(state)

57class LnTurnoutManager(Manager[LnTurnout]):
58    """Manager that constructs LocoNet turnouts and tracks them by address."""
59
60    type_letter = "T"
61
62    def __init__(self, system_prefix: str, *, tc: LocoNetTrafficController) -> None:
63        super().__init__(system_prefix)
64        self._tc = tc
65        self._by_address: dict[int, LnTurnout] = {}
66        self._unsub = tc.inbound.connect(self._on_message)
67
68    def _create(self, system_name: str) -> LnTurnout:
69        address = _parse_address(system_name, self.full_prefix)
70        bean = LnTurnout(system_name, address, self._tc)
71        self._by_address[address] = bean
72        return bean
73
74    async def deregister(self, bean: LnTurnout) -> None:
75        self._by_address.pop(bean.address, None)
76        await super().deregister(bean)
77
78    async def _on_message(self, msg: LocoNetMessage) -> None:
79        if not isinstance(msg, OpcSwRep):
80            return
81        bean = self._by_address.get(msg.address)
82        if bean is None:
83            return
84        state = TurnoutState.CLOSED if msg.closed else TurnoutState.THROWN
85        await bean._set_known_state(state)

 53class LocoNetConnection(SystemConnection):
 54    """LocoNet hardware connection. Prefix ``L``, name ``LocoNet``."""
 55
 56    name: ClassVar[str] = "LocoNet"
 57    prefix: ClassVar[str] = "L"
 58
 59    def __init__(self) -> None:
 60        super().__init__()
 61        # These are filled in by open(). The base-class defaults stay in place
 62        # until then so the connection has well-formed managers if opened with
 63        # no settings (rare; mainly the test path).
 64        self._port: LocoNetPort | None = None
 65        self._tc: LocoNetTrafficController | None = None
 66        self._slots: SlotManager | None = None
 67        self._programmer: LnProgrammer | None = None
 68        self._cs_monitor_task: asyncio.Task[None] | None = None
 69
 70    # ----- lifecycle --------------------------------------------------------
 71
 72    async def open(self, settings: ConnectionSettings | None = None) -> None:
 73        s = settings or {}
 74        self._port = await _build_port(s)
 75        self._tc = LocoNetTrafficController(self._port)
 76        await self._tc.start()
 77        self._slots = SlotManager(self._tc)
 78
 79        # Replace the base default managers with our LocoNet-aware ones.
 80        # ``LnTurnoutManager`` and friends are subclasses of the base managers,
 81        # but mypy's invariance in attribute typing means we have to ignore
 82        # the assignment — the runtime substitution is sound (Liskov).
 83        self.turnouts = LnTurnoutManager(self.prefix, tc=self._tc)   # type: ignore[assignment]
 84        self.sensors = LnSensorManager(self.prefix, tc=self._tc)     # type: ignore[assignment]
 85        throttle_mgr = LnThrottleManager(
 86            self.prefix, tc=self._tc, slot_manager=self._slots,
 87        )
 88        self.throttles = throttle_mgr
 89        # Rebind the consist manager onto the LocoNet-aware throttle
 90        # manager. Without this, consist fan-out drives placeholder
 91        # Throttle beans left over from base ``__init__``, and no
 92        # OPC_LOCO_SPD frames reach the wire.
 93        self.consists.rebind_throttles(self.throttles)
 94        throttle_mgr.start_refresh()
 95        self.consists = LnConsistManager(   # type: ignore[assignment]
 96            throttle_mgr, self.prefix, tc=self._tc, slots=self._slots,
 97        )
 98        self.power = LnPowerManager(self._tc)
 99        self._programmer = LnProgrammer(self._tc)
100        self.programmers = LnProgrammerManager(self._programmer)
101        # Wire the advanced-consist CV19 write path to the LocoNet
102        # programmer manager. Without this rebind, ``advanced`` consists
103        # go through the base placeholder manager and CV19 never
104        # actually programs.
105        self.consists.rebind_programmers(self.programmers)
106
107        # CS reachability monitor. LocoNet CS emit OPC_IDLE and other
108        # frames continuously; 3s of complete silence is a strong
109        # signal the bus is dark (CS off or unplugged). Start
110        # pessimistic (base class default is optimistic True) so a
111        # home-page render during the monitor's startup grace shows
112        # "Off" until we've actually seen a frame, not a false "On".
113        self._cs_reachable = False
114        self._cs_monitor_task = asyncio.create_task(
115            self._cs_reachability_loop(), name="loconet-cs-reachability",
116        )
117
118        await super().open(settings)
119
120    async def _cs_reachability_loop(self) -> None:
121        """Flip ``_cs_reachable`` on the 3-second inbound-silence rule.
122
123        LocoNet's continuous chatter (CS heartbeat + slot poll traffic)
124        means any real silence past 3 seconds indicates the CS has
125        stopped talking. Any fresh inbound frame flips reachable back
126        to True immediately.
127        """
128        import contextlib  # noqa: PLC0415
129        try:
130            # 3-second grace so the first CS heartbeat can arrive.
131            await asyncio.sleep(3.0)
132            while True:
133                await asyncio.sleep(1.0)
134                if self._tc is None:
135                    return
136                last = self._tc.last_inbound_at
137                if last is None:
138                    await self._set_cs_reachable(False)
139                    continue
140                age = time.monotonic() - last
141                await self._set_cs_reachable(age <= 3.0)
142        except asyncio.CancelledError:
143            with contextlib.suppress(Exception):
144                return
145
146    async def close(self) -> None:
147        if isinstance(self.throttles, LnThrottleManager):
148            await self.throttles.stop_refresh()
149        if self._cs_monitor_task is not None:
150            self._cs_monitor_task.cancel()
151            import contextlib  # noqa: PLC0415
152            with contextlib.suppress(asyncio.CancelledError, Exception):
153                await self._cs_monitor_task
154            self._cs_monitor_task = None
155        if self._tc is not None:
156            await self._tc.aclose()
157            self._tc = None
158        self._port = None
159        await super().close()
160
161    # ----- accessors --------------------------------------------------------
162
163    @property
164    def slots(self) -> SlotManager:
165        if self._slots is None:
166            raise RuntimeError("connection not open")
167        return self._slots
168
169    @property
170    def traffic(self) -> LocoNetTrafficController:
171        if self._tc is None:
172            raise RuntimeError("connection not open")
173        return self._tc
174
175    # ----- fast clock (slot 123) --------------------------------------------
176
177    async def read_hardware_clock(self):  # noqa: ANN201
178        """Read the CS fast clock from slot 123.
179
180        Sends ``OPC_RQ_SL_DATA(slot=123)`` and awaits the paired
181        ``OPC_SL_RD_DATA`` reply on the inbound signal. Returns a
182        :class:`~jmri_core.beans.fast_clock.HardwareClockState`.
183
184        The :class:`~jmri_server.services.fast_clock.TimeService`
185        prefers this method over the software bean whenever the
186        connection exposes it, so wiring this on ``LocoNetConnection``
187        automatically makes ``/json/time`` reflect the CS clock.
188        """
189        from jmri_loconet.clock import FAST_CLOCK_SLOT, decode_slot123  # noqa: PLC0415
190        from jmri_loconet.messages import OpcRqSlData, OpcSlRdData  # noqa: PLC0415
191
192        if self._tc is None:
193            raise RuntimeError("connection not open")
194        tc = self._tc
195        loop = asyncio.get_event_loop()
196        fut: asyncio.Future[OpcSlRdData] = loop.create_future()
197
198        async def on_inbound(msg) -> None:  # noqa: ANN001
199            if (
200                isinstance(msg, OpcSlRdData)
201                and msg.slot == FAST_CLOCK_SLOT
202                and not fut.done()
203            ):
204                fut.set_result(msg)
205
206        unsub = tc.inbound.connect(on_inbound)
207        try:
208            await tc.send(OpcRqSlData(slot=FAST_CLOCK_SLOT))
209            reply = await asyncio.wait_for(fut, timeout=2.0)
210        finally:
211            unsub()
212        return decode_slot123(reply)
213
214    async def _write_hardware_clock(self, state) -> None:  # noqa: ANN001
215        """Common write path — reads the current slot to preserve
216        non-clock fields, merges *state*, and sends the write."""
217        current = await self.read_hardware_clock()
218        # ``current`` gave us the CS's authoritative days / trk / snd
219        # bytes indirectly by reading the raw slot; we lose that in
220        # the decode. To keep this pass simple we send zeros for those
221        # fields — JMRI Java's LnClockControl does the same on write,
222        # relying on the CS to preserve day count internally.
223        from jmri_loconet.clock import build_slot123_write  # noqa: PLC0415
224
225        merged = state.__class__(
226            hours=state.hours if state.hours is not None else current.hours,
227            minutes=state.minutes if state.minutes is not None else current.minutes,
228            seconds=state.seconds if state.seconds is not None else current.seconds,
229            rate=state.rate if state.rate is not None else current.rate,
230            running=state.running if state.running is not None else current.running,
231            twenty_four_hour=True,
232        )
233        msg = build_slot123_write(merged)
234        if self._tc is None:
235            raise RuntimeError("connection not open")
236        await self._tc.send(msg)
237
238    async def set_hardware_clock_time(
239        self, hours: int, minutes: int, seconds: int = 0,
240    ) -> None:
241        from jmri_core.beans.fast_clock import HardwareClockState  # noqa: PLC0415
242        current = await self.read_hardware_clock()
243        await self._write_hardware_clock(HardwareClockState(
244            hours=hours, minutes=minutes, seconds=seconds,
245            rate=current.rate, running=current.running,
246            twenty_four_hour=True,
247        ))
248
249    async def set_hardware_clock_rate(self, rate: float) -> None:
250        from jmri_core.beans.fast_clock import HardwareClockState  # noqa: PLC0415
251        current = await self.read_hardware_clock()
252        await self._write_hardware_clock(HardwareClockState(
253            hours=current.hours, minutes=current.minutes,
254            seconds=current.seconds,
255            rate=rate, running=current.running,
256            twenty_four_hour=True,
257        ))
258
259    async def start_hardware_clock(self) -> None:
260        from jmri_core.beans.fast_clock import HardwareClockState  # noqa: PLC0415
261        current = await self.read_hardware_clock()
262        # LocoNet uses rate=0 to indicate stopped, so "start" without a
263        # remembered rate defaults to 1× (real time). If the CS already
264        # has a stored non-zero rate we preserve it.
265        rate = current.rate if current.rate > 0 else 1.0
266        await self._write_hardware_clock(HardwareClockState(
267            hours=current.hours, minutes=current.minutes,
268            seconds=current.seconds,
269            rate=rate, running=True,
270            twenty_four_hour=True,
271        ))
272
273    async def stop_hardware_clock(self) -> None:
274        from jmri_core.beans.fast_clock import HardwareClockState  # noqa: PLC0415
275        current = await self.read_hardware_clock()
276        await self._write_hardware_clock(HardwareClockState(
277            hours=current.hours, minutes=current.minutes,
278            seconds=current.seconds,
279            rate=current.rate, running=False,
280            twenty_four_hour=True,
281        ))
282
283    # ----- manager overrides ------------------------------------------------
284    # We only customize turnouts, sensors, and power; everything else uses the
285    # base default in-memory managers (suitable since they don't talk to the
286    # hardware). Stage 4.5 can add LnLightManager, LnReporterManager, etc.
287
288    def _make_turnout_manager(self) -> TurnoutManager:  # pragma: no cover - overridden in open
289        return TurnoutManager(self.prefix)
290
291    # Keep all other _make_*_manager hooks at the inherited in-memory defaults.
292    # The Manager generic type-cast below is just to keep mypy happy with the
293    # in-memory defaults that aren't LocoNet-aware.
294    _GENERIC_HINT: ClassVar[type[Manager[Any]]] = LightManager
295    _OTHER_HINTS: ClassVar[tuple[type, ...]] = (
296        BlockManager,
297        MemoryManager,
298        PowerManager,
299        ReporterManager,
300        RouteManager,
301        SignalHeadManager,
302        SignalMastManager,
303    )