jmri_openlcb.power

Well-known-event power manager for OpenLCB.

OpenLCB reserves two event IDs for global track power (S-9.7.4.3 and the JMRI Java OlcbPowerManager implementation):

01.00.00.00.00.00.FF.FE — POWER ON
01.00.00.00.00.00.FF.FF — POWER OFF

Any node that observes those events treats them as authoritative for global power state. Any node that sets power emits the corresponding PCER so every listener converges. This class:

On set_power(), sends the ON or OFF PCER via the shared ~jmri_openlcb.traffic.OlcbTrafficController.
On observed PCER of either well-known event, flips its own in-memory state so anyone watching the connection's power manager sees the change without polling.

An UNKNOWN request is treated as a no-op — there's no "please forget" event on the wire.

View Source

 1"""Well-known-event power manager for OpenLCB.
 2
 3OpenLCB reserves two event IDs for global track power (S-9.7.4.3 and
 4the JMRI Java ``OlcbPowerManager`` implementation):
 5
 6* ``01.00.00.00.00.00.FF.FE`` — POWER ON
 7* ``01.00.00.00.00.00.FF.FF`` — POWER OFF
 8
 9Any node that observes those events treats them as authoritative for
10global power state. Any node that sets power emits the corresponding
11PCER so every listener converges. This class:
12
13* On :meth:`set_power`, sends the ON or OFF PCER via the shared
14  :class:`~jmri_openlcb.traffic.OlcbTrafficController`.
15* On observed PCER of either well-known event, flips its own
16  in-memory state so anyone watching the connection's power
17  manager sees the change without polling.
18
19An UNKNOWN request is treated as a no-op — there's no "please forget"
20event on the wire.
21"""
22from __future__ import annotations
23
24from jmri_core.beans.power import PowerManager
25from jmri_core.state import PowerState
26
27from jmri_openlcb.events import EventId, EventReport
28from jmri_openlcb.traffic import OlcbTrafficController
29
30POWER_ON = EventId.parse("01.00.00.00.00.00.FF.FE")
31POWER_OFF = EventId.parse("01.00.00.00.00.00.FF.FF")
32
33
34class OlcbPowerManager(PowerManager):
35    """Power manager that produces + consumes the well-known events."""
36
37    def __init__(self, tc: OlcbTrafficController) -> None:
38        super().__init__()
39        self._tc = tc
40        self._unsubscribe = tc.events.connect(self._on_event)
41
42    async def set_power(self, state: PowerState) -> None:
43        # Emit the well-known event so every other node on the bus
44        # observes the transition. Local state is updated by the same
45        # inbound-event handler when our own broadcast comes back
46        # around, keeping the "we just heard about it from the bus"
47        # semantics uniform whether the change came from us or a peer.
48        if state is PowerState.ON:
49            await self._tc.produce_event(POWER_ON)
50        elif state is PowerState.OFF:
51            await self._tc.produce_event(POWER_OFF)
52        else:
53            # UNKNOWN is a "we don't know" state, not a command; no
54            # matching event exists.
55            return
56        # Some loopback / test transports don't echo our own frames
57        # back, so update the internal state explicitly. If the bus
58        # DOES echo, the inbound handler will see it and no-op on
59        # the already-current state (see ``_set_state`` in the base).
60        await self._set_state(state)
61
62    def close(self) -> None:
63        """Detach the inbound listener. Idempotent."""
64        if self._unsubscribe is not None:
65            self._unsubscribe()
66            self._unsubscribe = None
67
68    async def _on_event(self, report: EventReport) -> None:
69        if report.event == POWER_ON:
70            await self._set_state(PowerState.ON)
71        elif report.event == POWER_OFF:
72            await self._set_state(PowerState.OFF)

POWER_ON = EventId(value=72057594037993470)

POWER_OFF = EventId(value=72057594037993471)

class OlcbPowerManager(jmri_core.beans.power.PowerManager): View Source

35class OlcbPowerManager(PowerManager):
36    """Power manager that produces + consumes the well-known events."""
37
38    def __init__(self, tc: OlcbTrafficController) -> None:
39        super().__init__()
40        self._tc = tc
41        self._unsubscribe = tc.events.connect(self._on_event)
42
43    async def set_power(self, state: PowerState) -> None:
44        # Emit the well-known event so every other node on the bus
45        # observes the transition. Local state is updated by the same
46        # inbound-event handler when our own broadcast comes back
47        # around, keeping the "we just heard about it from the bus"
48        # semantics uniform whether the change came from us or a peer.
49        if state is PowerState.ON:
50            await self._tc.produce_event(POWER_ON)
51        elif state is PowerState.OFF:
52            await self._tc.produce_event(POWER_OFF)
53        else:
54            # UNKNOWN is a "we don't know" state, not a command; no
55            # matching event exists.
56            return
57        # Some loopback / test transports don't echo our own frames
58        # back, so update the internal state explicitly. If the bus
59        # DOES echo, the inbound handler will see it and no-op on
60        # the already-current state (see ``_set_state`` in the base).
61        await self._set_state(state)
62
63    def close(self) -> None:
64        """Detach the inbound listener. Idempotent."""
65        if self._unsubscribe is not None:
66            self._unsubscribe()
67            self._unsubscribe = None
68
69    async def _on_event(self, report: EventReport) -> None:
70        if report.event == POWER_ON:
71            await self._set_state(PowerState.ON)
72        elif report.event == POWER_OFF:
73            await self._set_state(PowerState.OFF)

Power manager that produces + consumes the well-known events.

OlcbPowerManager(tc: jmri_openlcb.OlcbTrafficController) View Source

38    def __init__(self, tc: OlcbTrafficController) -> None:
39        super().__init__()
40        self._tc = tc
41        self._unsubscribe = tc.events.connect(self._on_event)

async def set_power(self, state: jmri_core.state.PowerState) -> None: View Source

43    async def set_power(self, state: PowerState) -> None:
44        # Emit the well-known event so every other node on the bus
45        # observes the transition. Local state is updated by the same
46        # inbound-event handler when our own broadcast comes back
47        # around, keeping the "we just heard about it from the bus"
48        # semantics uniform whether the change came from us or a peer.
49        if state is PowerState.ON:
50            await self._tc.produce_event(POWER_ON)
51        elif state is PowerState.OFF:
52            await self._tc.produce_event(POWER_OFF)
53        else:
54            # UNKNOWN is a "we don't know" state, not a command; no
55            # matching event exists.
56            return
57        # Some loopback / test transports don't echo our own frames
58        # back, so update the internal state explicitly. If the bus
59        # DOES echo, the inbound handler will see it and no-op on
60        # the already-current state (see ``_set_state`` in the base).
61        await self._set_state(state)

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

def close(self) -> None: View Source

63    def close(self) -> None:
64        """Detach the inbound listener. Idempotent."""
65        if self._unsubscribe is not None:
66            self._unsubscribe()
67            self._unsubscribe = None

Detach the inbound listener. Idempotent.

Inherited Members

jmri_core.beans.power.PowerManager: state_changed; state