jmri_logixng.triggers

  1"""Trigger binding — fire a :class:`ConditionalNG` when watched beans change.
  2
  3A trigger is a small adapter that:
  4
  51. Subscribes to the ``state_changed`` Signal of every watched bean.
  62. On any change, evaluates an optional ``guard`` expression.
  73. If the guard returns true (or is absent), executes the conditional.
  8
  9Triggers are constructed and bound via :class:`AutomationEngine`, which
 10holds a list of triggers and the disconnect callbacks needed to tear
 11them down cleanly on shutdown.
 12"""
 13from __future__ import annotations
 14
 15import asyncio
 16from collections.abc import Callable
 17from dataclasses import dataclass, field
 18from typing import Any, Literal
 19
 20from jmri_logixng.context import ExecutionContext
 21from jmri_logixng.logix import ConditionalNG
 22from jmri_logixng.tree import Expression
 23
 24BeanKind = Literal["sensor", "turnout", "light", "memory", "block", "reporter"]
 25
 26
 27@dataclass
 28class TriggerBinding:
 29    """One trigger — watched beans + optional guard + the conditional to run."""
 30
 31    conditional: ConditionalNG
 32    watches: list[tuple[BeanKind, str]]  # (kind, system_name)
 33    guard: Expression[bool] | None = None
 34    user_name: str = ""
 35
 36
 37@dataclass
 38class AutomationEngine:
 39    """Runs a set of triggers against an :class:`ExecutionContext`."""
 40
 41    ctx: ExecutionContext
 42    triggers: list[TriggerBinding] = field(default_factory=list)
 43    _disconnects: list[Callable[[], None]] = field(default_factory=list)
 44
 45    def add(self, trigger: TriggerBinding) -> None:
 46        self.triggers.append(trigger)
 47
 48    def start(self) -> None:
 49        """Wire every trigger's watched beans up to the firing handler."""
 50        for trigger in self.triggers:
 51            self._wire(trigger)
 52
 53    def stop(self) -> None:
 54        """Disconnect every listener. Idempotent."""
 55        for disconnect in self._disconnects:
 56            disconnect()
 57        self._disconnects.clear()
 58
 59    def _wire(self, trigger: TriggerBinding) -> None:
 60        handler = self._make_handler(trigger)
 61        for kind, system_name in trigger.watches:
 62            bean = self._lookup_bean(kind, system_name)
 63            if bean is None:
 64                self.ctx.log.warning(
 65                    "trigger %r: no %s named %r — skipping watch",
 66                    trigger.user_name or trigger.conditional.id, kind, system_name,
 67                )
 68                continue
 69            self._disconnects.append(bean.state_changed.connect(handler))
 70
 71    def _make_handler(self, trigger: TriggerBinding) -> Callable[[Any], Any]:
 72        async def handle(_payload: Any) -> None:
 73            if trigger.guard is not None:
 74                guard_result = await trigger.guard.evaluate(self.ctx)
 75                if not guard_result:
 76                    self.ctx.emit_trace(
 77                        f"trigger {trigger.user_name or trigger.conditional.id}: guard false"
 78                    )
 79                    return
 80            self.ctx.emit_trace(
 81                f"trigger {trigger.user_name or trigger.conditional.id}: firing"
 82            )
 83            await trigger.conditional.execute(self.ctx)
 84        return handle
 85
 86    def _lookup_bean(self, kind: BeanKind, system_name: str) -> Any:
 87        # Each manager has a ``.get(name)`` method but they're different
 88        # generic types; an Any-typed bag keeps the union friendly.
 89        managers: dict[str, Any] = {
 90            "sensor": self.ctx.connection.sensors,
 91            "turnout": self.ctx.connection.turnouts,
 92            "light": self.ctx.connection.lights,
 93            "memory": self.ctx.connection.memories,
 94            "block": self.ctx.connection.blocks,
 95            "reporter": self.ctx.connection.reporters,
 96        }
 97        manager = managers.get(kind)
 98        if manager is None:
 99            return None
100        return manager.get(system_name)
101
102
103# Touch unused imports so static analyzers don't flag them.
104_ = asyncio