jmri_logixng.logix

 1"""``LogixNG`` and ``ConditionalNG`` containers.
 2
 3* :class:`ConditionalNG` runs a sequence of actions when invoked.
 4* :class:`LogixNG` is a named collection of conditionals — JMRI's
 5  user-facing organizational unit. Triggering one conditional is
 6  independent of the others; the LogixNG container is mainly for
 7  serialization grouping and enable/disable in bulk.
 8"""
 9from __future__ import annotations
10
11from typing import Any, ClassVar
12
13from jmri_logixng.context import ExecutionContext
14from jmri_logixng.registry import register
15from jmri_logixng.tree import Action, Node
16
17
18@register
19class ConditionalNG(Action):
20    """A sequence of root actions executed in order."""
21
22    type_tag: ClassVar[str] = "logixng:conditional"
23
24    def __init__(
25        self,
26        actions: list[Action] | None = None,
27        *,
28        user_name: str = "",
29        node_id: str | None = None,
30        enabled: bool = True,
31    ) -> None:
32        super().__init__(user_name=user_name, node_id=node_id)
33        self.actions: list[Action] = list(actions or [])
34        self.enabled = enabled
35
36    async def execute(self, ctx: ExecutionContext) -> None:
37        if not self.enabled:
38            ctx.emit_trace(f"conditional {self.id!r} skipped (disabled)")
39            return
40        ctx.emit_trace(f"conditional {self.id!r} executing {len(self.actions)} actions")
41        for action in self.actions:
42            await action.execute(ctx)
43
44    def extra_fields(self) -> dict[str, Any]:
45        return {
46            "enabled": self.enabled,
47            "actions": [a.to_dict() for a in self.actions],
48        }
49
50
51@register
52class LogixNG(Node):
53    """A named collection of :class:`ConditionalNG` instances."""
54
55    type_tag: ClassVar[str] = "logixng:root"
56
57    def __init__(
58        self,
59        conditionals: list[ConditionalNG] | None = None,
60        *,
61        user_name: str = "",
62        node_id: str | None = None,
63        enabled: bool = True,
64    ) -> None:
65        super().__init__(user_name=user_name, node_id=node_id)
66        self.conditionals: list[ConditionalNG] = list(conditionals or [])
67        self.enabled = enabled
68
69    def add(self, conditional: ConditionalNG) -> None:
70        self.conditionals.append(conditional)
71
72    async def execute_all(self, ctx: ExecutionContext) -> None:
73        """Execute every conditional in order (for fire-and-forget runs)."""
74        if not self.enabled:
75            return
76        for c in self.conditionals:
77            await c.execute(ctx)
78
79    def extra_fields(self) -> dict[str, Any]:
80        return {
81            "enabled": self.enabled,
82            "conditionals": [c.to_dict() for c in self.conditionals],
83        }