jmri_dccex.throttle
DCC-EX throttle wiring.
DCC-EX uses bracket-delimited ASCII for throttle commands:
<t cab speed direction>— set speed/direction for a cab.cab= DCC address (long addresses ≥128 work directly).speed= 0..127 (NMRA 128-step value; 1 = e-stop).direction= 1 (forward) or 0 (reverse).
<F cab function 1|0>— set function on/off.
The DCC-EX command station tracks loco state internally; we treat
throttle commands as fire-and-forget. Subsequent <l> notifications
from the CS update bean state if a feedback channel is wired.
1"""DCC-EX throttle wiring. 2 3DCC-EX uses bracket-delimited ASCII for throttle commands: 4 5* ``<t cab speed direction>`` — set speed/direction for a cab. 6 * ``cab`` = DCC address (long addresses ≥128 work directly). 7 * ``speed`` = 0..127 (NMRA 128-step value; 1 = e-stop). 8 * ``direction`` = 1 (forward) or 0 (reverse). 9* ``<F cab function 1|0>`` — set function on/off. 10 11The DCC-EX command station tracks loco state internally; we treat 12throttle commands as fire-and-forget. Subsequent ``<l>`` notifications 13from the CS update bean state if a feedback channel is wired. 14""" 15from __future__ import annotations 16 17from jmri_core.beans.throttle import Throttle 18from jmri_core.beans.throttle_manager import ThrottleManager 19from jmri_core.state import ThrottleDirection 20 21from jmri_dccex.codec import DccexMessage 22from jmri_dccex.traffic import DccexTrafficController 23 24 25def speed_to_wire_128(speed: float) -> int: 26 """Map [-1.0, 1.0] throttle speed to DCC-EX's 0..127 wire form. 27 28 * ``-1.0`` → 1 (NMRA emergency stop) 29 * ``0.0`` → 0 (stop) 30 * positive maps linearly to 2..127 31 """ 32 if speed == -1.0: 33 return 1 34 if speed == 0.0: 35 return 0 36 return min(127, max(2, round(abs(speed) * 127))) 37 38 39class DccexThrottle(Throttle): 40 """DCC-EX-aware throttle. Sends ``<t cab speed dir>`` on speed/direction.""" 41 42 def __init__( 43 self, address: int, tc: DccexTrafficController, *, long_address: bool = False 44 ) -> None: 45 super().__init__(address=address, long_address=long_address) 46 self._tc = tc 47 48 async def set_speed(self, speed: float) -> None: 49 await super().set_speed(speed) 50 await self._push_speed() 51 52 async def set_direction(self, direction: ThrottleDirection) -> None: 53 await super().set_direction(direction) 54 await self._push_speed() 55 56 async def set_function(self, index: int, on: bool) -> None: 57 await super().set_function(index, on) 58 # DCC-EX <F cab function 1|0>. 59 await self._tc.send(DccexMessage( 60 command="F", args=(str(self.address), str(index), "1" if on else "0"), 61 )) 62 63 async def _push_speed(self) -> None: 64 direction = "1" if self.direction == ThrottleDirection.FORWARD else "0" 65 wire = speed_to_wire_128(self.speed) 66 await self._tc.send(DccexMessage( 67 command="t", args=(str(self.address), str(wire), direction), 68 )) 69 70 71class DccexThrottleManager(ThrottleManager): 72 """DCC-EX throttle manager — uses the core ThrottleManager hook.""" 73 74 def __init__(self, system_prefix: str, *, tc: DccexTrafficController) -> None: 75 super().__init__(system_prefix) 76 self._tc = tc 77 78 async def _construct(self, address: int, *, long_address: bool) -> DccexThrottle: 79 return DccexThrottle(address=address, tc=self._tc, long_address=long_address)
def
speed_to_wire_128(speed: float) -> int:
26def speed_to_wire_128(speed: float) -> int: 27 """Map [-1.0, 1.0] throttle speed to DCC-EX's 0..127 wire form. 28 29 * ``-1.0`` → 1 (NMRA emergency stop) 30 * ``0.0`` → 0 (stop) 31 * positive maps linearly to 2..127 32 """ 33 if speed == -1.0: 34 return 1 35 if speed == 0.0: 36 return 0 37 return min(127, max(2, round(abs(speed) * 127)))
Map [-1.0, 1.0] throttle speed to DCC-EX's 0..127 wire form.
-1.0→ 1 (NMRA emergency stop)0.0→ 0 (stop)- positive maps linearly to 2..127
40class DccexThrottle(Throttle): 41 """DCC-EX-aware throttle. Sends ``<t cab speed dir>`` on speed/direction.""" 42 43 def __init__( 44 self, address: int, tc: DccexTrafficController, *, long_address: bool = False 45 ) -> None: 46 super().__init__(address=address, long_address=long_address) 47 self._tc = tc 48 49 async def set_speed(self, speed: float) -> None: 50 await super().set_speed(speed) 51 await self._push_speed() 52 53 async def set_direction(self, direction: ThrottleDirection) -> None: 54 await super().set_direction(direction) 55 await self._push_speed() 56 57 async def set_function(self, index: int, on: bool) -> None: 58 await super().set_function(index, on) 59 # DCC-EX <F cab function 1|0>. 60 await self._tc.send(DccexMessage( 61 command="F", args=(str(self.address), str(index), "1" if on else "0"), 62 )) 63 64 async def _push_speed(self) -> None: 65 direction = "1" if self.direction == ThrottleDirection.FORWARD else "0" 66 wire = speed_to_wire_128(self.speed) 67 await self._tc.send(DccexMessage( 68 command="t", args=(str(self.address), str(wire), direction), 69 ))
DCC-EX-aware throttle. Sends <t cab speed dir> on speed/direction.
DccexThrottle( address: int, tc: jmri_dccex.DccexTrafficController, *, long_address: bool = False)
72class DccexThrottleManager(ThrottleManager): 73 """DCC-EX throttle manager — uses the core ThrottleManager hook.""" 74 75 def __init__(self, system_prefix: str, *, tc: DccexTrafficController) -> None: 76 super().__init__(system_prefix) 77 self._tc = tc 78 79 async def _construct(self, address: int, *, long_address: bool) -> DccexThrottle: 80 return DccexThrottle(address=address, tc=self._tc, long_address=long_address)
DCC-EX throttle manager — uses the core ThrottleManager hook.
DccexThrottleManager(system_prefix: str, *, tc: jmri_dccex.DccexTrafficController)