jmri_nce.throttle

  1"""NCE throttle support — modern firmware (commandOptions >= 10) path.
  2
  3The throttle is driven via the NCE binary ``nceLocoCmd`` (opcode 0xA2),
  4not raw DCC packets. JMRI Java's ``NceThrottle.sendA2command`` defaults
  5to True for any firmware option >= 10, so this is the right path on
  6PowerCab, PowerPro, and SB5 with current firmware. The legacy
  7``sendPacketMessage`` opcodes (0x93..0x96) collide with the CV
  8write/read commands at the same opcode values; sending throttle
  9packets that way confuses the CS.
 10
 11Per-message wire form (5 bytes):
 12
 13    [0xA2] [addr_hi] [addr_lo] [sub_cmd] [data]
 14
 15* speed change → sub_cmd 3/4 (rev/fwd 128-step), data = 0..126
 16* emergency stop → sub_cmd 5/6 (rev/fwd estop), data = 0
 17* F0..F4 → sub_cmd 7, data bits: F0=bit4, F1=bit0, F2=bit1, F3=bit2, F4=bit3
 18* F5..F8 → sub_cmd 8, data bits: F5=bit0, F6=bit1, F7=bit2, F8=bit3
 19* F9..F12 → sub_cmd 9, same bit ordering
 20* F13..F20 → sub_cmd 21, data = bit-per-function (F13=bit0..F20=bit7)
 21* F21..F28 → sub_cmd 22, data = bit-per-function (F21=bit0..F28=bit7)
 22"""
 23from __future__ import annotations
 24
 25import logging
 26
 27from jmri_core.beans.throttle import Throttle
 28from jmri_core.beans.throttle_manager import ThrottleManager
 29from jmri_core.state import ThrottleDirection
 30
 31from jmri_nce.messages import nce_loco_cmd
 32from jmri_nce.opcodes import (
 33    LOCO_SUB_128_FWD,
 34    LOCO_SUB_128_REV,
 35    LOCO_SUB_ESTOP_FWD,
 36    LOCO_SUB_ESTOP_REV,
 37    LOCO_SUB_FUNC_GROUP_1,
 38    LOCO_SUB_FUNC_GROUP_2,
 39    LOCO_SUB_FUNC_GROUP_3,
 40    LOCO_SUB_FUNC_GROUP_4,
 41    LOCO_SUB_FUNC_GROUP_5,
 42)
 43from jmri_nce.traffic import NceTrafficController
 44
 45_log = logging.getLogger(__name__)
 46
 47
 48def speed_to_wire_128(speed: float) -> int:
 49    """Map [-1.0, 1.0] throttle speed to NCE 128-step magnitude.
 50
 51    NCE uses 0..126 (not 0..127 like NMRA raw packet). Direction is in
 52    the sub-command byte. ``speed == -1.0`` is the e-stop sentinel and
 53    is handled by the throttle's e-stop path, not via this mapping.
 54    """
 55    mag = abs(speed)
 56    if mag == 0.0:
 57        return 0
 58    value = round(126 * mag)
 59    return max(1, min(126, value))
 60
 61
 62def _func_group_1_data(f: list[bool]) -> int:
 63    """F0..F4 bitmask — F0 is bit 4, F1..F4 are bits 0..3 (NCE convention)."""
 64    return (
 65        (16 if f[0] else 0)
 66        | (1 if f[1] else 0)
 67        | (2 if f[2] else 0)
 68        | (4 if f[3] else 0)
 69        | (8 if f[4] else 0)
 70    )
 71
 72
 73def _func_group_2_data(f: list[bool]) -> int:
 74    """F5..F8 — F5..F8 in bits 0..3."""
 75    return (
 76        (1 if f[5] else 0)
 77        | (2 if f[6] else 0)
 78        | (4 if f[7] else 0)
 79        | (8 if f[8] else 0)
 80    )
 81
 82
 83def _func_group_3_data(f: list[bool]) -> int:
 84    """F9..F12 — F9..F12 in bits 0..3."""
 85    return (
 86        (1 if f[9] else 0)
 87        | (2 if f[10] else 0)
 88        | (4 if f[11] else 0)
 89        | (8 if f[12] else 0)
 90    )
 91
 92
 93def _func_group_4_data(f: list[bool]) -> int:
 94    """F13..F20 — bit-per-function, F13=bit0..F20=bit7."""
 95    data = 0
 96    for i in range(8):
 97        if f[13 + i]:
 98            data |= 1 << i
 99    return data
100
101
102def _func_group_5_data(f: list[bool]) -> int:
103    """F21..F28 — bit-per-function, F21=bit0..F28=bit7."""
104    data = 0
105    for i in range(8):
106        if f[21 + i]:
107            data |= 1 << i
108    return data
109
110
111class NceThrottle(Throttle):
112    """Throttle driving an NCE command station via the 0xA2 loco command."""
113
114    def __init__(
115        self, address: int, tc: NceTrafficController, *, long_address: bool = False
116    ) -> None:
117        super().__init__(address=address, long_address=long_address)
118        self._tc = tc
119
120    async def set_speed(self, speed: float) -> None:
121        await super().set_speed(speed)
122        await self._push_speed()
123
124    async def set_direction(self, direction: ThrottleDirection) -> None:
125        await super().set_direction(direction)
126        await self._push_speed()
127
128    async def set_function(self, index: int, on: bool) -> None:
129        await super().set_function(index, on)
130        if index <= 4:
131            await self._push_group(LOCO_SUB_FUNC_GROUP_1, _func_group_1_data(self.functions))
132        elif index <= 8:
133            await self._push_group(LOCO_SUB_FUNC_GROUP_2, _func_group_2_data(self.functions))
134        elif index <= 12:
135            await self._push_group(LOCO_SUB_FUNC_GROUP_3, _func_group_3_data(self.functions))
136        elif index <= 20:
137            await self._push_group(LOCO_SUB_FUNC_GROUP_4, _func_group_4_data(self.functions))
138        else:
139            await self._push_group(LOCO_SUB_FUNC_GROUP_5, _func_group_5_data(self.functions))
140
141    # ----- helpers --------------------------------------------------------
142
143    async def _push_speed(self) -> None:
144        forward = self.direction == ThrottleDirection.FORWARD
145        if self.speed == -1.0:
146            sub = LOCO_SUB_ESTOP_FWD if forward else LOCO_SUB_ESTOP_REV
147            await self._send(sub, 0)
148            return
149        sub = LOCO_SUB_128_FWD if forward else LOCO_SUB_128_REV
150        await self._send(sub, speed_to_wire_128(self.speed))
151
152    async def _push_group(self, sub_cmd: int, data: int) -> None:
153        await self._send(sub_cmd, data)
154
155    async def _send(self, sub_cmd: int, data: int) -> None:
156        cmd = nce_loco_cmd(
157            self.address,
158            long_address=self.long_address,
159            sub_cmd=sub_cmd,
160            data=data,
161        )
162        if _log.isEnabledFor(logging.DEBUG):
163            _log.debug(
164                "NCE throttle out: %s bytes=%s",
165                cmd.name, " ".join(f"{b:02X}" for b in cmd.request),
166            )
167        await self._tc.send_command(cmd)
168
169
170class NceThrottleManager(ThrottleManager):
171    """NCE throttle manager — uses the core ThrottleManager hook."""
172
173    def __init__(self, system_prefix: str, *, tc: NceTrafficController) -> None:
174        super().__init__(system_prefix)
175        self._tc = tc
176
177    async def _construct(self, address: int, *, long_address: bool) -> NceThrottle:
178        return NceThrottle(address=address, tc=self._tc, long_address=long_address)