jmri_logixng.expressions.digital

  1"""Digital (boolean) expressions over layout state + boolean algebra."""
  2from __future__ import annotations
  3
  4from typing import Any, ClassVar
  5
  6from jmri_core.state import LightState, SensorState, TurnoutState
  7
  8from jmri_logixng.context import ExecutionContext
  9from jmri_logixng.registry import register
 10from jmri_logixng.tree import Expression
 11
 12
 13@register
 14class ConstantBool(Expression[bool]):
 15    """Always evaluates to ``value``. Used for testing + as the leaf of trees."""
 16
 17    type_tag: ClassVar[str] = "expr:const-bool"
 18
 19    def __init__(
 20        self, *, value: bool, user_name: str = "", node_id: str | None = None
 21    ) -> None:
 22        super().__init__(user_name=user_name, node_id=node_id)
 23        self.value = value
 24
 25    async def evaluate(self, ctx: ExecutionContext) -> bool:
 26        return self.value
 27
 28    def extra_fields(self) -> dict[str, Any]:
 29        return {"value": self.value}
 30
 31
 32@register
 33class SensorIsActive(Expression[bool]):
 34    """True when sensor *sensor* has state ACTIVE."""
 35
 36    type_tag: ClassVar[str] = "expr:sensor-active"
 37
 38    def __init__(
 39        self, *, sensor: str, user_name: str = "", node_id: str | None = None
 40    ) -> None:
 41        super().__init__(user_name=user_name, node_id=node_id)
 42        self.sensor = sensor
 43
 44    async def evaluate(self, ctx: ExecutionContext) -> bool:
 45        bean = ctx.connection.sensors.get(self.sensor)
 46        if bean is None:
 47            ctx.log.warning("SensorIsActive: no sensor %r", self.sensor)
 48            return False
 49        return bean.known_state == SensorState.ACTIVE
 50
 51    def extra_fields(self) -> dict[str, Any]:
 52        return {"sensor": self.sensor}
 53
 54
 55@register
 56class TurnoutIsClosed(Expression[bool]):
 57    """True when turnout *turnout* has known state CLOSED."""
 58
 59    type_tag: ClassVar[str] = "expr:turnout-closed"
 60
 61    def __init__(
 62        self, *, turnout: str, user_name: str = "", node_id: str | None = None
 63    ) -> None:
 64        super().__init__(user_name=user_name, node_id=node_id)
 65        self.turnout = turnout
 66
 67    async def evaluate(self, ctx: ExecutionContext) -> bool:
 68        bean = ctx.connection.turnouts.get(self.turnout)
 69        if bean is None:
 70            ctx.log.warning("TurnoutIsClosed: no turnout %r", self.turnout)
 71            return False
 72        return bean.known_state == TurnoutState.CLOSED
 73
 74    def extra_fields(self) -> dict[str, Any]:
 75        return {"turnout": self.turnout}
 76
 77
 78@register
 79class LightIsOn(Expression[bool]):
 80    """True when light *light* has state ON."""
 81
 82    type_tag: ClassVar[str] = "expr:light-on"
 83
 84    def __init__(
 85        self, *, light: str, user_name: str = "", node_id: str | None = None
 86    ) -> None:
 87        super().__init__(user_name=user_name, node_id=node_id)
 88        self.light = light
 89
 90    async def evaluate(self, ctx: ExecutionContext) -> bool:
 91        bean = ctx.connection.lights.get(self.light)
 92        if bean is None:
 93            ctx.log.warning("LightIsOn: no light %r", self.light)
 94            return False
 95        return bean.state == LightState.ON
 96
 97    def extra_fields(self) -> dict[str, Any]:
 98        return {"light": self.light}
 99
100
101@register
102class MemoryEquals(Expression[bool]):
103    """True when memory *memory*'s value compares equal to *value*."""
104
105    type_tag: ClassVar[str] = "expr:memory-equals"
106
107    def __init__(
108        self,
109        *,
110        memory: str,
111        value: Any,
112        user_name: str = "",
113        node_id: str | None = None,
114    ) -> None:
115        super().__init__(user_name=user_name, node_id=node_id)
116        self.memory = memory
117        self.value = value
118
119    async def evaluate(self, ctx: ExecutionContext) -> bool:
120        bean = ctx.connection.memories.get(self.memory)
121        if bean is None:
122            ctx.log.warning("MemoryEquals: no memory %r", self.memory)
123            return False
124        return bool(bean.value == self.value)
125
126    def extra_fields(self) -> dict[str, Any]:
127        return {"memory": self.memory, "value": self.value}
128
129
130# --- boolean combinators ---------------------------------------------------
131
132
133@register
134class And(Expression[bool]):
135    """True iff every child expression evaluates true. Short-circuits."""
136
137    type_tag: ClassVar[str] = "expr:and"
138
139    def __init__(
140        self,
141        children: list[Expression[bool]],
142        *,
143        user_name: str = "",
144        node_id: str | None = None,
145    ) -> None:
146        super().__init__(user_name=user_name, node_id=node_id)
147        self.children = list(children)
148
149    async def evaluate(self, ctx: ExecutionContext) -> bool:
150        for child in self.children:
151            if not await child.evaluate(ctx):
152                return False
153        return True
154
155    def extra_fields(self) -> dict[str, Any]:
156        return {"children": [c.to_dict() for c in self.children]}
157
158
159@register
160class Or(Expression[bool]):
161    """True if any child expression evaluates true. Short-circuits."""
162
163    type_tag: ClassVar[str] = "expr:or"
164
165    def __init__(
166        self,
167        children: list[Expression[bool]],
168        *,
169        user_name: str = "",
170        node_id: str | None = None,
171    ) -> None:
172        super().__init__(user_name=user_name, node_id=node_id)
173        self.children = list(children)
174
175    async def evaluate(self, ctx: ExecutionContext) -> bool:
176        for child in self.children:
177            if await child.evaluate(ctx):
178                return True
179        return False
180
181    def extra_fields(self) -> dict[str, Any]:
182        return {"children": [c.to_dict() for c in self.children]}
183
184
185@register
186class Not(Expression[bool]):
187    """Negates its child expression."""
188
189    type_tag: ClassVar[str] = "expr:not"
190
191    def __init__(
192        self,
193        child: Expression[bool],
194        *,
195        user_name: str = "",
196        node_id: str | None = None,
197    ) -> None:
198        super().__init__(user_name=user_name, node_id=node_id)
199        self.child = child
200
201    async def evaluate(self, ctx: ExecutionContext) -> bool:
202        return not await self.child.evaluate(ctx)
203
204    def extra_fields(self) -> dict[str, Any]:
205        return {"child": self.child.to_dict()}