jmri_core.beans.route
Route — a named group of turnouts thrown together to a target state.
Setting a route fires each step: turnout.set_commanded_state(target) in
order. A route may have a lock sensor that, when active, prevents firing.
1"""``Route`` — a named group of turnouts thrown together to a target state. 2 3Setting a route fires each step: ``turnout.set_commanded_state(target)`` in 4order. A route may have a *lock* sensor that, when active, prevents firing. 5""" 6from __future__ import annotations 7 8from dataclasses import dataclass 9 10from jmri_core.bean import AbstractNamedBean 11from jmri_core.beans.sensor import Sensor 12from jmri_core.beans.turnout import Turnout 13from jmri_core.manager import Manager 14from jmri_core.state import RouteState, SensorState, TurnoutState 15 16 17@dataclass(slots=True) 18class RouteStep: 19 """One turnout-to-target pair in a route.""" 20 21 turnout: Turnout 22 target: TurnoutState 23 24 25@dataclass(frozen=True, slots=True) 26class RouteFired: 27 """Payload emitted on a successful :meth:`Route.fire`.""" 28 29 steps_executed: int 30 31 32class Route(AbstractNamedBean): 33 """Named collection of turnout commands. Fires in registration order.""" 34 35 __slots__ = ("_lock_sensor", "_state", "steps") 36 37 type_letter = "O" # JMRI uses 'O' for Route (no conflict with Output) 38 39 def __init__(self, system_name: str, user_name: str | None = None) -> None: 40 super().__init__(system_name, user_name) 41 self.steps: list[RouteStep] = [] 42 self._lock_sensor: Sensor | None = None 43 self._state: RouteState = RouteState.UNLOCKED 44 45 @property 46 def state(self) -> RouteState: 47 return self._state 48 49 @property 50 def lock_sensor(self) -> Sensor | None: 51 return self._lock_sensor 52 53 def set_lock_sensor(self, sensor: Sensor | None) -> None: 54 self._lock_sensor = sensor 55 56 def add_step(self, turnout: Turnout, target: TurnoutState) -> None: 57 self.steps.append(RouteStep(turnout, target)) 58 59 async def fire(self) -> int: 60 """Throw every step in order. Returns count of commands sent. 61 62 Raises :class:`RuntimeError` if the lock sensor is ACTIVE. 63 """ 64 if self._lock_sensor and self._lock_sensor.known_state == SensorState.ACTIVE: 65 raise RuntimeError( 66 f"route {self.system_name} is locked by {self._lock_sensor.system_name}" 67 ) 68 count = 0 69 for step in self.steps: 70 await step.turnout.set_commanded_state(step.target) 71 count += 1 72 await self.state_changed.emit(RouteFired(count)) 73 return count 74 75 76class RouteManager(Manager[Route]): 77 type_letter = "O" 78 79 def _create(self, system_name: str) -> Route: 80 return Route(system_name)
@dataclass(slots=True)
class
RouteStep:
18@dataclass(slots=True) 19class RouteStep: 20 """One turnout-to-target pair in a route.""" 21 22 turnout: Turnout 23 target: TurnoutState
One turnout-to-target pair in a route.
RouteStep( turnout: jmri_core.beans.Turnout, target: jmri_core.state.TurnoutState)
turnout: jmri_core.beans.Turnout
target: jmri_core.state.TurnoutState
@dataclass(frozen=True, slots=True)
class
RouteFired:
26@dataclass(frozen=True, slots=True) 27class RouteFired: 28 """Payload emitted on a successful :meth:`Route.fire`.""" 29 30 steps_executed: int
Payload emitted on a successful Route.fire().
33class Route(AbstractNamedBean): 34 """Named collection of turnout commands. Fires in registration order.""" 35 36 __slots__ = ("_lock_sensor", "_state", "steps") 37 38 type_letter = "O" # JMRI uses 'O' for Route (no conflict with Output) 39 40 def __init__(self, system_name: str, user_name: str | None = None) -> None: 41 super().__init__(system_name, user_name) 42 self.steps: list[RouteStep] = [] 43 self._lock_sensor: Sensor | None = None 44 self._state: RouteState = RouteState.UNLOCKED 45 46 @property 47 def state(self) -> RouteState: 48 return self._state 49 50 @property 51 def lock_sensor(self) -> Sensor | None: 52 return self._lock_sensor 53 54 def set_lock_sensor(self, sensor: Sensor | None) -> None: 55 self._lock_sensor = sensor 56 57 def add_step(self, turnout: Turnout, target: TurnoutState) -> None: 58 self.steps.append(RouteStep(turnout, target)) 59 60 async def fire(self) -> int: 61 """Throw every step in order. Returns count of commands sent. 62 63 Raises :class:`RuntimeError` if the lock sensor is ACTIVE. 64 """ 65 if self._lock_sensor and self._lock_sensor.known_state == SensorState.ACTIVE: 66 raise RuntimeError( 67 f"route {self.system_name} is locked by {self._lock_sensor.system_name}" 68 ) 69 count = 0 70 for step in self.steps: 71 await step.turnout.set_commanded_state(step.target) 72 count += 1 73 await self.state_changed.emit(RouteFired(count)) 74 return count
Named collection of turnout commands. Fires in registration order.
steps: list[RouteStep]
state: jmri_core.state.RouteState
lock_sensor: jmri_core.beans.Sensor | None
def
add_step( self, turnout: jmri_core.beans.Turnout, target: jmri_core.state.TurnoutState) -> None:
async def
fire(self) -> int:
60 async def fire(self) -> int: 61 """Throw every step in order. Returns count of commands sent. 62 63 Raises :class:`RuntimeError` if the lock sensor is ACTIVE. 64 """ 65 if self._lock_sensor and self._lock_sensor.known_state == SensorState.ACTIVE: 66 raise RuntimeError( 67 f"route {self.system_name} is locked by {self._lock_sensor.system_name}" 68 ) 69 count = 0 70 for step in self.steps: 71 await step.turnout.set_commanded_state(step.target) 72 count += 1 73 await self.state_changed.emit(RouteFired(count)) 74 return count
Throw every step in order. Returns count of commands sent.
Raises RuntimeError if the lock sensor is ACTIVE.
Inherited Members
77class RouteManager(Manager[Route]): 78 type_letter = "O" 79 80 def _create(self, system_name: str) -> Route: 81 return Route(system_name)
Per-type bean registry. Subclass and override _create().
Attributes set on the class describe the bean type:
type_letter— single uppercase letter the bean class owns (e.g."T"for Turnout). Must be set by subclass.system_prefix— connection prefix (e.g."L"). Set per-instance because one bean class can have many connection-specific managers.