jmri_core.beans.turnout

 1"""``Turnout`` — a layout switch (point) with CLOSED/THROWN states."""
 2from __future__ import annotations
 3
 4from dataclasses import dataclass
 5
 6from jmri_core.bean import AbstractNamedBean
 7from jmri_core.manager import Manager
 8from jmri_core.state import TurnoutFeedback, TurnoutState
 9
10
11@dataclass(frozen=True, slots=True)
12class TurnoutStateChange:
13    """Payload emitted on :attr:`Turnout.state_changed`."""
14
15    kind: str          # "commanded" or "known"
16    previous: TurnoutState
17    current: TurnoutState
18
19
20class Turnout(AbstractNamedBean):
21    """Layout turnout (point switch).
22
23    A turnout has two state values:
24
25    * ``commanded_state`` — what we last *asked* the layout to do.
26    * ``known_state``     — what the layout reports back.
27
28    Drivers update ``known_state`` when their feedback channel reports a
29    change; this may differ from ``commanded_state`` until the move completes
30    or if motorised feedback says the switch ended up in an INCONSISTENT pos.
31    """
32
33    __slots__ = (
34        "_commanded_state",
35        "_known_state",
36        "feedback_mode",
37        "inverted",
38        "locked",
39    )
40
41    type_letter = "T"
42
43    def __init__(self, system_name: str, user_name: str | None = None) -> None:
44        super().__init__(system_name, user_name)
45        self._commanded_state: TurnoutState = TurnoutState.UNKNOWN
46        self._known_state: TurnoutState = TurnoutState.UNKNOWN
47        self.feedback_mode: TurnoutFeedback = TurnoutFeedback.DIRECT
48        self.inverted: bool = False
49        self.locked: bool = False
50
51    @property
52    def commanded_state(self) -> TurnoutState:
53        return self._commanded_state
54
55    @property
56    def known_state(self) -> TurnoutState:
57        return self._known_state
58
59    async def set_commanded_state(self, state: TurnoutState) -> None:
60        """Ask the layout to throw or close the switch.
61
62        The base implementation only updates the in-memory commanded state
63        and emits the signal. Driver subclasses override this to push a
64        hardware message *and then* call :meth:`super().set_commanded_state`.
65        """
66        prev = self._commanded_state
67        if prev == state:
68            return
69        self._commanded_state = state
70        await self.state_changed.emit(TurnoutStateChange("commanded", prev, state))
71        # For DIRECT feedback, known state mirrors commanded immediately.
72        if self.feedback_mode == TurnoutFeedback.DIRECT:
73            await self._set_known_state(state)
74
75    async def _set_known_state(self, state: TurnoutState) -> None:
76        """Driver hook: layout reported this state."""
77        prev = self._known_state
78        if prev == state:
79            return
80        self._known_state = state
81        await self.state_changed.emit(TurnoutStateChange("known", prev, state))
82
83
84class TurnoutManager(Manager[Turnout]):
85    type_letter = "T"
86
87    def _create(self, system_name: str) -> Turnout:
88        return Turnout(system_name)