jmri_logixng.actions.digital

  1"""Digital actions — set bean state, conditionally execute, log.
  2
  3Each action holds a serializable description of *what* to do; the actual
  4lookup of the bean happens at :meth:`execute` time against the active
  5connection in the :class:`ExecutionContext`. This decouples authoring
  6from binding: the same automation file can run against
  7``InternalConnection`` in tests and a real LocoNet/NCE connection in
  8production.
  9"""
 10from __future__ import annotations
 11
 12from typing import TYPE_CHECKING, Any, ClassVar
 13
 14from jmri_core.state import LightState, SensorState, TurnoutState
 15
 16from jmri_logixng.context import ExecutionContext
 17from jmri_logixng.registry import register
 18from jmri_logixng.tree import Action, Expression
 19
 20if TYPE_CHECKING:
 21    pass
 22
 23
 24@register
 25class SetTurnoutState(Action):
 26    """Set a turnout's commanded state."""
 27
 28    type_tag: ClassVar[str] = "action:set-turnout"
 29
 30    def __init__(
 31        self,
 32        *,
 33        turnout: str,
 34        state: TurnoutState,
 35        user_name: str = "",
 36        node_id: str | None = None,
 37    ) -> None:
 38        super().__init__(user_name=user_name, node_id=node_id)
 39        self.turnout = turnout
 40        self.state = state
 41
 42    async def execute(self, ctx: ExecutionContext) -> None:
 43        bean = ctx.connection.turnouts.get(self.turnout)
 44        if bean is None:
 45            ctx.log.warning("SetTurnoutState: no turnout %r", self.turnout)
 46            return
 47        ctx.emit_trace(f"set turnout {self.turnout} → {self.state.name}")
 48        await bean.set_commanded_state(self.state)
 49
 50    def extra_fields(self) -> dict[str, Any]:
 51        return {"turnout": self.turnout, "state": self.state.name}
 52
 53
 54@register
 55class SetSensorState(Action):
 56    """Set a sensor's known state (mostly for tests/simulation)."""
 57
 58    type_tag: ClassVar[str] = "action:set-sensor"
 59
 60    def __init__(
 61        self,
 62        *,
 63        sensor: str,
 64        state: SensorState,
 65        user_name: str = "",
 66        node_id: str | None = None,
 67    ) -> None:
 68        super().__init__(user_name=user_name, node_id=node_id)
 69        self.sensor = sensor
 70        self.state = state
 71
 72    async def execute(self, ctx: ExecutionContext) -> None:
 73        bean = ctx.connection.sensors.get(self.sensor)
 74        if bean is None:
 75            ctx.log.warning("SetSensorState: no sensor %r", self.sensor)
 76            return
 77        ctx.emit_trace(f"set sensor {self.sensor} → {self.state.name}")
 78        await bean._set_known_state(self.state)
 79
 80    def extra_fields(self) -> dict[str, Any]:
 81        return {"sensor": self.sensor, "state": self.state.name}
 82
 83
 84@register
 85class SetLightState(Action):
 86    """Set a light's state."""
 87
 88    type_tag: ClassVar[str] = "action:set-light"
 89
 90    def __init__(
 91        self,
 92        *,
 93        light: str,
 94        state: LightState,
 95        user_name: str = "",
 96        node_id: str | None = None,
 97    ) -> None:
 98        super().__init__(user_name=user_name, node_id=node_id)
 99        self.light = light
100        self.state = state
101
102    async def execute(self, ctx: ExecutionContext) -> None:
103        bean = ctx.connection.lights.get(self.light)
104        if bean is None:
105            ctx.log.warning("SetLightState: no light %r", self.light)
106            return
107        ctx.emit_trace(f"set light {self.light} → {self.state.name}")
108        await bean.set_state(self.state)
109
110    def extra_fields(self) -> dict[str, Any]:
111        return {"light": self.light, "state": self.state.name}
112
113
114@register
115class SetMemory(Action):
116    """Write a value into a memory bean."""
117
118    type_tag: ClassVar[str] = "action:set-memory"
119
120    def __init__(
121        self,
122        *,
123        memory: str,
124        value: Any,
125        user_name: str = "",
126        node_id: str | None = None,
127    ) -> None:
128        super().__init__(user_name=user_name, node_id=node_id)
129        self.memory = memory
130        self.value = value
131
132    async def execute(self, ctx: ExecutionContext) -> None:
133        bean = ctx.connection.memories.get(self.memory)
134        if bean is None:
135            ctx.log.warning("SetMemory: no memory %r", self.memory)
136            return
137        ctx.emit_trace(f"set memory {self.memory} → {self.value!r}")
138        await bean.set_value(self.value)
139
140    def extra_fields(self) -> dict[str, Any]:
141        return {"memory": self.memory, "value": self.value}
142
143
144@register
145class IfThenElse(Action):
146    """Evaluate *condition*; run *then_action* if true, *else_action* otherwise."""
147
148    type_tag: ClassVar[str] = "action:if-then-else"
149
150    def __init__(
151        self,
152        *,
153        condition: Expression[bool],
154        then_action: Action,
155        else_action: Action | None = None,
156        user_name: str = "",
157        node_id: str | None = None,
158    ) -> None:
159        super().__init__(user_name=user_name, node_id=node_id)
160        self.condition = condition
161        self.then_action = then_action
162        self.else_action = else_action
163
164    async def execute(self, ctx: ExecutionContext) -> None:
165        result = await self.condition.evaluate(ctx)
166        ctx.emit_trace(f"if-then-else condition={result}")
167        if result:
168            await self.then_action.execute(ctx)
169        elif self.else_action is not None:
170            await self.else_action.execute(ctx)
171
172    def extra_fields(self) -> dict[str, Any]:
173        out: dict[str, Any] = {
174            "condition": self.condition.to_dict(),
175            "then": self.then_action.to_dict(),
176        }
177        if self.else_action is not None:
178            out["else"] = self.else_action.to_dict()
179        return out
180
181
182@register
183class LogAction(Action):
184    """Emit a log message. Mostly useful for debugging automations."""
185
186    type_tag: ClassVar[str] = "action:log"
187
188    def __init__(
189        self,
190        *,
191        message: str,
192        level: str = "info",
193        user_name: str = "",
194        node_id: str | None = None,
195    ) -> None:
196        super().__init__(user_name=user_name, node_id=node_id)
197        self.message = message
198        self.level = level
199
200    async def execute(self, ctx: ExecutionContext) -> None:
201        ctx.emit_trace(f"log[{self.level}]: {self.message}")
202        getattr(ctx.log, self.level, ctx.log.info)(self.message)
203
204    def extra_fields(self) -> dict[str, Any]:
205        return {"message": self.message, "level": self.level}