jmri_core.beans

Concrete bean classes — one module per type.

Each module exports the bean class plus a manager subclass (e.g. Turnout and TurnoutManager). Drivers further subclass these to add hardware-specific behaviour.

 1"""Concrete bean classes — one module per type.
 2
 3Each module exports the bean class plus a manager subclass (e.g.
 4``Turnout`` and ``TurnoutManager``). Drivers further subclass these to add
 5hardware-specific behaviour.
 6"""
 7from jmri_core.beans.block import Block, BlockManager
 8from jmri_core.beans.consist import Consist, ConsistLoco
 9from jmri_core.beans.consist_manager import ConsistManager
10from jmri_core.beans.fast_clock import FastClock, FastClockChange, HardwareClockState
11from jmri_core.beans.light import Light, LightManager
12from jmri_core.beans.memory import Memory, MemoryManager
13from jmri_core.beans.power import PowerManager
14from jmri_core.beans.programmer import Programmer, ProgrammerManager, ProgrammingMode
15from jmri_core.beans.reporter import Reporter, ReporterManager
16from jmri_core.beans.roster_entry import RosterEntry
17from jmri_core.beans.route import Route, RouteManager, RouteStep
18from jmri_core.beans.sensor import Sensor, SensorManager
19from jmri_core.beans.signal_head import SignalHead, SignalHeadManager
20from jmri_core.beans.signal_mast import SignalMast, SignalMastManager
21from jmri_core.beans.throttle import Throttle
22from jmri_core.beans.throttle_manager import ThrottleManager
23from jmri_core.beans.turnout import Turnout, TurnoutManager
24
25__all__ = [
26    "Block",
27    "BlockManager",
28    "Consist",
29    "ConsistLoco",
30    "ConsistManager",
31    "FastClock",
32    "FastClockChange",
33    "HardwareClockState",
34    "Light",
35    "LightManager",
36    "Memory",
37    "MemoryManager",
38    "PowerManager",
39    "Programmer",
40    "ProgrammerManager",
41    "ProgrammingMode",
42    "Reporter",
43    "ReporterManager",
44    "RosterEntry",
45    "Route",
46    "RouteManager",
47    "RouteStep",
48    "Sensor",
49    "SensorManager",
50    "SignalHead",
51    "SignalHeadManager",
52    "SignalMast",
53    "SignalMastManager",
54    "Throttle",
55    "ThrottleManager",
56    "Turnout",
57    "TurnoutManager",
58]
class Block(jmri_core.bean.AbstractNamedBean): 
29class Block(AbstractNamedBean):
30    """Track block — possibly with an attached occupancy sensor."""
31
32    __slots__ = ("_disconnect_sensor", "_sensor", "_state", "_value")
33
34    type_letter = "B"
35
36    def __init__(self, system_name: str, user_name: str | None = None) -> None:
37        super().__init__(system_name, user_name)
38        self._state: BlockState = BlockState.UNKNOWN
39        self._value: Any = None
40        self._sensor: Sensor | None = None
41        self._disconnect_sensor: Any = None
42
43    @property
44    def state(self) -> BlockState:
45        return self._state
46
47    @property
48    def value(self) -> Any:
49        return self._value
50
51    @property
52    def sensor(self) -> Sensor | None:
53        return self._sensor
54
55    async def set_value(self, value: Any) -> None:
56        prev = self._value
57        if prev == value:
58            return
59        self._value = value
60        await self.state_changed.emit(BlockChange("value", prev, value))
61
62    async def _set_state(self, state: BlockState) -> None:
63        prev = self._state
64        if prev == state:
65            return
66        self._state = state
67        await self.state_changed.emit(BlockChange("state", prev, state))
68
69    def attach_sensor(self, sensor: Sensor | None) -> None:
70        """Bind a sensor whose state drives this block's state.
71
72        Pass ``None`` to detach. Detaching disconnects the listener so the
73        sensor reference can be garbage-collected.
74        """
75        if self._disconnect_sensor is not None:
76            self._disconnect_sensor()
77            self._disconnect_sensor = None
78        self._sensor = sensor
79        if sensor is not None:
80            self._disconnect_sensor = sensor.state_changed.connect(self._on_sensor_change)
81            # Seed initial state from the sensor.
82            self._state = _sensor_to_block_state(sensor.known_state)
83
84    async def _on_sensor_change(self, change: SensorStateChange) -> None:
85        await self._set_state(_sensor_to_block_state(change.current))

Track block — possibly with an attached occupancy sensor.

Block(system_name: str, user_name: str | None = None) 
36    def __init__(self, system_name: str, user_name: str | None = None) -> None:
37        super().__init__(system_name, user_name)
38        self._state: BlockState = BlockState.UNKNOWN
39        self._value: Any = None
40        self._sensor: Sensor | None = None
41        self._disconnect_sensor: Any = None
type_letter = 'B'
state: jmri_core.state.BlockState 
43    @property
44    def state(self) -> BlockState:
45        return self._state
value: Any 
47    @property
48    def value(self) -> Any:
49        return self._value
sensor: Sensor | None 
51    @property
52    def sensor(self) -> Sensor | None:
53        return self._sensor
async def set_value(self, value: Any) -> None: 
55    async def set_value(self, value: Any) -> None:
56        prev = self._value
57        if prev == value:
58            return
59        self._value = value
60        await self.state_changed.emit(BlockChange("value", prev, value))
def attach_sensor(self, sensor: Sensor | None) -> None: 
69    def attach_sensor(self, sensor: Sensor | None) -> None:
70        """Bind a sensor whose state drives this block's state.
71
72        Pass ``None`` to detach. Detaching disconnects the listener so the
73        sensor reference can be garbage-collected.
74        """
75        if self._disconnect_sensor is not None:
76            self._disconnect_sensor()
77            self._disconnect_sensor = None
78        self._sensor = sensor
79        if sensor is not None:
80            self._disconnect_sensor = sensor.state_changed.connect(self._on_sensor_change)
81            # Seed initial state from the sensor.
82            self._state = _sensor_to_block_state(sensor.known_state)

Bind a sensor whose state drives this block's state.

Pass None to detach. Detaching disconnects the listener so the sensor reference can be garbage-collected.

Inherited Members
jmri_core.bean.AbstractNamedBean
system_name
user_name
comment
state_changed
properties
display_name
class BlockManager(jmri_core.manager.Manager[jmri_core.beans.block.Block]): 
 96class BlockManager(Manager[Block]):
 97    type_letter = "B"
 98
 99    def _create(self, system_name: str) -> Block:
100        return Block(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.
type_letter = 'B'
Inherited Members
jmri_core.manager.Manager
Manager
system_prefix
added
removed
full_prefix
get_by_system_name
get_by_user_name
get
named_bean_set
provide
register
deregister
to_dict
class Consist(pydantic.main.BaseModel): 
39class Consist(BaseModel):
40    """An ordered group of locomotives.
41
42    The "consist address" identifies the group on the bus. For advanced
43    (CV19) consists this is the active short address shared across locos;
44    for command-station consists it's a slot/handle the command station
45    assigns; for software consists it's the bookkeeping key. All three
46    variants are held to the 1..127 short-address range for consistency
47    with the CV19 wire limit.
48    """
49
50    address: int = Field(ge=1, le=CONSIST_ADDRESS_MAX)
51    kind: ConsistKind = "advanced"
52    locos: list[ConsistLoco] = Field(default_factory=list)
53    comment: str = ""
54
55    def add(self, loco: ConsistLoco) -> None:
56        self.locos.append(loco)
57
58    def remove(self, address: int) -> bool:
59        """Drop the loco at *address*. Returns True if found."""
60        for i, loco in enumerate(self.locos):
61            if loco.address == address:
62                del self.locos[i]
63                return True
64        return False

An ordered group of locomotives.

The "consist address" identifies the group on the bus. For advanced (CV19) consists this is the active short address shared across locos; for command-station consists it's a slot/handle the command station assigns; for software consists it's the bookkeeping key. All three variants are held to the 1..127 short-address range for consistency with the CV19 wire limit.

address: int = PydanticUndefined
kind: Literal['advanced', 'command_station', 'software'] = 'advanced'
locos: list[ConsistLoco] = PydanticUndefined
comment: str = ''
def add(self, loco: ConsistLoco) -> None: 
55    def add(self, loco: ConsistLoco) -> None:
56        self.locos.append(loco)
def remove(self, address: int) -> bool: 
58    def remove(self, address: int) -> bool:
59        """Drop the loco at *address*. Returns True if found."""
60        for i, loco in enumerate(self.locos):
61            if loco.address == address:
62                del self.locos[i]
63                return True
64        return False

Drop the loco at address. Returns True if found.

class ConsistLoco(pydantic.main.BaseModel): 
27class ConsistLoco(BaseModel):
28    """One member of a consist. Individual member locomotives may still
29    have long addresses — only the *consist*'s own address is
30    restricted to short 1..127."""
31
32    address: int = Field(ge=1, le=10239)
33    long_address: bool = False
34    direction_normal: bool = True
35    # Position is the index within the consist; first added = 0.
36    position: int = Field(ge=0)

One member of a consist. Individual member locomotives may still have long addresses — only the consist's own address is restricted to short 1..127.

address: int = PydanticUndefined
long_address: bool = False
direction_normal: bool = True
position: int = PydanticUndefined
class ConsistManager: 
 48class ConsistManager:
 49    """Track a set of consists and fan throttle commands to their members.
 50
 51    The base implementation is the **software consist** manager: it
 52    stores consists in a dict keyed by the group's DCC address and
 53    executes group speed/direction/function commands by looping over
 54    every member's :class:`Throttle`. Drivers that support hardware
 55    consisting subclass this and override :meth:`_after_create` and
 56    :meth:`_before_delete` to install/remove the wire-level group;
 57    they typically leave the fan-out helpers alone since the base's
 58    per-loco commands are still valid even when the CS is grouping.
 59
 60    ``created`` and ``deleted`` are :class:`Signal` fan-outs so the
 61    push side of the JSON server, the WiThrottle broadcast, and the
 62    web UI's live table can react without polling.
 63    """
 64
 65    default_kind: ConsistKind = "software"
 66
 67    def __init__(
 68        self, throttles: ThrottleManager, system_prefix: str,
 69        programmers: ProgrammerManager | None = None,
 70    ) -> None:
 71        if not system_prefix:
 72            raise ValueError("system_prefix must be a non-empty string")
 73        self._throttles = throttles
 74        # Ops-mode programmer manager, used by the ``advanced`` kind to
 75        # write CV19 on each member decoder as they join or leave. May
 76        # be ``None`` on non-hardware setups (tests, InternalConnection);
 77        # advanced consists degrade silently to a no-op in that case
 78        # rather than raising.
 79        self._programmers: ProgrammerManager | None = programmers
 80        self.system_prefix = system_prefix
 81        self._consists: dict[int, Consist] = {}
 82        # Runtime drive state per consist address. Not persisted — the
 83        # consist itself is pure config (membership + wire kind), and
 84        # loco throttles carry the authoritative state. But the UI
 85        # needs a per-consist "last commanded" value to render the
 86        # slider position and highlight the active direction button;
 87        # without it every drive command re-renders the card at the
 88        # hardcoded defaults and the user sees the slider snap back to
 89        # 0 on every drag-release. Reset on delete.
 90        self._speeds: dict[int, float] = {}
 91        self._directions: dict[int, ThrottleDirection] = {}
 92        self.created: Signal[Consist] = Signal()
 93        self.deleted: Signal[Consist] = Signal()
 94        self.updated: Signal[Consist] = Signal()
 95
 96    def rebind_throttles(self, throttles: ThrottleManager) -> None:
 97        """Point the fan-out at a new :class:`ThrottleManager`.
 98
 99        Base :meth:`SystemConnection.__init__` builds the consist manager
100        before drivers get a chance to install their hardware-aware
101        throttle manager in :meth:`open`. Without this hook the consist
102        manager keeps a stale reference to the pre-open placeholder and
103        every fan-out silently drives plain-Python :class:`Throttle`
104        beans that never touch the wire — the consist would look like
105        it's working (bean state advances, signals fire) while the
106        actual locomotive sits still. Drivers call this from ``open()``
107        immediately after replacing ``self.throttles``.
108        """
109        self._throttles = throttles
110
111    def rebind_programmers(
112        self, programmers: ProgrammerManager | None
113    ) -> None:
114        """Point the CV19 write path at a new :class:`ProgrammerManager`.
115
116        Same rationale as :meth:`rebind_throttles` — the driver's
117        hardware-aware programmer manager isn't in place until
118        :meth:`open` runs. Advanced consists need ops-mode access to
119        program CV19 on each member; without a real programmer the
120        writes silently become no-ops.
121        """
122        self._programmers = programmers
123
124    def current_speed(self, address: int) -> float:
125        """Last commanded fan-out speed for this consist, or 0.0."""
126        return self._speeds.get(address, 0.0)
127
128    def current_direction(self, address: int) -> ThrottleDirection:
129        """Last commanded direction for this consist, or FORWARD."""
130        return self._directions.get(address, ThrottleDirection.FORWARD)
131
132    # ---- CRUD ------------------------------------------------------------
133
134    async def create(
135        self,
136        address: int,
137        *,
138        kind: ConsistKind | None = None,
139        comment: str = "",
140    ) -> Consist:
141        """Create a new consist and register it. Idempotent by address.
142
143        Returns the existing consist if one already lives at *address*
144        rather than raising — matches JMRI's tolerance for repeated
145        create calls from restore paths.
146
147        Consist addresses are constrained to 1..127 short by the
148        ``Consist`` model; the caller doesn't need a ``long_address``
149        flag because there is no long-address form of a consist. See
150        :data:`jmri_core.beans.consist.CONSIST_ADDRESS_MAX`.
151        """
152        existing = self._consists.get(address)
153        if existing is not None:
154            return existing
155        consist = Consist(
156            address=address,
157            kind=kind or self.default_kind, comment=comment,
158        )
159        self._consists[address] = consist
160        await self._after_create(consist)
161        await self.created.emit(consist)
162        return consist
163
164    async def delete(self, address: int) -> bool:
165        """Remove the consist. Returns ``True`` if one was found."""
166        consist = self._consists.pop(address, None)
167        if consist is None:
168            return False
169        self._speeds.pop(address, None)
170        self._directions.pop(address, None)
171        await self._before_delete(consist)
172        await self.deleted.emit(consist)
173        return True
174
175    def get(self, address: int) -> Consist | None:
176        return self._consists.get(address)
177
178    def __iter__(self) -> Iterator[Consist]:
179        return iter(self._consists.values())
180
181    def __len__(self) -> int:
182        return len(self._consists)
183
184    def __contains__(self, address: int) -> bool:
185        return address in self._consists
186
187    # ---- membership -----------------------------------------------------
188
189    async def add_member(
190        self,
191        consist_address: int,
192        member_address: int,
193        *,
194        long_address: bool = False,
195        direction_normal: bool = True,
196    ) -> ConsistLoco:
197        """Add *member_address* to the consist at *consist_address*.
198
199        The member is appended at the end. If a loco with the same
200        address already exists in the consist, its direction flag is
201        updated in place and the existing :class:`ConsistLoco` is
202        returned. Fires :attr:`updated` on the consist so subscribers
203        can react to membership changes.
204        """
205        consist = self._consists.get(consist_address)
206        if consist is None:
207            raise KeyError(f"no consist at address {consist_address}")
208        for loco in consist.locos:
209            if loco.address == member_address:
210                loco.direction_normal = direction_normal
211                await self._after_membership_change(consist)
212                if consist.kind == "advanced":
213                    await self._program_cv19(
214                        member_address, long_address=loco.long_address,
215                        consist_address=consist_address,
216                        direction_normal=direction_normal,
217                    )
218                await self.updated.emit(consist)
219                return loco
220        loco = ConsistLoco(
221            address=member_address,
222            long_address=long_address,
223            direction_normal=direction_normal,
224            position=len(consist.locos),
225        )
226        consist.locos.append(loco)
227        await self._after_membership_change(consist)
228        if consist.kind == "advanced":
229            await self._program_cv19(
230                member_address, long_address=long_address,
231                consist_address=consist_address,
232                direction_normal=direction_normal,
233            )
234        await self.updated.emit(consist)
235        return loco
236
237    async def _program_cv19(
238        self,
239        member_address: int,
240        *,
241        long_address: bool,
242        consist_address: int,
243        direction_normal: bool,
244    ) -> None:
245        """Write CV19 on *member_address*'s decoder for an advanced
246        consist.
247
248        Per NMRA S-9.2.2, CV19 holds a 7-bit consist address in bits
249        0..6 and the direction flag in bit 7. Bit 7 set means the loco
250        runs in the reverse direction relative to the consist — i.e.
251        ``direction_normal=False``. A value of 0 in the low 7 bits
252        disables the consist for that decoder.
253        """
254        prog_mgr = self._programmers
255        if prog_mgr is None:
256            _log.debug(
257                "advanced consist %d: no programmer manager, skipping "
258                "CV19 write for member %d",
259                consist_address, member_address,
260            )
261            return
262        prog = prog_mgr.ops_programmer(
263            member_address, long_address=long_address,
264        )
265        if prog is None:
266            _log.debug(
267                "advanced consist %d: driver returned no ops "
268                "programmer for member %d; CV19 not written",
269                consist_address, member_address,
270            )
271            return
272        value = consist_address & 0x7F
273        if not direction_normal:
274            value |= 0x80
275        try:
276            await prog.write_cv(19, value)
277        except Exception as exc:  # noqa: BLE001
278            # POM writes can fail (no ack, decoder not on the track,
279            # RailCom garble). Log and keep going — the consist itself
280            # remains configured; the user can re-add the member after
281            # putting the loco on the rails.
282            _log.warning(
283                "CV19 write failed for consist %d, member %d "
284                "(value=%d): %s",
285                consist_address, member_address, value, exc,
286            )
287
288    async def _clear_cv19(
289        self, member_address: int, *, long_address: bool,
290        consist_address: int,
291    ) -> None:
292        """Reset CV19 to 0 on the ex-member's decoder so it stops
293        answering to the consist address."""
294        prog_mgr = self._programmers
295        if prog_mgr is None:
296            return
297        prog = prog_mgr.ops_programmer(
298            member_address, long_address=long_address,
299        )
300        if prog is None:
301            return
302        try:
303            await prog.write_cv(19, 0)
304        except Exception as exc:  # noqa: BLE001
305            _log.warning(
306                "CV19 clear failed for consist %d, ex-member %d: %s",
307                consist_address, member_address, exc,
308            )
309
310    async def remove_member(
311        self, consist_address: int, member_address: int,
312    ) -> bool:
313        """Drop *member_address* from the consist. Returns True if found.
314
315        Re-numbers surviving members' ``position`` fields so they stay
316        contiguous — the ordering is used for lead-loco / rear-loco
317        semantics in DCC-EX and NCE displays.
318        """
319        consist = self._consists.get(consist_address)
320        if consist is None:
321            return False
322        removed = False
323        removed_long_address = False
324        surviving: list[ConsistLoco] = []
325        for loco in consist.locos:
326            if loco.address == member_address and not removed:
327                removed = True
328                # Capture before the loco leaves — the ops programmer
329                # for the CV19 clear is keyed by the decoder's real
330                # address form (short vs long).
331                removed_long_address = loco.long_address
332                continue
333            surviving.append(loco)
334        if not removed:
335            return False
336        # Renumber positions.
337        for i, loco in enumerate(surviving):
338            loco.position = i
339        consist.locos = surviving
340        await self._after_membership_change(consist)
341        if consist.kind == "advanced":
342            await self._clear_cv19(
343                member_address, long_address=removed_long_address,
344                consist_address=consist_address,
345            )
346        await self.updated.emit(consist)
347        return True
348
349    # ---- fan-out throttle control ---------------------------------------
350
351    async def set_speed(self, consist_address: int, speed: float) -> None:
352        """Fan a speed command to every member throttle.
353
354        Members with ``direction_normal=False`` are treated identically
355        for speed (only direction is inverted). Missing throttles
356        (member never acquired) are silently acquired first so the
357        first speed command on a fresh consist works.
358
359        Records the commanded speed on the manager and emits ``updated``
360        so any subscriber (the /consists page's WebSocket bus in
361        particular) re-renders the card with the slider at the new
362        position instead of snapping back to zero.
363        """
364        consist = self._require(consist_address)
365        for member in consist.locos:
366            throttle = await self._throttles.acquire(
367                member.address, long_address=member.long_address,
368            )
369            await throttle.set_speed(speed)
370        self._speeds[consist_address] = speed
371        await self.updated.emit(consist)
372
373    async def set_direction(
374        self, consist_address: int, direction: ThrottleDirection,
375    ) -> None:
376        """Fan a direction command to every member, honouring per-loco
377        orientation.
378
379        A member with ``direction_normal=False`` is oriented backwards
380        in the consist, so it receives the *opposite* direction to keep
381        the whole train moving the same way — this is the core reason
382        consists exist at the software level.
383        """
384        consist = self._require(consist_address)
385        opposite = (
386            ThrottleDirection.REVERSE
387            if direction is ThrottleDirection.FORWARD
388            else ThrottleDirection.FORWARD
389        )
390        for member in consist.locos:
391            throttle = await self._throttles.acquire(
392                member.address, long_address=member.long_address,
393            )
394            wire = direction if member.direction_normal else opposite
395            await throttle.set_direction(wire)
396        self._directions[consist_address] = direction
397        await self.updated.emit(consist)
398
399    async def set_function(
400        self, consist_address: int, fn: int, on: bool,
401    ) -> None:
402        """Fan a function command to every member.
403
404        JMRI's convention is that all lit functions (F0/headlight, F1/
405        bell, F2/horn) fire on every loco; layout-specific automation
406        that wants function isolation should drive individual throttles
407        directly.
408        """
409        consist = self._require(consist_address)
410        for member in consist.locos:
411            throttle = await self._throttles.acquire(
412                member.address, long_address=member.long_address,
413            )
414            await throttle.set_function(fn, on)
415
416    async def emergency_stop(self, consist_address: int) -> None:
417        """Fan an emergency-stop to every member throttle."""
418        consist = self._require(consist_address)
419        for member in consist.locos:
420            throttle = await self._throttles.acquire(
421                member.address, long_address=member.long_address,
422            )
423            await throttle.emergency_stop()
424        # Reset UI speed so the slider snaps back to 0 after an e-stop
425        # rather than staying at whatever the user last commanded.
426        self._speeds[consist_address] = 0.0
427        await self.updated.emit(consist)
428
429    # ---- driver hooks ---------------------------------------------------
430
431    async def _after_create(self, consist: Consist) -> None:
432        """Called after a new consist is registered. Driver hook.
433
434        Subclasses install the wire-level group here — LocoNet's
435        ``LINK_SLOTS``, NCE's consist-memory write, DCC-EX's ``<C>``.
436        The base impl is a no-op (software consist).
437        """
438        return
439
440    async def _before_delete(self, consist: Consist) -> None:
441        """Called before a consist is dropped from the registry.
442
443        Subclasses tear the wire-level group down here.
444        """
445        return
446
447    async def _after_membership_change(self, consist: Consist) -> None:
448        """Called after ``add_member`` / ``remove_member`` mutates the
449        consist's locos list. Subclasses re-sync the wire-level group.
450        """
451        return
452
453    # ---- internals ------------------------------------------------------
454
455    def _require(self, address: int) -> Consist:
456        consist = self._consists.get(address)
457        if consist is None:
458            raise KeyError(f"no consist at address {address}")
459        return consist

Track a set of consists and fan throttle commands to their members.

The base implementation is the software consist manager: it stores consists in a dict keyed by the group's DCC address and executes group speed/direction/function commands by looping over every member's Throttle. Drivers that support hardware consisting subclass this and override _after_create() and _before_delete() to install/remove the wire-level group; they typically leave the fan-out helpers alone since the base's per-loco commands are still valid even when the CS is grouping.

created and deleted are Signal fan-outs so the push side of the JSON server, the WiThrottle broadcast, and the web UI's live table can react without polling.

ConsistManager( throttles: ThrottleManager, system_prefix: str, programmers: ProgrammerManager | None = None) 
67    def __init__(
68        self, throttles: ThrottleManager, system_prefix: str,
69        programmers: ProgrammerManager | None = None,
70    ) -> None:
71        if not system_prefix:
72            raise ValueError("system_prefix must be a non-empty string")
73        self._throttles = throttles
74        # Ops-mode programmer manager, used by the ``advanced`` kind to
75        # write CV19 on each member decoder as they join or leave. May
76        # be ``None`` on non-hardware setups (tests, InternalConnection);
77        # advanced consists degrade silently to a no-op in that case
78        # rather than raising.
79        self._programmers: ProgrammerManager | None = programmers
80        self.system_prefix = system_prefix
81        self._consists: dict[int, Consist] = {}
82        # Runtime drive state per consist address. Not persisted — the
83        # consist itself is pure config (membership + wire kind), and
84        # loco throttles carry the authoritative state. But the UI
85        # needs a per-consist "last commanded" value to render the
86        # slider position and highlight the active direction button;
87        # without it every drive command re-renders the card at the
88        # hardcoded defaults and the user sees the slider snap back to
89        # 0 on every drag-release. Reset on delete.
90        self._speeds: dict[int, float] = {}
91        self._directions: dict[int, ThrottleDirection] = {}
92        self.created: Signal[Consist] = Signal()
93        self.deleted: Signal[Consist] = Signal()
94        self.updated: Signal[Consist] = Signal()
default_kind: Literal['advanced', 'command_station', 'software'] = 'software'
system_prefix
def rebind_throttles( self, throttles: ThrottleManager) -> None: 
 96    def rebind_throttles(self, throttles: ThrottleManager) -> None:
 97        """Point the fan-out at a new :class:`ThrottleManager`.
 98
 99        Base :meth:`SystemConnection.__init__` builds the consist manager
100        before drivers get a chance to install their hardware-aware
101        throttle manager in :meth:`open`. Without this hook the consist
102        manager keeps a stale reference to the pre-open placeholder and
103        every fan-out silently drives plain-Python :class:`Throttle`
104        beans that never touch the wire — the consist would look like
105        it's working (bean state advances, signals fire) while the
106        actual locomotive sits still. Drivers call this from ``open()``
107        immediately after replacing ``self.throttles``.
108        """
109        self._throttles = throttles

Point the fan-out at a new ThrottleManager.

Base SystemConnection.__init__() builds the consist manager before drivers get a chance to install their hardware-aware throttle manager in open(). Without this hook the consist manager keeps a stale reference to the pre-open placeholder and every fan-out silently drives plain-Python Throttle beans that never touch the wire — the consist would look like it's working (bean state advances, signals fire) while the actual locomotive sits still. Drivers call this from open() immediately after replacing self.throttles.

def rebind_programmers( self, programmers: ProgrammerManager | None) -> None: 
111    def rebind_programmers(
112        self, programmers: ProgrammerManager | None
113    ) -> None:
114        """Point the CV19 write path at a new :class:`ProgrammerManager`.
115
116        Same rationale as :meth:`rebind_throttles` — the driver's
117        hardware-aware programmer manager isn't in place until
118        :meth:`open` runs. Advanced consists need ops-mode access to
119        program CV19 on each member; without a real programmer the
120        writes silently become no-ops.
121        """
122        self._programmers = programmers

Point the CV19 write path at a new ProgrammerManager.

Same rationale as rebind_throttles() — the driver's hardware-aware programmer manager isn't in place until open() runs. Advanced consists need ops-mode access to program CV19 on each member; without a real programmer the writes silently become no-ops.

def current_speed(self, address: int) -> float: 
124    def current_speed(self, address: int) -> float:
125        """Last commanded fan-out speed for this consist, or 0.0."""
126        return self._speeds.get(address, 0.0)

Last commanded fan-out speed for this consist, or 0.0.

def current_direction(self, address: int) -> jmri_core.state.ThrottleDirection: 
128    def current_direction(self, address: int) -> ThrottleDirection:
129        """Last commanded direction for this consist, or FORWARD."""
130        return self._directions.get(address, ThrottleDirection.FORWARD)

Last commanded direction for this consist, or FORWARD.

async def create( self, address: int, *, kind: Optional[Literal['advanced', 'command_station', 'software']] = None, comment: str = '') -> Consist: 
134    async def create(
135        self,
136        address: int,
137        *,
138        kind: ConsistKind | None = None,
139        comment: str = "",
140    ) -> Consist:
141        """Create a new consist and register it. Idempotent by address.
142
143        Returns the existing consist if one already lives at *address*
144        rather than raising — matches JMRI's tolerance for repeated
145        create calls from restore paths.
146
147        Consist addresses are constrained to 1..127 short by the
148        ``Consist`` model; the caller doesn't need a ``long_address``
149        flag because there is no long-address form of a consist. See
150        :data:`jmri_core.beans.consist.CONSIST_ADDRESS_MAX`.
151        """
152        existing = self._consists.get(address)
153        if existing is not None:
154            return existing
155        consist = Consist(
156            address=address,
157            kind=kind or self.default_kind, comment=comment,
158        )
159        self._consists[address] = consist
160        await self._after_create(consist)
161        await self.created.emit(consist)
162        return consist

Create a new consist and register it. Idempotent by address.

Returns the existing consist if one already lives at address rather than raising — matches JMRI's tolerance for repeated create calls from restore paths.

Consist addresses are constrained to 1..127 short by the Consist model; the caller doesn't need a long_address flag because there is no long-address form of a consist. See jmri_core.beans.consist.CONSIST_ADDRESS_MAX.

async def delete(self, address: int) -> bool: 
164    async def delete(self, address: int) -> bool:
165        """Remove the consist. Returns ``True`` if one was found."""
166        consist = self._consists.pop(address, None)
167        if consist is None:
168            return False
169        self._speeds.pop(address, None)
170        self._directions.pop(address, None)
171        await self._before_delete(consist)
172        await self.deleted.emit(consist)
173        return True

Remove the consist. Returns True if one was found.

def get(self, address: int) -> Consist | None: 
175    def get(self, address: int) -> Consist | None:
176        return self._consists.get(address)
async def add_member( self, consist_address: int, member_address: int, *, long_address: bool = False, direction_normal: bool = True) -> ConsistLoco: 
189    async def add_member(
190        self,
191        consist_address: int,
192        member_address: int,
193        *,
194        long_address: bool = False,
195        direction_normal: bool = True,
196    ) -> ConsistLoco:
197        """Add *member_address* to the consist at *consist_address*.
198
199        The member is appended at the end. If a loco with the same
200        address already exists in the consist, its direction flag is
201        updated in place and the existing :class:`ConsistLoco` is
202        returned. Fires :attr:`updated` on the consist so subscribers
203        can react to membership changes.
204        """
205        consist = self._consists.get(consist_address)
206        if consist is None:
207            raise KeyError(f"no consist at address {consist_address}")
208        for loco in consist.locos:
209            if loco.address == member_address:
210                loco.direction_normal = direction_normal
211                await self._after_membership_change(consist)
212                if consist.kind == "advanced":
213                    await self._program_cv19(
214                        member_address, long_address=loco.long_address,
215                        consist_address=consist_address,
216                        direction_normal=direction_normal,
217                    )
218                await self.updated.emit(consist)
219                return loco
220        loco = ConsistLoco(
221            address=member_address,
222            long_address=long_address,
223            direction_normal=direction_normal,
224            position=len(consist.locos),
225        )
226        consist.locos.append(loco)
227        await self._after_membership_change(consist)
228        if consist.kind == "advanced":
229            await self._program_cv19(
230                member_address, long_address=long_address,
231                consist_address=consist_address,
232                direction_normal=direction_normal,
233            )
234        await self.updated.emit(consist)
235        return loco

Add member_address to the consist at consist_address.

The member is appended at the end. If a loco with the same address already exists in the consist, its direction flag is updated in place and the existing ConsistLoco is returned. Fires updated on the consist so subscribers can react to membership changes.

async def remove_member(self, consist_address: int, member_address: int) -> bool: 
310    async def remove_member(
311        self, consist_address: int, member_address: int,
312    ) -> bool:
313        """Drop *member_address* from the consist. Returns True if found.
314
315        Re-numbers surviving members' ``position`` fields so they stay
316        contiguous — the ordering is used for lead-loco / rear-loco
317        semantics in DCC-EX and NCE displays.
318        """
319        consist = self._consists.get(consist_address)
320        if consist is None:
321            return False
322        removed = False
323        removed_long_address = False
324        surviving: list[ConsistLoco] = []
325        for loco in consist.locos:
326            if loco.address == member_address and not removed:
327                removed = True
328                # Capture before the loco leaves — the ops programmer
329                # for the CV19 clear is keyed by the decoder's real
330                # address form (short vs long).
331                removed_long_address = loco.long_address
332                continue
333            surviving.append(loco)
334        if not removed:
335            return False
336        # Renumber positions.
337        for i, loco in enumerate(surviving):
338            loco.position = i
339        consist.locos = surviving
340        await self._after_membership_change(consist)
341        if consist.kind == "advanced":
342            await self._clear_cv19(
343                member_address, long_address=removed_long_address,
344                consist_address=consist_address,
345            )
346        await self.updated.emit(consist)
347        return True

Drop member_address from the consist. Returns True if found.

Re-numbers surviving members' position fields so they stay contiguous — the ordering is used for lead-loco / rear-loco semantics in DCC-EX and NCE displays.

async def set_speed(self, consist_address: int, speed: float) -> None: 
351    async def set_speed(self, consist_address: int, speed: float) -> None:
352        """Fan a speed command to every member throttle.
353
354        Members with ``direction_normal=False`` are treated identically
355        for speed (only direction is inverted). Missing throttles
356        (member never acquired) are silently acquired first so the
357        first speed command on a fresh consist works.
358
359        Records the commanded speed on the manager and emits ``updated``
360        so any subscriber (the /consists page's WebSocket bus in
361        particular) re-renders the card with the slider at the new
362        position instead of snapping back to zero.
363        """
364        consist = self._require(consist_address)
365        for member in consist.locos:
366            throttle = await self._throttles.acquire(
367                member.address, long_address=member.long_address,
368            )
369            await throttle.set_speed(speed)
370        self._speeds[consist_address] = speed
371        await self.updated.emit(consist)

Fan a speed command to every member throttle.

Members with direction_normal=False are treated identically for speed (only direction is inverted). Missing throttles (member never acquired) are silently acquired first so the first speed command on a fresh consist works.

Records the commanded speed on the manager and emits updated so any subscriber (the /consists page's WebSocket bus in particular) re-renders the card with the slider at the new position instead of snapping back to zero.

async def set_direction( self, consist_address: int, direction: jmri_core.state.ThrottleDirection) -> None: 
373    async def set_direction(
374        self, consist_address: int, direction: ThrottleDirection,
375    ) -> None:
376        """Fan a direction command to every member, honouring per-loco
377        orientation.
378
379        A member with ``direction_normal=False`` is oriented backwards
380        in the consist, so it receives the *opposite* direction to keep
381        the whole train moving the same way — this is the core reason
382        consists exist at the software level.
383        """
384        consist = self._require(consist_address)
385        opposite = (
386            ThrottleDirection.REVERSE
387            if direction is ThrottleDirection.FORWARD
388            else ThrottleDirection.FORWARD
389        )
390        for member in consist.locos:
391            throttle = await self._throttles.acquire(
392                member.address, long_address=member.long_address,
393            )
394            wire = direction if member.direction_normal else opposite
395            await throttle.set_direction(wire)
396        self._directions[consist_address] = direction
397        await self.updated.emit(consist)

Fan a direction command to every member, honouring per-loco orientation.

A member with direction_normal=False is oriented backwards in the consist, so it receives the opposite direction to keep the whole train moving the same way — this is the core reason consists exist at the software level.

async def set_function(self, consist_address: int, fn: int, on: bool) -> None: 
399    async def set_function(
400        self, consist_address: int, fn: int, on: bool,
401    ) -> None:
402        """Fan a function command to every member.
403
404        JMRI's convention is that all lit functions (F0/headlight, F1/
405        bell, F2/horn) fire on every loco; layout-specific automation
406        that wants function isolation should drive individual throttles
407        directly.
408        """
409        consist = self._require(consist_address)
410        for member in consist.locos:
411            throttle = await self._throttles.acquire(
412                member.address, long_address=member.long_address,
413            )
414            await throttle.set_function(fn, on)

Fan a function command to every member.

JMRI's convention is that all lit functions (F0/headlight, F1/ bell, F2/horn) fire on every loco; layout-specific automation that wants function isolation should drive individual throttles directly.

async def emergency_stop(self, consist_address: int) -> None: 
416    async def emergency_stop(self, consist_address: int) -> None:
417        """Fan an emergency-stop to every member throttle."""
418        consist = self._require(consist_address)
419        for member in consist.locos:
420            throttle = await self._throttles.acquire(
421                member.address, long_address=member.long_address,
422            )
423            await throttle.emergency_stop()
424        # Reset UI speed so the slider snaps back to 0 after an e-stop
425        # rather than staying at whatever the user last commanded.
426        self._speeds[consist_address] = 0.0
427        await self.updated.emit(consist)

Fan an emergency-stop to every member throttle.

@dataclass
class FastClock: 
 73@dataclass
 74class FastClock:
 75    """Software fast clock.
 76
 77    Default state: midnight, stopped, 1× rate. Real-world time source
 78    is ``time.monotonic`` so the bean is unaffected by wall-clock
 79    adjustments (NTP step, manual changes, DST).
 80    """
 81
 82    rate: float = 1.0
 83    running: bool = False
 84    _base_seconds: float = 0.0       # displayed fast-clock time at the anchor
 85    _real_anchor: float = field(default_factory=time.monotonic)
 86    state_changed: Signal[FastClockChange] = field(default_factory=Signal)
 87
 88    # ----- read --------------------------------------------------------
 89
 90    def time_seconds(self, *, now: float | None = None) -> float:
 91        """Current displayed time, in seconds since midnight (0..86399)."""
 92        if not self.running:
 93            return self._base_seconds % SECONDS_PER_DAY
 94        if now is None:
 95            now = time.monotonic()
 96        elapsed = (now - self._real_anchor) * self.rate
 97        return (self._base_seconds + elapsed) % SECONDS_PER_DAY
 98
 99    def time_hhmm(self, *, now: float | None = None) -> tuple[int, int]:
100        """Convenience: ``(hours, minutes)`` ready for display."""
101        secs = int(self.time_seconds(now=now))
102        return (secs // SECONDS_PER_HOUR, (secs % SECONDS_PER_HOUR) // SECONDS_PER_MINUTE)
103
104    # ----- mutate ------------------------------------------------------
105
106    async def set_time(self, hours: int, minutes: int) -> None:
107        """Set the displayed time. Does not affect the running flag."""
108        if not 0 <= hours <= 23:
109            raise ValueError(f"hours must be 0..23, got {hours}")
110        if not 0 <= minutes <= 59:
111            raise ValueError(f"minutes must be 0..59, got {minutes}")
112        self._base_seconds = hours * SECONDS_PER_HOUR + minutes * SECONDS_PER_MINUTE
113        self._real_anchor = time.monotonic()
114        await self._notify()
115
116    async def set_rate(self, rate: float) -> None:
117        """Change the speedup multiplier without losing the displayed time."""
118        if not MIN_RATE <= rate <= MAX_RATE:
119            raise ValueError(f"rate must be {MIN_RATE}..{MAX_RATE}, got {rate}")
120        # Capture current displayed time first so the change doesn't jump.
121        current = self.time_seconds()
122        self._base_seconds = current
123        self._real_anchor = time.monotonic()
124        self.rate = rate
125        await self._notify()
126
127    async def start(self) -> None:
128        """Begin advancing the clock. Idempotent."""
129        if self.running:
130            return
131        # Keep the displayed time stable across the start.
132        self._real_anchor = time.monotonic()
133        self.running = True
134        await self._notify()
135
136    async def stop(self) -> None:
137        """Hold the clock at the current displayed time. Idempotent."""
138        if not self.running:
139            return
140        self._base_seconds = self.time_seconds()
141        self.running = False
142        await self._notify()
143
144    # ----- internal ----------------------------------------------------
145
146    async def _notify(self) -> None:
147        await self.state_changed.emit(FastClockChange(
148            rate=self.rate,
149            running=self.running,
150            seconds_since_midnight=self.time_seconds(),
151        ))

Software fast clock.

Default state: midnight, stopped, 1× rate. Real-world time source is time.monotonic so the bean is unaffected by wall-clock adjustments (NTP step, manual changes, DST).

FastClock( rate: float = 1.0, running: bool = False, _base_seconds: float = 0.0, _real_anchor: float = <factory>, state_changed: jmri_core.Signal[FastClockChange] = <factory>)
rate: float = 1.0
running: bool = False
def time_seconds(self, *, now: float | None = None) -> float: 
90    def time_seconds(self, *, now: float | None = None) -> float:
91        """Current displayed time, in seconds since midnight (0..86399)."""
92        if not self.running:
93            return self._base_seconds % SECONDS_PER_DAY
94        if now is None:
95            now = time.monotonic()
96        elapsed = (now - self._real_anchor) * self.rate
97        return (self._base_seconds + elapsed) % SECONDS_PER_DAY

Current displayed time, in seconds since midnight (0..86399).

def time_hhmm(self, *, now: float | None = None) -> tuple[int, int]: 
 99    def time_hhmm(self, *, now: float | None = None) -> tuple[int, int]:
100        """Convenience: ``(hours, minutes)`` ready for display."""
101        secs = int(self.time_seconds(now=now))
102        return (secs // SECONDS_PER_HOUR, (secs % SECONDS_PER_HOUR) // SECONDS_PER_MINUTE)

Convenience: (hours, minutes) ready for display.

async def set_time(self, hours: int, minutes: int) -> None: 
106    async def set_time(self, hours: int, minutes: int) -> None:
107        """Set the displayed time. Does not affect the running flag."""
108        if not 0 <= hours <= 23:
109            raise ValueError(f"hours must be 0..23, got {hours}")
110        if not 0 <= minutes <= 59:
111            raise ValueError(f"minutes must be 0..59, got {minutes}")
112        self._base_seconds = hours * SECONDS_PER_HOUR + minutes * SECONDS_PER_MINUTE
113        self._real_anchor = time.monotonic()
114        await self._notify()

Set the displayed time. Does not affect the running flag.

async def set_rate(self, rate: float) -> None: 
116    async def set_rate(self, rate: float) -> None:
117        """Change the speedup multiplier without losing the displayed time."""
118        if not MIN_RATE <= rate <= MAX_RATE:
119            raise ValueError(f"rate must be {MIN_RATE}..{MAX_RATE}, got {rate}")
120        # Capture current displayed time first so the change doesn't jump.
121        current = self.time_seconds()
122        self._base_seconds = current
123        self._real_anchor = time.monotonic()
124        self.rate = rate
125        await self._notify()

Change the speedup multiplier without losing the displayed time.

async def start(self) -> None: 
127    async def start(self) -> None:
128        """Begin advancing the clock. Idempotent."""
129        if self.running:
130            return
131        # Keep the displayed time stable across the start.
132        self._real_anchor = time.monotonic()
133        self.running = True
134        await self._notify()

Begin advancing the clock. Idempotent.

async def stop(self) -> None: 
136    async def stop(self) -> None:
137        """Hold the clock at the current displayed time. Idempotent."""
138        if not self.running:
139            return
140        self._base_seconds = self.time_seconds()
141        self.running = False
142        await self._notify()

Hold the clock at the current displayed time. Idempotent.

@dataclass
class FastClockChange: 
39@dataclass
40class FastClockChange:
41    """Emitted whenever the fast clock is mutated. Listeners can poll
42    :meth:`FastClock.time_seconds` directly if they need the current value."""
43
44    rate: float
45    running: bool
46    seconds_since_midnight: float

Emitted whenever the fast clock is mutated. Listeners can poll FastClock.time_seconds() directly if they need the current value.

FastClockChange(rate: float, running: bool, seconds_since_midnight: float)
rate: float
running: bool
seconds_since_midnight: float
@dataclass(frozen=True, slots=True)
class HardwareClockState: 
49@dataclass(frozen=True, slots=True)
50class HardwareClockState:
51    """Snapshot of a command-station-driven fast clock.
52
53    Drivers whose command station has an authoritative fast clock (NCE,
54    LocoNet's LNCP FastClock, etc.) return this from an async
55    ``read_hardware_clock()`` method; the JSON ``TimeService`` prefers
56    the hardware value over the software :class:`FastClock` bean when
57    the connection exposes one.
58
59    * *hours* / *minutes* / *seconds* — the displayed 24-hour time
60    * *rate* — the CS's fast-clock ratio (typically integer)
61    * *running* — whether the CS clock is advancing
62    * *twenty_four_hour* — the CS's display mode (12h with AM/PM vs 24h)
63    """
64
65    hours: int
66    minutes: int
67    seconds: int
68    rate: float
69    running: bool
70    twenty_four_hour: bool

Snapshot of a command-station-driven fast clock.

Drivers whose command station has an authoritative fast clock (NCE, LocoNet's LNCP FastClock, etc.) return this from an async read_hardware_clock() method; the JSON TimeService prefers the hardware value over the software FastClock bean when the connection exposes one.

  • hours / minutes / seconds — the displayed 24-hour time
  • rate — the CS's fast-clock ratio (typically integer)
  • running — whether the CS clock is advancing
  • twenty_four_hour — the CS's display mode (12h with AM/PM vs 24h)
HardwareClockState( hours: int, minutes: int, seconds: int, rate: float, running: bool, twenty_four_hour: bool)
hours: int
minutes: int
seconds: int
rate: float
running: bool
twenty_four_hour: bool
class Light(jmri_core.bean.AbstractNamedBean): 
24class Light(AbstractNamedBean):
25    """Layout light. Supports binary or dimmable control."""
26
27    __slots__ = ("_intensity", "_state", "supports_dimming")
28
29    type_letter = "L"
30
31    def __init__(self, system_name: str, user_name: str | None = None) -> None:
32        super().__init__(system_name, user_name)
33        self._state: LightState = LightState.UNKNOWN
34        self._intensity: float = 0.0
35        self.supports_dimming: bool = False
36
37    @property
38    def state(self) -> LightState:
39        return self._state
40
41    @property
42    def intensity(self) -> float:
43        return self._intensity
44
45    async def set_state(self, state: LightState) -> None:
46        prev = self._state
47        if state == LightState.ON:
48            self._intensity = 1.0
49        elif state == LightState.OFF:
50            self._intensity = 0.0
51        self._state = state
52        if prev != state:
53            await self.state_changed.emit(LightChange(prev, state, self._intensity))
54
55    async def set_intensity(self, intensity: float) -> None:
56        if not 0.0 <= intensity <= 1.0:
57            raise ValueError(f"intensity must be in [0,1], got {intensity}")
58        if not self.supports_dimming and intensity not in (0.0, 1.0):
59            raise ValueError("this light does not support dimming")
60        prev = self._state
61        self._intensity = intensity
62        if intensity == 0.0:
63            self._state = LightState.OFF
64        elif intensity == 1.0:
65            self._state = LightState.ON
66        else:
67            self._state = LightState.INTERMEDIATE
68        await self.state_changed.emit(LightChange(prev, self._state, intensity))

Layout light. Supports binary or dimmable control.

Light(system_name: str, user_name: str | None = None) 
31    def __init__(self, system_name: str, user_name: str | None = None) -> None:
32        super().__init__(system_name, user_name)
33        self._state: LightState = LightState.UNKNOWN
34        self._intensity: float = 0.0
35        self.supports_dimming: bool = False
type_letter = 'L'
supports_dimming: bool
state: jmri_core.state.LightState 
37    @property
38    def state(self) -> LightState:
39        return self._state
intensity: float 
41    @property
42    def intensity(self) -> float:
43        return self._intensity
async def set_state(self, state: jmri_core.state.LightState) -> None: 
45    async def set_state(self, state: LightState) -> None:
46        prev = self._state
47        if state == LightState.ON:
48            self._intensity = 1.0
49        elif state == LightState.OFF:
50            self._intensity = 0.0
51        self._state = state
52        if prev != state:
53            await self.state_changed.emit(LightChange(prev, state, self._intensity))
async def set_intensity(self, intensity: float) -> None: 
55    async def set_intensity(self, intensity: float) -> None:
56        if not 0.0 <= intensity <= 1.0:
57            raise ValueError(f"intensity must be in [0,1], got {intensity}")
58        if not self.supports_dimming and intensity not in (0.0, 1.0):
59            raise ValueError("this light does not support dimming")
60        prev = self._state
61        self._intensity = intensity
62        if intensity == 0.0:
63            self._state = LightState.OFF
64        elif intensity == 1.0:
65            self._state = LightState.ON
66        else:
67            self._state = LightState.INTERMEDIATE
68        await self.state_changed.emit(LightChange(prev, self._state, intensity))
Inherited Members
jmri_core.bean.AbstractNamedBean
system_name
user_name
comment
state_changed
properties
display_name
class LightManager(jmri_core.manager.Manager[jmri_core.beans.light.Light]): 
71class LightManager(Manager[Light]):
72    type_letter = "L"
73
74    def _create(self, system_name: str) -> Light:
75        return Light(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.
type_letter = 'L'
Inherited Members
jmri_core.manager.Manager
Manager
system_prefix
added
removed
full_prefix
get_by_system_name
get_by_user_name
get
named_bean_set
provide
register
deregister
to_dict
class Memory(jmri_core.bean.AbstractNamedBean): 
23class Memory(AbstractNamedBean):
24    """A named slot holding any value. ``None`` means unset."""
25
26    __slots__ = ("_value",)
27
28    type_letter = "M"
29
30    def __init__(self, system_name: str, user_name: str | None = None) -> None:
31        super().__init__(system_name, user_name)
32        self._value: Any = None
33
34    @property
35    def value(self) -> Any:
36        return self._value
37
38    async def set_value(self, value: Any) -> None:
39        prev = self._value
40        if prev == value:
41            return
42        self._value = value
43        await self.state_changed.emit(MemoryChange(prev, value))

A named slot holding any value. None means unset.

Memory(system_name: str, user_name: str | None = None) 
30    def __init__(self, system_name: str, user_name: str | None = None) -> None:
31        super().__init__(system_name, user_name)
32        self._value: Any = None
type_letter = 'M'
value: Any 
34    @property
35    def value(self) -> Any:
36        return self._value
async def set_value(self, value: Any) -> None: 
38    async def set_value(self, value: Any) -> None:
39        prev = self._value
40        if prev == value:
41            return
42        self._value = value
43        await self.state_changed.emit(MemoryChange(prev, value))
Inherited Members
jmri_core.bean.AbstractNamedBean
system_name
user_name
comment
state_changed
properties
display_name
class MemoryManager(jmri_core.manager.Manager[jmri_core.beans.memory.Memory]): 
46class MemoryManager(Manager[Memory]):
47    type_letter = "M"
48
49    def _create(self, system_name: str) -> Memory:
50        return Memory(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.
type_letter = 'M'
Inherited Members
jmri_core.manager.Manager
Manager
system_prefix
added
removed
full_prefix
get_by_system_name
get_by_user_name
get
named_bean_set
provide
register
deregister
to_dict
class PowerManager: 
22class PowerManager:
23    """Track power state for one system connection.
24
25    Not a NamedBean — it has no system name and there is exactly one per
26    connection. Drivers override :meth:`set_power` to send the hardware
27    command and call :meth:`_set_state` when the layout reports a change.
28    """
29
30    def __init__(self) -> None:
31        self._state: PowerState = PowerState.UNKNOWN
32        self.state_changed: Signal[PowerChange] = Signal()
33
34    @property
35    def state(self) -> PowerState:
36        return self._state
37
38    async def set_power(self, state: PowerState) -> None:
39        """Command the layout to *state*. Base impl just updates in-memory."""
40        await self._set_state(state)
41
42    async def _set_state(self, state: PowerState) -> None:
43        prev = self._state
44        if prev == state:
45            return
46        self._state = state
47        await self.state_changed.emit(PowerChange(prev, state))

Track power state for one system connection.

Not a NamedBean — it has no system name and there is exactly one per connection. Drivers override set_power() to send the hardware command and call _set_state() when the layout reports a change.

state: jmri_core.state.PowerState 
34    @property
35    def state(self) -> PowerState:
36        return self._state
async def set_power(self, state: jmri_core.state.PowerState) -> None: 
38    async def set_power(self, state: PowerState) -> None:
39        """Command the layout to *state*. Base impl just updates in-memory."""
40        await self._set_state(state)

Command the layout to state. Base impl just updates in-memory.

class Programmer(abc.ABC): 
35class Programmer(ABC):
36    """Reads and writes one DCC decoder's CVs.
37
38    Subclasses implement the two abstract methods. They may raise
39    :class:`asyncio.TimeoutError` if the CS doesn't respond in time, or
40    any driver-specific exception on programming errors (no acknowledge
41    pulse, decoder refused write, etc.).
42    """
43
44    @property
45    @abstractmethod
46    def mode(self) -> ProgrammingMode:
47        """Which mode this programmer instance uses."""
48
49    @abstractmethod
50    async def read_cv(self, cv: int) -> int:
51        """Read CV *cv*, returns 0..255."""
52
53    @abstractmethod
54    async def write_cv(self, cv: int, value: int) -> None:
55        """Write *value* to CV *cv*."""

Reads and writes one DCC decoder's CVs.

Subclasses implement the two abstract methods. They may raise asyncio.TimeoutError if the CS doesn't respond in time, or any driver-specific exception on programming errors (no acknowledge pulse, decoder refused write, etc.).

mode: ProgrammingMode 
44    @property
45    @abstractmethod
46    def mode(self) -> ProgrammingMode:
47        """Which mode this programmer instance uses."""

Which mode this programmer instance uses.

@abstractmethod
async def read_cv(self, cv: int) -> int: 
49    @abstractmethod
50    async def read_cv(self, cv: int) -> int:
51        """Read CV *cv*, returns 0..255."""

Read CV cv, returns 0..255.

@abstractmethod
async def write_cv(self, cv: int, value: int) -> None: 
53    @abstractmethod
54    async def write_cv(self, cv: int, value: int) -> None:
55        """Write *value* to CV *cv*."""

Write value to CV cv.

class ProgrammerManager: 
58class ProgrammerManager:
59    """Hands out programmer instances.
60
61    * :meth:`global_programmer` returns the service-mode programmer (one
62      per connection — the programming track is a singleton).
63    * :meth:`ops_programmer(address, long_address)` returns an ops-mode
64      programmer scoped to the given DCC address.
65
66    The base class returns ``None`` from both; drivers override.
67    """
68
69    def global_programmer(self) -> Programmer | None:
70        return None
71
72    def ops_programmer(
73        self, address: int, *, long_address: bool = False
74    ) -> Programmer | None:
75        return None

Hands out programmer instances.

  • global_programmer() returns the service-mode programmer (one per connection — the programming track is a singleton).
  • ops_programmer(address, long_address)() returns an ops-mode programmer scoped to the given DCC address.

The base class returns None from both; drivers override.

def global_programmer(self) -> Programmer | None: 
69    def global_programmer(self) -> Programmer | None:
70        return None
def ops_programmer( self, address: int, *, long_address: bool = False) -> Programmer | None: 
72    def ops_programmer(
73        self, address: int, *, long_address: bool = False
74    ) -> Programmer | None:
75        return None
class ProgrammingMode(enum.StrEnum): 
26class ProgrammingMode(StrEnum):
27    """How the programmer talks to the decoder."""
28
29    SERVICE_DIRECT = "service_direct"
30    SERVICE_PAGED  = "service_paged"
31    SERVICE_REGISTER = "service_register"
32    OPS_MODE = "ops_mode"

How the programmer talks to the decoder.

SERVICE_DIRECT = <ProgrammingMode.SERVICE_DIRECT: 'service_direct'>
SERVICE_PAGED = <ProgrammingMode.SERVICE_PAGED: 'service_paged'>
SERVICE_REGISTER = <ProgrammingMode.SERVICE_REGISTER: 'service_register'>
OPS_MODE = <ProgrammingMode.OPS_MODE: 'ops_mode'>
class Reporter(jmri_core.bean.AbstractNamedBean): 
24class Reporter(AbstractNamedBean):
25    """Identification-data input. ``current_report`` is None when no read yet."""
26
27    __slots__ = ("_current_report", "_last_report")
28
29    type_letter = "R"
30
31    def __init__(self, system_name: str, user_name: str | None = None) -> None:
32        super().__init__(system_name, user_name)
33        self._current_report: Any = None
34        self._last_report: Any = None
35
36    @property
37    def current_report(self) -> Any:
38        return self._current_report
39
40    @property
41    def last_report(self) -> Any:
42        return self._last_report
43
44    async def set_report(self, value: Any) -> None:
45        prev = self._current_report
46        self._last_report = prev
47        self._current_report = value
48        await self.state_changed.emit(ReporterChange(prev, value))

Identification-data input. current_report is None when no read yet.

Reporter(system_name: str, user_name: str | None = None) 
31    def __init__(self, system_name: str, user_name: str | None = None) -> None:
32        super().__init__(system_name, user_name)
33        self._current_report: Any = None
34        self._last_report: Any = None
type_letter = 'R'
current_report: Any 
36    @property
37    def current_report(self) -> Any:
38        return self._current_report
last_report: Any 
40    @property
41    def last_report(self) -> Any:
42        return self._last_report
async def set_report(self, value: Any) -> None: 
44    async def set_report(self, value: Any) -> None:
45        prev = self._current_report
46        self._last_report = prev
47        self._current_report = value
48        await self.state_changed.emit(ReporterChange(prev, value))
Inherited Members
jmri_core.bean.AbstractNamedBean
system_name
user_name
comment
state_changed
properties
display_name
class ReporterManager(jmri_core.manager.Manager[jmri_core.beans.reporter.Reporter]): 
51class ReporterManager(Manager[Reporter]):
52    type_letter = "R"
53
54    def _create(self, system_name: str) -> Reporter:
55        return Reporter(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.
type_letter = 'R'
Inherited Members
jmri_core.manager.Manager
Manager
system_prefix
added
removed
full_prefix
get_by_system_name
get_by_user_name
get
named_bean_set
provide
register
deregister
to_dict
class RosterEntry(pydantic.main.BaseModel): 
32class RosterEntry(BaseModel):
33    """A roster entry — what JMRI's ``RosterEntry`` carries."""
34
35    model_config = ConfigDict(populate_by_name=True)
36
37    # Identity
38    id: str
39    file_name: str = ""
40
41    # DCC
42    dcc_address: str = "3"
43    long_address: bool = False
44    protocol: str = "dcc_short"
45
46    # Decoder identity
47    decoder_family: str = ""
48    decoder_model: str = ""
49    decoder_comment: str = ""
50
51    # Display
52    road_name: str = ""
53    road_number: str = ""
54    mfg: str = ""
55    owner: str = ""
56    model: str = ""
57    comment: str = ""
58
59    # Operations
60    max_speed_pct: int = Field(default=100, ge=0, le=100)
61
62    # Imagery
63    image_file_path: str | None = None
64    icon_file_path: str | None = None
65    url: str | None = None
66    date_modified: str | None = None
67
68    # Function & sound labels
69    function_labels: list[FunctionLabel] = Field(default_factory=list)
70    sound_labels: list[SoundLabel] = Field(default_factory=list)
71
72    # Free-form attributes (JMRI <attributepairs>)
73    attributes: dict[str, str] = Field(default_factory=dict)
74
75    # Unmodeled child XML — for elements like ``<values>`` (CV state),
76    # ``<dateUpdated>``, custom developer/manufacturer fields, etc. that
77    # JMRI authors but we don't yet have typed bindings for. Stored as
78    # serialized XML strings so the model stays JSON-friendly. Preserved
79    # on round-trip so a load → save doesn't destroy decoder CV state.
80    preserved_xml: list[str] = Field(default_factory=list)

A roster entry — what JMRI's RosterEntry carries.

id: str = PydanticUndefined
file_name: str = ''
dcc_address: str = '3'
long_address: bool = False
protocol: str = 'dcc_short'
decoder_family: str = ''
decoder_model: str = ''
decoder_comment: str = ''
road_name: str = ''
road_number: str = ''
mfg: str = ''
owner: str = ''
model: str = ''
comment: str = ''
max_speed_pct: int = 100
image_file_path: str | None = None
icon_file_path: str | None = None
url: str | None = None
date_modified: str | None = None
function_labels: list[jmri_core.beans.roster_entry.FunctionLabel] = PydanticUndefined
sound_labels: list[jmri_core.beans.roster_entry.SoundLabel] = PydanticUndefined
attributes: dict[str, str] = PydanticUndefined
preserved_xml: list[str] = PydanticUndefined
class Route(jmri_core.bean.AbstractNamedBean): 
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.

Route(system_name: str, user_name: str | None = None) 
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
type_letter = 'O'
steps: list[RouteStep]
state: jmri_core.state.RouteState 
46    @property
47    def state(self) -> RouteState:
48        return self._state
lock_sensor: Sensor | None 
50    @property
51    def lock_sensor(self) -> Sensor | None:
52        return self._lock_sensor
def set_lock_sensor(self, sensor: Sensor | None) -> None: 
54    def set_lock_sensor(self, sensor: Sensor | None) -> None:
55        self._lock_sensor = sensor
def add_step( self, turnout: Turnout, target: jmri_core.state.TurnoutState) -> None: 
57    def add_step(self, turnout: Turnout, target: TurnoutState) -> None:
58        self.steps.append(RouteStep(turnout, target))
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
jmri_core.bean.AbstractNamedBean
system_name
user_name
comment
state_changed
properties
display_name
class RouteManager(jmri_core.manager.Manager[jmri_core.beans.route.Route]): 
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.
type_letter = 'O'
Inherited Members
jmri_core.manager.Manager
Manager
system_prefix
added
removed
full_prefix
get_by_system_name
get_by_user_name
get
named_bean_set
provide
register
deregister
to_dict
@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: Turnout, target: jmri_core.state.TurnoutState)
turnout: Turnout
class Sensor(jmri_core.bean.AbstractNamedBean): 
18class Sensor(AbstractNamedBean):
19    """Generic layout input. Drivers call :meth:`_set_known_state` on receipt.
20
21    Sensors are write-only from the layout's perspective — the application
22    doesn't *command* them, only observes. The ``inverted`` flag flips the
23    semantic meaning of the raw hardware bit.
24    """
25
26    __slots__ = ("_known_state", "inverted")
27
28    type_letter = "S"
29
30    def __init__(self, system_name: str, user_name: str | None = None) -> None:
31        super().__init__(system_name, user_name)
32        self._known_state: SensorState = SensorState.UNKNOWN
33        self.inverted: bool = False
34
35    @property
36    def known_state(self) -> SensorState:
37        return self._known_state
38
39    async def _set_known_state(self, state: SensorState) -> None:
40        prev = self._known_state
41        if prev == state:
42            return
43        self._known_state = state
44        await self.state_changed.emit(SensorStateChange(prev, state))
45
46    @property
47    def is_active(self) -> bool:
48        return self._known_state == SensorState.ACTIVE

Generic layout input. Drivers call _set_known_state() on receipt.

Sensors are write-only from the layout's perspective — the application doesn't command them, only observes. The inverted flag flips the semantic meaning of the raw hardware bit.

Sensor(system_name: str, user_name: str | None = None) 
30    def __init__(self, system_name: str, user_name: str | None = None) -> None:
31        super().__init__(system_name, user_name)
32        self._known_state: SensorState = SensorState.UNKNOWN
33        self.inverted: bool = False
type_letter = 'S'
inverted: bool
known_state: jmri_core.state.SensorState 
35    @property
36    def known_state(self) -> SensorState:
37        return self._known_state
is_active: bool 
46    @property
47    def is_active(self) -> bool:
48        return self._known_state == SensorState.ACTIVE
Inherited Members
jmri_core.bean.AbstractNamedBean
system_name
user_name
comment
state_changed
properties
display_name
class SensorManager(jmri_core.manager.Manager[jmri_core.beans.sensor.Sensor]): 
51class SensorManager(Manager[Sensor]):
52    type_letter = "S"
53
54    def _create(self, system_name: str) -> Sensor:
55        return Sensor(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.
type_letter = 'S'
Inherited Members
jmri_core.manager.Manager
Manager
system_prefix
added
removed
full_prefix
get_by_system_name
get_by_user_name
get
named_bean_set
provide
register
deregister
to_dict
class SignalHead(jmri_core.bean.AbstractNamedBean): 
18class SignalHead(AbstractNamedBean):
19    """A single signal head (one physical signal, one face).
20
21    The ``held`` flag forces the head to RED regardless of its commanded
22    appearance — used by the dispatcher to override aspect logic.
23    """
24
25    __slots__ = ("_appearance", "_held")
26
27    type_letter = "H"
28
29    def __init__(self, system_name: str, user_name: str | None = None) -> None:
30        super().__init__(system_name, user_name)
31        self._appearance: SignalAppearance = SignalAppearance.DARK
32        self._held: bool = False
33
34    @property
35    def appearance(self) -> SignalAppearance:
36        return SignalAppearance.RED if self._held else self._appearance
37
38    @property
39    def commanded_appearance(self) -> SignalAppearance:
40        """The appearance ignoring ``held`` — what was last set explicitly."""
41        return self._appearance
42
43    @property
44    def held(self) -> bool:
45        return self._held
46
47    async def set_appearance(self, appearance: SignalAppearance) -> None:
48        prev_effective = self.appearance
49        self._appearance = appearance
50        new_effective = self.appearance
51        if prev_effective != new_effective:
52            await self.state_changed.emit(SignalHeadChange(prev_effective, new_effective))
53
54    async def set_held(self, held: bool) -> None:
55        if held == self._held:
56            return
57        prev_effective = self.appearance
58        self._held = held
59        new_effective = self.appearance
60        if prev_effective != new_effective:
61            await self.state_changed.emit(SignalHeadChange(prev_effective, new_effective))

A single signal head (one physical signal, one face).

The held flag forces the head to RED regardless of its commanded appearance — used by the dispatcher to override aspect logic.

SignalHead(system_name: str, user_name: str | None = None) 
29    def __init__(self, system_name: str, user_name: str | None = None) -> None:
30        super().__init__(system_name, user_name)
31        self._appearance: SignalAppearance = SignalAppearance.DARK
32        self._held: bool = False
type_letter = 'H'
appearance: jmri_core.state.SignalAppearance 
34    @property
35    def appearance(self) -> SignalAppearance:
36        return SignalAppearance.RED if self._held else self._appearance
commanded_appearance: jmri_core.state.SignalAppearance 
38    @property
39    def commanded_appearance(self) -> SignalAppearance:
40        """The appearance ignoring ``held`` — what was last set explicitly."""
41        return self._appearance

The appearance ignoring held — what was last set explicitly.

held: bool 
43    @property
44    def held(self) -> bool:
45        return self._held
async def set_appearance(self, appearance: jmri_core.state.SignalAppearance) -> None: 
47    async def set_appearance(self, appearance: SignalAppearance) -> None:
48        prev_effective = self.appearance
49        self._appearance = appearance
50        new_effective = self.appearance
51        if prev_effective != new_effective:
52            await self.state_changed.emit(SignalHeadChange(prev_effective, new_effective))
async def set_held(self, held: bool) -> None: 
54    async def set_held(self, held: bool) -> None:
55        if held == self._held:
56            return
57        prev_effective = self.appearance
58        self._held = held
59        new_effective = self.appearance
60        if prev_effective != new_effective:
61            await self.state_changed.emit(SignalHeadChange(prev_effective, new_effective))
Inherited Members
jmri_core.bean.AbstractNamedBean
system_name
user_name
comment
state_changed
properties
display_name
class SignalHeadManager(jmri_core.manager.Manager[jmri_core.beans.signal_head.SignalHead]): 
64class SignalHeadManager(Manager[SignalHead]):
65    type_letter = "H"
66
67    def _create(self, system_name: str) -> SignalHead:
68        return SignalHead(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.
type_letter = 'H'
Inherited Members
jmri_core.manager.Manager
Manager
system_prefix
added
removed
full_prefix
get_by_system_name
get_by_user_name
get
named_bean_set
provide
register
deregister
to_dict
class SignalMast(jmri_core.bean.AbstractNamedBean): 
24class SignalMast(AbstractNamedBean):
25    """Logical signal exposing named aspects from a signal system.
26
27    Attributes:
28
29    * ``aspect`` — current displayed aspect name; empty string means unknown.
30    * ``signal_system`` — the name of the signal-system XML this mast follows
31      (e.g. ``"basic"``, ``"AAR-1946"``). Resolved against
32      ``xml/signals/<system>/`` at load time by persistence code.
33    * ``held`` — forces ``"Stop"`` (or the system's most-restrictive aspect).
34    """
35
36    __slots__ = ("_aspect", "_held", "signal_system")
37
38    type_letter = "F"  # JMRI uses "F" for SignalMast system-type letter
39
40    def __init__(
41        self,
42        system_name: str,
43        signal_system: str = "",
44        user_name: str | None = None,
45    ) -> None:
46        super().__init__(system_name, user_name)
47        self.signal_system = signal_system
48        self._aspect: str = ""
49        self._held: bool = False
50
51    @property
52    def aspect(self) -> str:
53        return self._aspect
54
55    @property
56    def held(self) -> bool:
57        return self._held
58
59    async def set_aspect(self, aspect: str) -> None:
60        prev = self._aspect
61        if prev == aspect:
62            return
63        self._aspect = aspect
64        await self.state_changed.emit(SignalMastChange(prev, aspect))
65
66    async def set_held(self, held: bool) -> None:
67        # Held flips a flag; aspect display is computed elsewhere by the
68        # signal-mast logic. We emit a state-changed so listeners know
69        # something changed even if `aspect` didn't.
70        if held == self._held:
71            return
72        self._held = held
73        await self.state_changed.emit(SignalMastChange(self._aspect, self._aspect))

Logical signal exposing named aspects from a signal system.

Attributes:

  • aspect — current displayed aspect name; empty string means unknown.
  • signal_system — the name of the signal-system XML this mast follows (e.g. "basic", "AAR-1946"). Resolved against xml/signals/<system>/ at load time by persistence code.
  • held — forces "Stop" (or the system's most-restrictive aspect).
SignalMast( system_name: str, signal_system: str = '', user_name: str | None = None) 
40    def __init__(
41        self,
42        system_name: str,
43        signal_system: str = "",
44        user_name: str | None = None,
45    ) -> None:
46        super().__init__(system_name, user_name)
47        self.signal_system = signal_system
48        self._aspect: str = ""
49        self._held: bool = False
type_letter = 'F'
signal_system
aspect: str 
51    @property
52    def aspect(self) -> str:
53        return self._aspect
held: bool 
55    @property
56    def held(self) -> bool:
57        return self._held
async def set_aspect(self, aspect: str) -> None: 
59    async def set_aspect(self, aspect: str) -> None:
60        prev = self._aspect
61        if prev == aspect:
62            return
63        self._aspect = aspect
64        await self.state_changed.emit(SignalMastChange(prev, aspect))
async def set_held(self, held: bool) -> None: 
66    async def set_held(self, held: bool) -> None:
67        # Held flips a flag; aspect display is computed elsewhere by the
68        # signal-mast logic. We emit a state-changed so listeners know
69        # something changed even if `aspect` didn't.
70        if held == self._held:
71            return
72        self._held = held
73        await self.state_changed.emit(SignalMastChange(self._aspect, self._aspect))
Inherited Members
jmri_core.bean.AbstractNamedBean
system_name
user_name
comment
state_changed
properties
display_name
class SignalMastManager(jmri_core.manager.Manager[jmri_core.beans.signal_mast.SignalMast]): 
76class SignalMastManager(Manager[SignalMast]):
77    type_letter = "F"
78
79    def _create(self, system_name: str) -> SignalMast:
80        return SignalMast(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.
type_letter = 'F'
Inherited Members
jmri_core.manager.Manager
Manager
system_prefix
added
removed
full_prefix
get_by_system_name
get_by_user_name
get
named_bean_set
provide
register
deregister
to_dict
@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[jmri_core.beans.throttle.ThrottleChange] = <factory>)
address: int
long_address: bool
speed: float
functions: list[bool]
async def set_speed(self, speed: float) -> None: 
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))
async def set_direction(self, direction: jmri_core.state.ThrottleDirection) -> None: 
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))
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))
async def emergency_stop(self) -> None: 
71    async def emergency_stop(self) -> None:
72        await self.set_speed(-1.0)
class ThrottleManager: 
26class ThrottleManager:
27    """Acquire/release throttle handles by DCC address."""
28
29    def __init__(self, system_prefix: str) -> None:
30        if not system_prefix:
31            raise ValueError("system_prefix must be a non-empty string")
32        self.system_prefix = system_prefix
33        self._throttles: dict[int, Throttle] = {}
34        self.acquired: Signal[Throttle] = Signal()
35        self.released: Signal[Throttle] = Signal()
36
37    async def acquire(self, address: int, *, long_address: bool = False) -> Throttle:
38        if address in self._throttles:
39            return self._throttles[address]
40        throttle = await self._construct(address, long_address=long_address)
41        self._throttles[address] = throttle
42        await self.acquired.emit(throttle)
43        return throttle
44
45    async def release(self, throttle: Throttle) -> None:
46        existing = self._throttles.get(throttle.address)
47        if existing is not throttle:
48            return
49        del self._throttles[throttle.address]
50        await self.released.emit(throttle)
51
52    def get(self, address: int) -> Throttle | None:
53        return self._throttles.get(address)
54
55    def __iter__(self) -> Iterator[Throttle]:
56        return iter(self._throttles.values())
57
58    def __len__(self) -> int:
59        return len(self._throttles)
60
61    async def _construct(self, address: int, *, long_address: bool) -> Throttle:
62        """Build the throttle for *address*. Driver hook."""
63        return Throttle(address=address, long_address=long_address)

Acquire/release throttle handles by DCC address.

ThrottleManager(system_prefix: str) 
29    def __init__(self, system_prefix: str) -> None:
30        if not system_prefix:
31            raise ValueError("system_prefix must be a non-empty string")
32        self.system_prefix = system_prefix
33        self._throttles: dict[int, Throttle] = {}
34        self.acquired: Signal[Throttle] = Signal()
35        self.released: Signal[Throttle] = Signal()
system_prefix
async def acquire( self, address: int, *, long_address: bool = False) -> Throttle: 
37    async def acquire(self, address: int, *, long_address: bool = False) -> Throttle:
38        if address in self._throttles:
39            return self._throttles[address]
40        throttle = await self._construct(address, long_address=long_address)
41        self._throttles[address] = throttle
42        await self.acquired.emit(throttle)
43        return throttle
async def release(self, throttle: Throttle) -> None: 
45    async def release(self, throttle: Throttle) -> None:
46        existing = self._throttles.get(throttle.address)
47        if existing is not throttle:
48            return
49        del self._throttles[throttle.address]
50        await self.released.emit(throttle)
def get(self, address: int) -> Throttle | None: 
52    def get(self, address: int) -> Throttle | None:
53        return self._throttles.get(address)
class Turnout(jmri_core.bean.AbstractNamedBean): 
21class Turnout(AbstractNamedBean):
22    """Layout turnout (point switch).
23
24    A turnout has two state values:
25
26    * ``commanded_state`` — what we last *asked* the layout to do.
27    * ``known_state``     — what the layout reports back.
28
29    Drivers update ``known_state`` when their feedback channel reports a
30    change; this may differ from ``commanded_state`` until the move completes
31    or if motorised feedback says the switch ended up in an INCONSISTENT pos.
32    """
33
34    __slots__ = (
35        "_commanded_state",
36        "_known_state",
37        "feedback_mode",
38        "inverted",
39        "locked",
40    )
41
42    type_letter = "T"
43
44    def __init__(self, system_name: str, user_name: str | None = None) -> None:
45        super().__init__(system_name, user_name)
46        self._commanded_state: TurnoutState = TurnoutState.UNKNOWN
47        self._known_state: TurnoutState = TurnoutState.UNKNOWN
48        self.feedback_mode: TurnoutFeedback = TurnoutFeedback.DIRECT
49        self.inverted: bool = False
50        self.locked: bool = False
51
52    @property
53    def commanded_state(self) -> TurnoutState:
54        return self._commanded_state
55
56    @property
57    def known_state(self) -> TurnoutState:
58        return self._known_state
59
60    async def set_commanded_state(self, state: TurnoutState) -> None:
61        """Ask the layout to throw or close the switch.
62
63        The base implementation only updates the in-memory commanded state
64        and emits the signal. Driver subclasses override this to push a
65        hardware message *and then* call :meth:`super().set_commanded_state`.
66        """
67        prev = self._commanded_state
68        if prev == state:
69            return
70        self._commanded_state = state
71        await self.state_changed.emit(TurnoutStateChange("commanded", prev, state))
72        # For DIRECT feedback, known state mirrors commanded immediately.
73        if self.feedback_mode == TurnoutFeedback.DIRECT:
74            await self._set_known_state(state)
75
76    async def _set_known_state(self, state: TurnoutState) -> None:
77        """Driver hook: layout reported this state."""
78        prev = self._known_state
79        if prev == state:
80            return
81        self._known_state = state
82        await self.state_changed.emit(TurnoutStateChange("known", prev, state))

Layout turnout (point switch).

A turnout has two state values:

Drivers update known_state when their feedback channel reports a change; this may differ from commanded_state until the move completes or if motorised feedback says the switch ended up in an INCONSISTENT pos.

Turnout(system_name: str, user_name: str | None = None) 
44    def __init__(self, system_name: str, user_name: str | None = None) -> None:
45        super().__init__(system_name, user_name)
46        self._commanded_state: TurnoutState = TurnoutState.UNKNOWN
47        self._known_state: TurnoutState = TurnoutState.UNKNOWN
48        self.feedback_mode: TurnoutFeedback = TurnoutFeedback.DIRECT
49        self.inverted: bool = False
50        self.locked: bool = False
type_letter = 'T'
inverted: bool
locked: bool
commanded_state: jmri_core.state.TurnoutState 
52    @property
53    def commanded_state(self) -> TurnoutState:
54        return self._commanded_state
known_state: jmri_core.state.TurnoutState 
56    @property
57    def known_state(self) -> TurnoutState:
58        return self._known_state
async def set_commanded_state(self, state: jmri_core.state.TurnoutState) -> None: 
60    async def set_commanded_state(self, state: TurnoutState) -> None:
61        """Ask the layout to throw or close the switch.
62
63        The base implementation only updates the in-memory commanded state
64        and emits the signal. Driver subclasses override this to push a
65        hardware message *and then* call :meth:`super().set_commanded_state`.
66        """
67        prev = self._commanded_state
68        if prev == state:
69            return
70        self._commanded_state = state
71        await self.state_changed.emit(TurnoutStateChange("commanded", prev, state))
72        # For DIRECT feedback, known state mirrors commanded immediately.
73        if self.feedback_mode == TurnoutFeedback.DIRECT:
74            await self._set_known_state(state)

Ask the layout to throw or close the switch.

The base implementation only updates the in-memory commanded state and emits the signal. Driver subclasses override this to push a hardware message and then call super().set_commanded_state().

Inherited Members
jmri_core.bean.AbstractNamedBean
system_name
user_name
comment
state_changed
properties
display_name
class TurnoutManager(jmri_core.manager.Manager[jmri_core.beans.turnout.Turnout]): 
85class TurnoutManager(Manager[Turnout]):
86    type_letter = "T"
87
88    def _create(self, system_name: str) -> Turnout:
89        return Turnout(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.
type_letter = 'T'
Inherited Members
jmri_core.manager.Manager
Manager
system_prefix
added
removed
full_prefix
get_by_system_name
get_by_user_name
get
named_bean_set
provide
register
deregister
to_dict