jmri_loconet.throttle

  1"""``LnThrottle`` + ``LnThrottleManager`` — LocoNet throttle handling.
  2
  3Acquires a slot via :class:`SlotManager`, then routes speed/direction/function
  4changes to ``OPC_LOCO_SPD`` / ``OPC_LOCO_DIRF`` / ``OPC_LOCO_SND`` against
  5that slot. F9–F28 are sent as raw NMRA function-group packets wrapped in
  6``OPC_IMM_PACKET`` (see :mod:`jmri_loconet.dcc`).
  7
  8Function packing (matches Digitrax's DIRF byte layout):
  9    bit 5 = direction (0 = forward, 1 = reverse)
 10    bit 4 = F0 (headlight)
 11    bits 3..0 = F4..F1
 12And SND:
 13    bits 3..0 = F8..F5
 14
 15The :class:`LnThrottleManager` runs a periodic *slot refresh task* —
 16every ~90 seconds it sends a null-move on every active slot, so the
 17command station doesn't purge them during long ops sessions.
 18"""
 19from __future__ import annotations
 20
 21import asyncio
 22import logging
 23from collections.abc import Callable
 24
 25from jmri_core.beans.throttle import Throttle
 26from jmri_core.beans.throttle_manager import ThrottleManager
 27from jmri_core.state import ThrottleDirection
 28
 29from jmri_loconet.dcc import function_high_packet_dcc, wrap_in_imm_packet
 30from jmri_loconet.messages import OpcLocoDirF, OpcLocoSnd, OpcLocoSpd, OpcMoveSlots, OpcUnknown
 31from jmri_loconet.opcodes import OPC_IMM_PACKET
 32from jmri_loconet.slots import Slot, SlotManager
 33from jmri_loconet.traffic import LocoNetTrafficController
 34
 35_log = logging.getLogger(__name__)
 36
 37# CS purges an idle slot after ~120s; refresh well before that.
 38SLOT_REFRESH_INTERVAL_S = 90.0
 39
 40
 41def pack_dirf(direction: ThrottleDirection, functions: list[bool]) -> int:
 42    """Pack direction + F0..F4 into the DIRF byte."""
 43    byte = 0
 44    if direction == ThrottleDirection.REVERSE:
 45        byte |= 0x20
 46    if len(functions) >= 1 and functions[0]:
 47        byte |= 0x10  # F0
 48    if len(functions) >= 5 and functions[4]:
 49        byte |= 0x08  # F4
 50    if len(functions) >= 4 and functions[3]:
 51        byte |= 0x04  # F3
 52    if len(functions) >= 3 and functions[2]:
 53        byte |= 0x02  # F2
 54    if len(functions) >= 2 and functions[1]:
 55        byte |= 0x01  # F1
 56    return byte
 57
 58
 59def pack_snd(functions: list[bool]) -> int:
 60    """Pack F5..F8 into the SND byte."""
 61    byte = 0
 62    if len(functions) >= 9 and functions[8]:
 63        byte |= 0x08  # F8
 64    if len(functions) >= 8 and functions[7]:
 65        byte |= 0x04  # F7
 66    if len(functions) >= 7 and functions[6]:
 67        byte |= 0x02  # F6
 68    if len(functions) >= 6 and functions[5]:
 69        byte |= 0x01  # F5
 70    return byte
 71
 72
 73def speed_to_wire(speed: float) -> int:
 74    """Map [-1.0, 1.0] throttle speed to LocoNet's 0..127 wire form.
 75
 76    -1.0 is emergency stop (wire 1); 0.0 is stop (wire 0); positive maps
 77    linearly into 2..127. Negative non-emergency values (direction is set
 78    separately) are treated as |speed|.
 79    """
 80    if speed == -1.0:
 81        return 1
 82    if speed == 0.0:
 83        return 0
 84    return min(127, max(2, round(abs(speed) * 127)))
 85
 86
 87class LnThrottle(Throttle):
 88    """A LocoNet throttle bound to one acquired slot."""
 89
 90    def __init__(self, address: int, slot: Slot, tc: LocoNetTrafficController,
 91                 long_address: bool = False) -> None:
 92        super().__init__(address=address, long_address=long_address)
 93        self._slot = slot
 94        self._tc = tc
 95
 96    @property
 97    def slot(self) -> Slot:
 98        return self._slot
 99
100    async def set_speed(self, speed: float) -> None:
101        await super().set_speed(speed)
102        await self._tc.send(OpcLocoSpd(slot=self._slot.number, speed=speed_to_wire(speed)))
103
104    async def set_direction(self, direction: ThrottleDirection) -> None:
105        await super().set_direction(direction)
106        await self._push_dirf()
107
108    async def set_function(self, index: int, on: bool) -> None:
109        await super().set_function(index, on)
110        if index <= 4:
111            await self._push_dirf()
112        elif index <= 8:
113            await self._tc.send(OpcLocoSnd(slot=self._slot.number, snd=pack_snd(self.functions)))
114        elif index <= 20:
115            await self._push_func_group(group=3)
116        elif index <= 28:
117            await self._push_func_group(group=4)
118
119    async def _push_func_group(self, *, group: int) -> None:
120        """Build an NMRA function-high DCC packet and ship it via OPC_IMM_PACKET."""
121        dcc = function_high_packet_dcc(
122            address=self.address,
123            long_address=self.long_address,
124            functions=self.functions,
125            group=group,
126        )
127        body = wrap_in_imm_packet(dcc, repeat=2)
128        # Send as a raw OpcUnknown with the IMM_PACKET opcode — the
129        # traffic controller will compute and append the checksum.
130        # The first byte of body is already the opcode; strip it so
131        # OpcUnknown's payload contains just the rest.
132        assert body[0] == OPC_IMM_PACKET
133        await self._tc.send(OpcUnknown(opcode=OPC_IMM_PACKET, payload=body[1:]))
134
135    async def _push_dirf(self) -> None:
136        await self._tc.send(OpcLocoDirF(
137            slot=self._slot.number,
138            dirf=pack_dirf(self.direction, self.functions),
139        ))
140
141
142class LnThrottleManager(ThrottleManager):
143    """LocoNet throttle manager — extends the core ThrottleManager.
144
145    Override hook :meth:`_construct` acquires a slot via :class:`SlotManager`
146    before handing the throttle back. :meth:`release` is overridden to
147    return the slot to the command station as well as drop the bookkeeping.
148
149    Owns the slot-refresh task: periodically sends a null-move
150    (``OPC_MOVE_SLOTS src=N dest=N``) on every active slot so the CS
151    doesn't purge it during long operations.
152    """
153
154    def __init__(
155        self,
156        system_prefix: str,
157        *,
158        tc: LocoNetTrafficController,
159        slot_manager: SlotManager,
160        refresh_interval: float = SLOT_REFRESH_INTERVAL_S,
161    ) -> None:
162        super().__init__(system_prefix)
163        self._tc = tc
164        self._slots = slot_manager
165        self._refresh_interval = refresh_interval
166        self._refresh_task: asyncio.Task[None] | None = None
167        self._refresh_stop = asyncio.Event()
168
169    async def _construct(self, address: int, *, long_address: bool) -> LnThrottle:
170        slot = await self._slots.acquire(address)
171        await self._slots.claim(slot)
172        return LnThrottle(
173            address=address, slot=slot, tc=self._tc, long_address=long_address
174        )
175
176    async def release(self, throttle: Throttle) -> None:
177        if isinstance(throttle, LnThrottle):
178            await self._slots.release(throttle.slot)
179        await super().release(throttle)
180
181    # ----- slot refresh task ------------------------------------------------
182
183    def start_refresh(self) -> None:
184        """Start the periodic slot-refresh background task. Idempotent."""
185        if self._refresh_task is not None:
186            return
187        self._refresh_stop.clear()
188        self._refresh_task = asyncio.create_task(
189            self._refresh_loop(), name="loconet-slot-refresh",
190        )
191
192    async def stop_refresh(self) -> None:
193        """Stop the slot-refresh task. Idempotent."""
194        if self._refresh_task is None:
195            return
196        self._refresh_stop.set()
197        self._refresh_task.cancel()
198        import contextlib  # noqa: PLC0415
199        with contextlib.suppress(asyncio.CancelledError, Exception):
200            await self._refresh_task
201        self._refresh_task = None
202
203    async def _refresh_loop(self) -> None:
204        import contextlib  # noqa: PLC0415
205        try:
206            while not self._refresh_stop.is_set():
207                with contextlib.suppress(TimeoutError):
208                    await asyncio.wait_for(
209                        self._refresh_stop.wait(),
210                        timeout=self._refresh_interval,
211                    )
212                if self._refresh_stop.is_set():
213                    return
214                await self._refresh_all_slots()
215        except asyncio.CancelledError:
216            return
217
218    async def _refresh_all_slots(self) -> None:
219        """Send a null-move for every currently-active throttle's slot."""
220        for throttle in list(self):
221            if not isinstance(throttle, LnThrottle):
222                continue
223            try:
224                await self._tc.send(OpcMoveSlots(
225                    src=throttle.slot.number, dest=throttle.slot.number,
226                ))
227            except Exception as e:  # noqa: BLE001
228                _log.warning("slot refresh failed for slot %d: %s", throttle.slot.number, e)
229
230
231# Touch unused import to keep ruff happy until the next change uses it.
232_ = (Callable,)