jmri_core.beans.block

 1"""``Block`` — a track section with optional occupancy detection.
 2
 3A block carries:
 4  * a state (UNOCCUPIED/OCCUPIED/UNKNOWN/UNDETECTED);
 5  * an opaque value (typically a loco identifier, set when occupancy detection
 6    + reporters can identify the train);
 7  * an optional bound sensor; when bound, the block follows the sensor's
 8    active/inactive state automatically.
 9"""
10from __future__ import annotations
11
12from dataclasses import dataclass
13from typing import Any
14
15from jmri_core.bean import AbstractNamedBean
16from jmri_core.beans.sensor import Sensor, SensorStateChange
17from jmri_core.manager import Manager
18from jmri_core.state import BlockState, SensorState
19
20
21@dataclass(frozen=True, slots=True)
22class BlockChange:
23    kind: str  # "state" | "value"
24    previous: Any
25    current: Any
26
27
28class Block(AbstractNamedBean):
29    """Track block — possibly with an attached occupancy sensor."""
30
31    __slots__ = ("_disconnect_sensor", "_sensor", "_state", "_value")
32
33    type_letter = "B"
34
35    def __init__(self, system_name: str, user_name: str | None = None) -> None:
36        super().__init__(system_name, user_name)
37        self._state: BlockState = BlockState.UNKNOWN
38        self._value: Any = None
39        self._sensor: Sensor | None = None
40        self._disconnect_sensor: Any = None
41
42    @property
43    def state(self) -> BlockState:
44        return self._state
45
46    @property
47    def value(self) -> Any:
48        return self._value
49
50    @property
51    def sensor(self) -> Sensor | None:
52        return self._sensor
53
54    async def set_value(self, value: Any) -> None:
55        prev = self._value
56        if prev == value:
57            return
58        self._value = value
59        await self.state_changed.emit(BlockChange("value", prev, value))
60
61    async def _set_state(self, state: BlockState) -> None:
62        prev = self._state
63        if prev == state:
64            return
65        self._state = state
66        await self.state_changed.emit(BlockChange("state", prev, state))
67
68    def attach_sensor(self, sensor: Sensor | None) -> None:
69        """Bind a sensor whose state drives this block's state.
70
71        Pass ``None`` to detach. Detaching disconnects the listener so the
72        sensor reference can be garbage-collected.
73        """
74        if self._disconnect_sensor is not None:
75            self._disconnect_sensor()
76            self._disconnect_sensor = None
77        self._sensor = sensor
78        if sensor is not None:
79            self._disconnect_sensor = sensor.state_changed.connect(self._on_sensor_change)
80            # Seed initial state from the sensor.
81            self._state = _sensor_to_block_state(sensor.known_state)
82
83    async def _on_sensor_change(self, change: SensorStateChange) -> None:
84        await self._set_state(_sensor_to_block_state(change.current))
85
86
87def _sensor_to_block_state(s: SensorState) -> BlockState:
88    if s == SensorState.ACTIVE:
89        return BlockState.OCCUPIED
90    if s == SensorState.INACTIVE:
91        return BlockState.UNOCCUPIED
92    return BlockState.UNKNOWN
93
94
95class BlockManager(Manager[Block]):
96    type_letter = "B"
97
98    def _create(self, system_name: str) -> Block:
99        return Block(system_name)