jmri_logixng.actions

37@register
38class Delay(Action):
39    """Sleep for *seconds*. Useful for timed sequences (signal aspect change → hold)."""
40
41    type_tag: ClassVar[str] = "action:delay"
42
43    def __init__(
44        self,
45        *,
46        seconds: float,
47        user_name: str = "",
48        node_id: str | None = None,
49    ) -> None:
50        super().__init__(user_name=user_name, node_id=node_id)
51        if seconds < 0:
52            raise ValueError(f"delay seconds must be non-negative, got {seconds}")
53        self.seconds = seconds
54
55    async def execute(self, ctx: ExecutionContext) -> None:
56        ctx.emit_trace(f"delay {self.seconds}s")
57        await asyncio.sleep(self.seconds)
58
59    def extra_fields(self) -> dict[str, Any]:
60        return {"seconds": self.seconds}

145@register
146class IfThenElse(Action):
147    """Evaluate *condition*; run *then_action* if true, *else_action* otherwise."""
148
149    type_tag: ClassVar[str] = "action:if-then-else"
150
151    def __init__(
152        self,
153        *,
154        condition: Expression[bool],
155        then_action: Action,
156        else_action: Action | None = None,
157        user_name: str = "",
158        node_id: str | None = None,
159    ) -> None:
160        super().__init__(user_name=user_name, node_id=node_id)
161        self.condition = condition
162        self.then_action = then_action
163        self.else_action = else_action
164
165    async def execute(self, ctx: ExecutionContext) -> None:
166        result = await self.condition.evaluate(ctx)
167        ctx.emit_trace(f"if-then-else condition={result}")
168        if result:
169            await self.then_action.execute(ctx)
170        elif self.else_action is not None:
171            await self.else_action.execute(ctx)
172
173    def extra_fields(self) -> dict[str, Any]:
174        out: dict[str, Any] = {
175            "condition": self.condition.to_dict(),
176            "then": self.then_action.to_dict(),
177        }
178        if self.else_action is not None:
179            out["else"] = self.else_action.to_dict()
180        return out

183@register
184class LogAction(Action):
185    """Emit a log message. Mostly useful for debugging automations."""
186
187    type_tag: ClassVar[str] = "action:log"
188
189    def __init__(
190        self,
191        *,
192        message: str,
193        level: str = "info",
194        user_name: str = "",
195        node_id: str | None = None,
196    ) -> None:
197        super().__init__(user_name=user_name, node_id=node_id)
198        self.message = message
199        self.level = level
200
201    async def execute(self, ctx: ExecutionContext) -> None:
202        ctx.emit_trace(f"log[{self.level}]: {self.message}")
203        getattr(ctx.log, self.level, ctx.log.info)(self.message)
204
205    def extra_fields(self) -> dict[str, Any]:
206        return {"message": self.message, "level": self.level}

13@register
14class Sequence(Action):
15    """Run a list of actions in order."""
16
17    type_tag: ClassVar[str] = "action:sequence"
18
19    def __init__(
20        self,
21        actions: list[Action],
22        *,
23        user_name: str = "",
24        node_id: str | None = None,
25    ) -> None:
26        super().__init__(user_name=user_name, node_id=node_id)
27        self.actions = list(actions)
28
29    async def execute(self, ctx: ExecutionContext) -> None:
30        for action in self.actions:
31            await action.execute(ctx)
32
33    def extra_fields(self) -> dict[str, Any]:
34        return {"actions": [a.to_dict() for a in self.actions]}

 85@register
 86class SetLightState(Action):
 87    """Set a light's state."""
 88
 89    type_tag: ClassVar[str] = "action:set-light"
 90
 91    def __init__(
 92        self,
 93        *,
 94        light: str,
 95        state: LightState,
 96        user_name: str = "",
 97        node_id: str | None = None,
 98    ) -> None:
 99        super().__init__(user_name=user_name, node_id=node_id)
100        self.light = light
101        self.state = state
102
103    async def execute(self, ctx: ExecutionContext) -> None:
104        bean = ctx.connection.lights.get(self.light)
105        if bean is None:
106            ctx.log.warning("SetLightState: no light %r", self.light)
107            return
108        ctx.emit_trace(f"set light {self.light} → {self.state.name}")
109        await bean.set_state(self.state)
110
111    def extra_fields(self) -> dict[str, Any]:
112        return {"light": self.light, "state": self.state.name}

115@register
116class SetMemory(Action):
117    """Write a value into a memory bean."""
118
119    type_tag: ClassVar[str] = "action:set-memory"
120
121    def __init__(
122        self,
123        *,
124        memory: str,
125        value: Any,
126        user_name: str = "",
127        node_id: str | None = None,
128    ) -> None:
129        super().__init__(user_name=user_name, node_id=node_id)
130        self.memory = memory
131        self.value = value
132
133    async def execute(self, ctx: ExecutionContext) -> None:
134        bean = ctx.connection.memories.get(self.memory)
135        if bean is None:
136            ctx.log.warning("SetMemory: no memory %r", self.memory)
137            return
138        ctx.emit_trace(f"set memory {self.memory} → {self.value!r}")
139        await bean.set_value(self.value)
140
141    def extra_fields(self) -> dict[str, Any]:
142        return {"memory": self.memory, "value": self.value}

55@register
56class SetSensorState(Action):
57    """Set a sensor's known state (mostly for tests/simulation)."""
58
59    type_tag: ClassVar[str] = "action:set-sensor"
60
61    def __init__(
62        self,
63        *,
64        sensor: str,
65        state: SensorState,
66        user_name: str = "",
67        node_id: str | None = None,
68    ) -> None:
69        super().__init__(user_name=user_name, node_id=node_id)
70        self.sensor = sensor
71        self.state = state
72
73    async def execute(self, ctx: ExecutionContext) -> None:
74        bean = ctx.connection.sensors.get(self.sensor)
75        if bean is None:
76            ctx.log.warning("SetSensorState: no sensor %r", self.sensor)
77            return
78        ctx.emit_trace(f"set sensor {self.sensor} → {self.state.name}")
79        await bean._set_known_state(self.state)
80
81    def extra_fields(self) -> dict[str, Any]:
82        return {"sensor": self.sensor, "state": self.state.name}

25@register
26class SetTurnoutState(Action):
27    """Set a turnout's commanded state."""
28
29    type_tag: ClassVar[str] = "action:set-turnout"
30
31    def __init__(
32        self,
33        *,
34        turnout: str,
35        state: TurnoutState,
36        user_name: str = "",
37        node_id: str | None = None,
38    ) -> None:
39        super().__init__(user_name=user_name, node_id=node_id)
40        self.turnout = turnout
41        self.state = state
42
43    async def execute(self, ctx: ExecutionContext) -> None:
44        bean = ctx.connection.turnouts.get(self.turnout)
45        if bean is None:
46            ctx.log.warning("SetTurnoutState: no turnout %r", self.turnout)
47            return
48        ctx.emit_trace(f"set turnout {self.turnout} → {self.state.name}")
49        await bean.set_commanded_state(self.state)
50
51    def extra_fields(self) -> dict[str, Any]:
52        return {"turnout": self.turnout, "state": self.state.name}