jmri_logixng.expressions

134@register
135class And(Expression[bool]):
136    """True iff every child expression evaluates true. Short-circuits."""
137
138    type_tag: ClassVar[str] = "expr:and"
139
140    def __init__(
141        self,
142        children: list[Expression[bool]],
143        *,
144        user_name: str = "",
145        node_id: str | None = None,
146    ) -> None:
147        super().__init__(user_name=user_name, node_id=node_id)
148        self.children = list(children)
149
150    async def evaluate(self, ctx: ExecutionContext) -> bool:
151        for child in self.children:
152            if not await child.evaluate(ctx):
153                return False
154        return True
155
156    def extra_fields(self) -> dict[str, Any]:
157        return {"children": [c.to_dict() for c in self.children]}

14@register
15class ConstantBool(Expression[bool]):
16    """Always evaluates to ``value``. Used for testing + as the leaf of trees."""
17
18    type_tag: ClassVar[str] = "expr:const-bool"
19
20    def __init__(
21        self, *, value: bool, user_name: str = "", node_id: str | None = None
22    ) -> None:
23        super().__init__(user_name=user_name, node_id=node_id)
24        self.value = value
25
26    async def evaluate(self, ctx: ExecutionContext) -> bool:
27        return self.value
28
29    def extra_fields(self) -> dict[str, Any]:
30        return {"value": self.value}

79@register
80class LightIsOn(Expression[bool]):
81    """True when light *light* has state ON."""
82
83    type_tag: ClassVar[str] = "expr:light-on"
84
85    def __init__(
86        self, *, light: str, user_name: str = "", node_id: str | None = None
87    ) -> None:
88        super().__init__(user_name=user_name, node_id=node_id)
89        self.light = light
90
91    async def evaluate(self, ctx: ExecutionContext) -> bool:
92        bean = ctx.connection.lights.get(self.light)
93        if bean is None:
94            ctx.log.warning("LightIsOn: no light %r", self.light)
95            return False
96        return bean.state == LightState.ON
97
98    def extra_fields(self) -> dict[str, Any]:
99        return {"light": self.light}

102@register
103class MemoryEquals(Expression[bool]):
104    """True when memory *memory*'s value compares equal to *value*."""
105
106    type_tag: ClassVar[str] = "expr:memory-equals"
107
108    def __init__(
109        self,
110        *,
111        memory: str,
112        value: Any,
113        user_name: str = "",
114        node_id: str | None = None,
115    ) -> None:
116        super().__init__(user_name=user_name, node_id=node_id)
117        self.memory = memory
118        self.value = value
119
120    async def evaluate(self, ctx: ExecutionContext) -> bool:
121        bean = ctx.connection.memories.get(self.memory)
122        if bean is None:
123            ctx.log.warning("MemoryEquals: no memory %r", self.memory)
124            return False
125        return bool(bean.value == self.value)
126
127    def extra_fields(self) -> dict[str, Any]:
128        return {"memory": self.memory, "value": self.value}

186@register
187class Not(Expression[bool]):
188    """Negates its child expression."""
189
190    type_tag: ClassVar[str] = "expr:not"
191
192    def __init__(
193        self,
194        child: Expression[bool],
195        *,
196        user_name: str = "",
197        node_id: str | None = None,
198    ) -> None:
199        super().__init__(user_name=user_name, node_id=node_id)
200        self.child = child
201
202    async def evaluate(self, ctx: ExecutionContext) -> bool:
203        return not await self.child.evaluate(ctx)
204
205    def extra_fields(self) -> dict[str, Any]:
206        return {"child": self.child.to_dict()}

160@register
161class Or(Expression[bool]):
162    """True if any child expression evaluates true. Short-circuits."""
163
164    type_tag: ClassVar[str] = "expr:or"
165
166    def __init__(
167        self,
168        children: list[Expression[bool]],
169        *,
170        user_name: str = "",
171        node_id: str | None = None,
172    ) -> None:
173        super().__init__(user_name=user_name, node_id=node_id)
174        self.children = list(children)
175
176    async def evaluate(self, ctx: ExecutionContext) -> bool:
177        for child in self.children:
178            if await child.evaluate(ctx):
179                return True
180        return False
181
182    def extra_fields(self) -> dict[str, Any]:
183        return {"children": [c.to_dict() for c in self.children]}

33@register
34class SensorIsActive(Expression[bool]):
35    """True when sensor *sensor* has state ACTIVE."""
36
37    type_tag: ClassVar[str] = "expr:sensor-active"
38
39    def __init__(
40        self, *, sensor: str, user_name: str = "", node_id: str | None = None
41    ) -> None:
42        super().__init__(user_name=user_name, node_id=node_id)
43        self.sensor = sensor
44
45    async def evaluate(self, ctx: ExecutionContext) -> bool:
46        bean = ctx.connection.sensors.get(self.sensor)
47        if bean is None:
48            ctx.log.warning("SensorIsActive: no sensor %r", self.sensor)
49            return False
50        return bean.known_state == SensorState.ACTIVE
51
52    def extra_fields(self) -> dict[str, Any]:
53        return {"sensor": self.sensor}

56@register
57class TurnoutIsClosed(Expression[bool]):
58    """True when turnout *turnout* has known state CLOSED."""
59
60    type_tag: ClassVar[str] = "expr:turnout-closed"
61
62    def __init__(
63        self, *, turnout: str, user_name: str = "", node_id: str | None = None
64    ) -> None:
65        super().__init__(user_name=user_name, node_id=node_id)
66        self.turnout = turnout
67
68    async def evaluate(self, ctx: ExecutionContext) -> bool:
69        bean = ctx.connection.turnouts.get(self.turnout)
70        if bean is None:
71            ctx.log.warning("TurnoutIsClosed: no turnout %r", self.turnout)
72            return False
73        return bean.known_state == TurnoutState.CLOSED
74
75    def extra_fields(self) -> dict[str, Any]:
76        return {"turnout": self.turnout}