jmri_core.beans.throttle
Throttle — a runtime control handle for one locomotive (or consist).
A throttle is not a NamedBean — it is acquired and released against a ThrottleManager (in the HAL), and represents a transient hold on a slot in the command station rather than a configured piece of layout equipment.
Attributes:
address— DCC address of the loco (or consist) under control.long_address— whether the address is a 14-bit (long) value.speed— normalized to[0.0, 1.0].-1.0means emergency stop.direction— FORWARD or REVERSE.functions— list of 32 booleans (F0..F31).
1"""``Throttle`` — a runtime control handle for one locomotive (or consist). 2 3A throttle is *not* a NamedBean — it is acquired and released against a 4ThrottleManager (in the HAL), and represents a transient hold on a slot in 5the command station rather than a configured piece of layout equipment. 6 7Attributes: 8 9* ``address`` — DCC address of the loco (or consist) under control. 10* ``long_address`` — whether the address is a 14-bit (long) value. 11* ``speed`` — normalized to ``[0.0, 1.0]``. ``-1.0`` means emergency stop. 12* ``direction`` — FORWARD or REVERSE. 13* ``functions`` — list of 32 booleans (F0..F31). 14""" 15from __future__ import annotations 16 17from dataclasses import dataclass, field 18 19from jmri_core.events import Signal 20from jmri_core.state import ThrottleDirection 21 22NUM_FUNCTIONS = 32 23 24 25@dataclass(frozen=True, slots=True) 26class ThrottleChange: 27 kind: str # "speed" | "direction" | "function" 28 index: int # function index if kind=="function", else -1 29 30 31@dataclass(slots=True) 32class Throttle: 33 """One handle on one locomotive.""" 34 35 address: int 36 long_address: bool = False 37 speed: float = 0.0 38 direction: ThrottleDirection = ThrottleDirection.FORWARD 39 functions: list[bool] = field(default_factory=lambda: [False] * NUM_FUNCTIONS) 40 state_changed: Signal[ThrottleChange] = field(default_factory=Signal) 41 42 def __post_init__(self) -> None: 43 if len(self.functions) != NUM_FUNCTIONS: 44 raise ValueError(f"functions must be length {NUM_FUNCTIONS}") 45 if not (-1.0 <= self.speed <= 1.0): 46 raise ValueError("speed must be in [-1, 1]") 47 48 async def set_speed(self, speed: float) -> None: 49 if not (-1.0 <= speed <= 1.0): 50 raise ValueError("speed must be in [-1, 1]") 51 if speed == self.speed: 52 return 53 self.speed = speed 54 await self.state_changed.emit(ThrottleChange("speed", -1)) 55 56 async def set_direction(self, direction: ThrottleDirection) -> None: 57 if direction == self.direction: 58 return 59 self.direction = direction 60 await self.state_changed.emit(ThrottleChange("direction", -1)) 61 62 async def set_function(self, index: int, on: bool) -> None: 63 if not 0 <= index < NUM_FUNCTIONS: 64 raise ValueError(f"function index out of range: {index}") 65 if self.functions[index] == on: 66 return 67 self.functions[index] = on 68 await self.state_changed.emit(ThrottleChange("function", index)) 69 70 async def emergency_stop(self) -> None: 71 await self.set_speed(-1.0)
NUM_FUNCTIONS =
32
@dataclass(frozen=True, slots=True)
class
ThrottleChange:
@dataclass(slots=True)
class
Throttle:
32@dataclass(slots=True) 33class Throttle: 34 """One handle on one locomotive.""" 35 36 address: int 37 long_address: bool = False 38 speed: float = 0.0 39 direction: ThrottleDirection = ThrottleDirection.FORWARD 40 functions: list[bool] = field(default_factory=lambda: [False] * NUM_FUNCTIONS) 41 state_changed: Signal[ThrottleChange] = field(default_factory=Signal) 42 43 def __post_init__(self) -> None: 44 if len(self.functions) != NUM_FUNCTIONS: 45 raise ValueError(f"functions must be length {NUM_FUNCTIONS}") 46 if not (-1.0 <= self.speed <= 1.0): 47 raise ValueError("speed must be in [-1, 1]") 48 49 async def set_speed(self, speed: float) -> None: 50 if not (-1.0 <= speed <= 1.0): 51 raise ValueError("speed must be in [-1, 1]") 52 if speed == self.speed: 53 return 54 self.speed = speed 55 await self.state_changed.emit(ThrottleChange("speed", -1)) 56 57 async def set_direction(self, direction: ThrottleDirection) -> None: 58 if direction == self.direction: 59 return 60 self.direction = direction 61 await self.state_changed.emit(ThrottleChange("direction", -1)) 62 63 async def set_function(self, index: int, on: bool) -> None: 64 if not 0 <= index < NUM_FUNCTIONS: 65 raise ValueError(f"function index out of range: {index}") 66 if self.functions[index] == on: 67 return 68 self.functions[index] = on 69 await self.state_changed.emit(ThrottleChange("function", index)) 70 71 async def emergency_stop(self) -> None: 72 await self.set_speed(-1.0)
One handle on one locomotive.
Throttle( address: int, long_address: bool = False, speed: float = 0.0, direction: jmri_core.state.ThrottleDirection = <ThrottleDirection.FORWARD: 1>, functions: list[bool] = <factory>, state_changed: jmri_core.Signal[ThrottleChange] = <factory>)
direction: jmri_core.state.ThrottleDirection
state_changed: jmri_core.Signal[ThrottleChange]
async def
set_function(self, index: int, on: bool) -> None:
63 async def set_function(self, index: int, on: bool) -> None: 64 if not 0 <= index < NUM_FUNCTIONS: 65 raise ValueError(f"function index out of range: {index}") 66 if self.functions[index] == on: 67 return 68 self.functions[index] = on 69 await self.state_changed.emit(ThrottleChange("function", index))