jmri_logixng

64class Action(Node):
65    """A node that *does* something when executed."""
66
67    @abstractmethod
68    async def execute(self, ctx: ExecutionContext) -> None: ...

 38@dataclass
 39class AutomationEngine:
 40    """Runs a set of triggers against an :class:`ExecutionContext`."""
 41
 42    ctx: ExecutionContext
 43    triggers: list[TriggerBinding] = field(default_factory=list)
 44    _disconnects: list[Callable[[], None]] = field(default_factory=list)
 45
 46    def add(self, trigger: TriggerBinding) -> None:
 47        self.triggers.append(trigger)
 48
 49    def start(self) -> None:
 50        """Wire every trigger's watched beans up to the firing handler."""
 51        for trigger in self.triggers:
 52            self._wire(trigger)
 53
 54    def stop(self) -> None:
 55        """Disconnect every listener. Idempotent."""
 56        for disconnect in self._disconnects:
 57            disconnect()
 58        self._disconnects.clear()
 59
 60    def _wire(self, trigger: TriggerBinding) -> None:
 61        handler = self._make_handler(trigger)
 62        for kind, system_name in trigger.watches:
 63            bean = self._lookup_bean(kind, system_name)
 64            if bean is None:
 65                self.ctx.log.warning(
 66                    "trigger %r: no %s named %r — skipping watch",
 67                    trigger.user_name or trigger.conditional.id, kind, system_name,
 68                )
 69                continue
 70            self._disconnects.append(bean.state_changed.connect(handler))
 71
 72    def _make_handler(self, trigger: TriggerBinding) -> Callable[[Any], Any]:
 73        async def handle(_payload: Any) -> None:
 74            if trigger.guard is not None:
 75                guard_result = await trigger.guard.evaluate(self.ctx)
 76                if not guard_result:
 77                    self.ctx.emit_trace(
 78                        f"trigger {trigger.user_name or trigger.conditional.id}: guard false"
 79                    )
 80                    return
 81            self.ctx.emit_trace(
 82                f"trigger {trigger.user_name or trigger.conditional.id}: firing"
 83            )
 84            await trigger.conditional.execute(self.ctx)
 85        return handle
 86
 87    def _lookup_bean(self, kind: BeanKind, system_name: str) -> Any:
 88        # Each manager has a ``.get(name)`` method but they're different
 89        # generic types; an Any-typed bag keeps the union friendly.
 90        managers: dict[str, Any] = {
 91            "sensor": self.ctx.connection.sensors,
 92            "turnout": self.ctx.connection.turnouts,
 93            "light": self.ctx.connection.lights,
 94            "memory": self.ctx.connection.memories,
 95            "block": self.ctx.connection.blocks,
 96            "reporter": self.ctx.connection.reporters,
 97        }
 98        manager = managers.get(kind)
 99        if manager is None:
100            return None
101        return manager.get(system_name)

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]}

19@register
20class ConditionalNG(Action):
21    """A sequence of root actions executed in order."""
22
23    type_tag: ClassVar[str] = "logixng:conditional"
24
25    def __init__(
26        self,
27        actions: list[Action] | None = None,
28        *,
29        user_name: str = "",
30        node_id: str | None = None,
31        enabled: bool = True,
32    ) -> None:
33        super().__init__(user_name=user_name, node_id=node_id)
34        self.actions: list[Action] = list(actions or [])
35        self.enabled = enabled
36
37    async def execute(self, ctx: ExecutionContext) -> None:
38        if not self.enabled:
39            ctx.emit_trace(f"conditional {self.id!r} skipped (disabled)")
40            return
41        ctx.emit_trace(f"conditional {self.id!r} executing {len(self.actions)} actions")
42        for action in self.actions:
43            await action.execute(ctx)
44
45    def extra_fields(self) -> dict[str, Any]:
46        return {
47            "enabled": self.enabled,
48            "actions": [a.to_dict() for a in self.actions],
49        }

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}

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}

20@dataclass
21class ExecutionContext:
22    """Runtime context passed through every node visit."""
23
24    connection: SystemConnection
25    log: logging.Logger = field(default_factory=lambda: logging.getLogger("jmri_logixng"))
26    trace_enabled: bool = False
27    trace: list[str] = field(default_factory=list)
28
29    def emit_trace(self, message: str) -> None:
30        if self.trace_enabled:
31            self.trace.append(message)

71class Expression[T](Node):
72    """A node that evaluates to a value of type *T*.
73
74    The most common variant is ``Expression[bool]``. Concrete subclasses
75    implement :meth:`evaluate`.
76    """
77
78    @abstractmethod
79    async def evaluate(self, ctx: ExecutionContext) -> T: ...

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

103@dataclass(slots=True)
104class ImportResult:
105    """Outcome of :func:`import_jmri_logixng_xml`."""
106
107    programs: list[LogixNG] = field(default_factory=list)
108    written: list[pathlib.Path] = field(default_factory=list)
109    warnings: list[ImportWarning] = field(default_factory=list)

 94@dataclass(frozen=True, slots=True)
 95class ImportWarning:
 96    """One diagnostic the importer recorded during a run."""
 97
 98    kind: str          # e.g. "unknown-tag", "non-direct-addressing"
 99    detail: str        # human-readable specifics
100    system_name: str   # the JMRI system name that produced it

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}

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}

52@register
53class LogixNG(Node):
54    """A named collection of :class:`ConditionalNG` instances."""
55
56    type_tag: ClassVar[str] = "logixng:root"
57
58    def __init__(
59        self,
60        conditionals: list[ConditionalNG] | None = None,
61        *,
62        user_name: str = "",
63        node_id: str | None = None,
64        enabled: bool = True,
65    ) -> None:
66        super().__init__(user_name=user_name, node_id=node_id)
67        self.conditionals: list[ConditionalNG] = list(conditionals or [])
68        self.enabled = enabled
69
70    def add(self, conditional: ConditionalNG) -> None:
71        self.conditionals.append(conditional)
72
73    async def execute_all(self, ctx: ExecutionContext) -> None:
74        """Execute every conditional in order (for fire-and-forget runs)."""
75        if not self.enabled:
76            return
77        for c in self.conditionals:
78            await c.execute(ctx)
79
80    def extra_fields(self) -> dict[str, Any]:
81        return {
82            "enabled": self.enabled,
83            "conditionals": [c.to_dict() for c in self.conditionals],
84        }

 51class LogixNGManager:
 52    """Registry of :class:`LogixNG` programs, one per profile.
 53
 54    Not a :class:`~jmri_core.manager.Manager` — LogixNG isn't a
 55    NamedBean and the system-name shape is program-oriented rather
 56    than address-oriented. The signal / index shape mirrors Manager
 57    closely so callers can subscribe in the same pattern.
 58    """
 59
 60    def __init__(self, prefix: str = _DEFAULT_PREFIX) -> None:
 61        self.prefix = prefix
 62        self._by_name: dict[str, LogixNG] = {}
 63        self.added: Signal[LogixNG] = Signal()
 64        self.removed: Signal[LogixNG] = Signal()
 65        self.updated: Signal[LogixNG] = Signal()
 66
 67    # ---- read ------------------------------------------------------------
 68
 69    def __iter__(self) -> Iterator[LogixNG]:
 70        return iter(self._by_name.values())
 71
 72    def __len__(self) -> int:
 73        return len(self._by_name)
 74
 75    def __contains__(self, name: str) -> bool:
 76        return name in self._by_name
 77
 78    def get(self, system_name: str) -> LogixNG | None:
 79        return self._by_name.get(system_name)
 80
 81    def names(self) -> list[str]:
 82        return list(self._by_name)
 83
 84    # ---- mutate ---------------------------------------------------------
 85
 86    async def put(self, system_name: str, logixng: LogixNG) -> LogixNG:
 87        """Insert or replace *logixng* under *system_name*. Fires
 88        ``added`` on a new entry and ``updated`` on a replace."""
 89        if not _SAFE_NAME_RE.match(system_name):
 90            raise ValueError(
 91                f"logixng system name {system_name!r} must match "
 92                f"[A-Za-z0-9_.:-]+"
 93            )
 94        existing = self._by_name.get(system_name)
 95        self._by_name[system_name] = logixng
 96        if existing is None:
 97            await self.added.emit(logixng)
 98        else:
 99            await self.updated.emit(logixng)
100        return logixng
101
102    async def create(
103        self, system_name: str, *, user_name: str = "",
104        enabled: bool = True,
105    ) -> LogixNG:
106        """Create an empty LogixNG program at *system_name*. Idempotent
107        by name — returns the existing program if one already lives
108        there (matches how the base bean managers behave)."""
109        if system_name in self._by_name:
110            return self._by_name[system_name]
111        prog = LogixNG(user_name=user_name, enabled=enabled)
112        return await self.put(system_name, prog)
113
114    async def delete(self, system_name: str) -> bool:
115        """Remove the program at *system_name*. Returns True if one
116        was found. Fires ``removed``."""
117        prog = self._by_name.pop(system_name, None)
118        if prog is None:
119            return False
120        await self.removed.emit(prog)
121        return True
122
123    # ---- disk ----------------------------------------------------------
124
125    def save_all(self, root: pathlib.Path) -> None:
126        """Write every program to ``root/<system_name>.yaml``.
127
128        Creates ``root`` if absent. Overwrites any existing files;
129        stale files (programs deleted since the last save) are
130        removed so the on-disk set matches memory.
131        """
132        root.mkdir(parents=True, exist_ok=True)
133        keep: set[pathlib.Path] = set()
134        for system_name, prog in self._by_name.items():
135            path = root / f"{system_name}.yaml"
136            save_yaml(prog, path)
137            keep.add(path)
138        for existing in root.glob("*.yaml"):
139            if existing not in keep:
140                try:
141                    existing.unlink()
142                except OSError as e:
143                    _log.warning(
144                        "logixng: couldn't remove stale %s (%s)", existing, e,
145                    )
146
147    def save_one(self, root: pathlib.Path, system_name: str) -> None:
148        """Persist a single program without touching siblings. Cheaper
149        than :meth:`save_all` when only one entry changed."""
150        prog = self._by_name.get(system_name)
151        if prog is None:
152            return
153        root.mkdir(parents=True, exist_ok=True)
154        save_yaml(prog, root / f"{system_name}.yaml")
155
156    def delete_on_disk(self, root: pathlib.Path, system_name: str) -> None:
157        """Remove ``<root>/<system_name>.yaml`` if it exists. Used by
158        the service's ``delete`` handler so the file goes away when the
159        in-memory entry does."""
160        path = root / f"{system_name}.yaml"
161        if path.exists():
162            try:
163                path.unlink()
164            except OSError as e:
165                _log.warning(
166                    "logixng: couldn't unlink %s (%s)", path, e,
167                )
168
169    async def load_all(self, root: pathlib.Path) -> int:
170        """Populate the manager from ``root/*.yaml``.
171
172        Returns the number of programs loaded. Files that fail to
173        parse are logged and skipped — one broken file doesn't
174        prevent the rest from loading.
175        """
176        if not root.is_dir():
177            return 0
178        count = 0
179        for yaml_path in sorted(root.glob("*.yaml")):
180            system_name = yaml_path.stem
181            if not _SAFE_NAME_RE.match(system_name):
182                _log.warning(
183                    "logixng: skipping %s (unsafe filename)", yaml_path,
184                )
185                continue
186            try:
187                node = load_yaml(yaml_path)
188            except Exception as e:  # noqa: BLE001 — one bad file mustn't block the rest
189                # YAML parse errors, decoder failures, IO errors — all
190                # get logged and skipped. The manager keeps loading
191                # siblings so one broken file doesn't nuke the whole
192                # set.
193                _log.warning(
194                    "logixng: skipping %s (%s)", yaml_path, e,
195                )
196                continue
197            if not isinstance(node, LogixNG):
198                _log.warning(
199                    "logixng: %s isn't a LogixNG root, skipping", yaml_path,
200                )
201                continue
202            await self.put(system_name, node)
203            count += 1
204        return count
205
206    async def load_yaml_text(
207        self, system_name: str, text: str,
208    ) -> LogixNG:
209        """Parse *text* as a LogixNG YAML program and register it at
210        *system_name*. Convenient for JSON-service uploads that ship
211        the tree inline."""
212        node = from_yaml(text)
213        if not isinstance(node, LogixNG):
214            raise ValueError("payload is not a LogixNG root")
215        return await self.put(system_name, node)