jmri_logixng.manager

  1"""``LogixNGManager`` — profile-scoped registry of :class:`LogixNG` programs.
  2
  3Programs live on disk as YAML files under ``<profile>/logixng/``.
  4Filenames are the program's system name plus ``.yaml``. The manager
  5provides:
  6
  7* ``list()`` / iteration
  8* ``get(system_name)``
  9* ``create(system_name, ...)``
 10* ``put(system_name, logixng)`` — replace by name
 11* ``delete(system_name)``
 12* ``save_all(root)`` / ``load_all(root)``
 13
 14Signals ``added`` / ``removed`` / ``updated`` fire on mutations so the
 15JSON push side and any future UI can react without polling.
 16"""
 17from __future__ import annotations
 18
 19import logging
 20import pathlib
 21import re
 22from collections.abc import Iterator
 23
 24from jmri_core.events import Signal
 25
 26from jmri_logixng.logix import LogixNG
 27from jmri_logixng.serialize import from_yaml, load_yaml, save_yaml
 28
 29_log = logging.getLogger(__name__)
 30
 31# JMRI Java's convention for LogixNG system names is ``IQ`` (LogixNG
 32# root) + digits, distinct from the ``IX`` Logix (legacy) prefix. We
 33# follow it so system names round-trip against real JMRI XML. Callers
 34# using scripts can also pass any string — validation is deliberately
 35# lax (a user-friendly ``IQ:MyRoutine`` is fine).
 36_DEFAULT_PREFIX = "IQ"
 37
 38# Filenames are only allowed the same character class as JMRI system
 39# names to keep the ``<system_name>.yaml`` mapping unambiguous. Slashes
 40# and other separators would be a path-traversal risk on disk. Colons
 41# in particular are excluded because Windows won't allow them as file
 42# names — the save silently truncates ``"IQ:x.yaml"`` to ``"IQ"``.
 43# Colons are permitted because JMRI Java's LogixNG system names use
 44# them (``IQ:AUTO:0001``); a name imported from a JMRI XML would be
 45# rejected without them. Slashes and spaces still excluded — those
 46# would break file-per-program persistence.
 47_SAFE_NAME_RE = re.compile(r"^[A-Za-z0-9_.:-]+$")
 48
 49
 50class LogixNGManager:
 51    """Registry of :class:`LogixNG` programs, one per profile.
 52
 53    Not a :class:`~jmri_core.manager.Manager` — LogixNG isn't a
 54    NamedBean and the system-name shape is program-oriented rather
 55    than address-oriented. The signal / index shape mirrors Manager
 56    closely so callers can subscribe in the same pattern.
 57    """
 58
 59    def __init__(self, prefix: str = _DEFAULT_PREFIX) -> None:
 60        self.prefix = prefix
 61        self._by_name: dict[str, LogixNG] = {}
 62        self.added: Signal[LogixNG] = Signal()
 63        self.removed: Signal[LogixNG] = Signal()
 64        self.updated: Signal[LogixNG] = Signal()
 65
 66    # ---- read ------------------------------------------------------------
 67
 68    def __iter__(self) -> Iterator[LogixNG]:
 69        return iter(self._by_name.values())
 70
 71    def __len__(self) -> int:
 72        return len(self._by_name)
 73
 74    def __contains__(self, name: str) -> bool:
 75        return name in self._by_name
 76
 77    def get(self, system_name: str) -> LogixNG | None:
 78        return self._by_name.get(system_name)
 79
 80    def names(self) -> list[str]:
 81        return list(self._by_name)
 82
 83    # ---- mutate ---------------------------------------------------------
 84
 85    async def put(self, system_name: str, logixng: LogixNG) -> LogixNG:
 86        """Insert or replace *logixng* under *system_name*. Fires
 87        ``added`` on a new entry and ``updated`` on a replace."""
 88        if not _SAFE_NAME_RE.match(system_name):
 89            raise ValueError(
 90                f"logixng system name {system_name!r} must match "
 91                f"[A-Za-z0-9_.:-]+"
 92            )
 93        existing = self._by_name.get(system_name)
 94        self._by_name[system_name] = logixng
 95        if existing is None:
 96            await self.added.emit(logixng)
 97        else:
 98            await self.updated.emit(logixng)
 99        return logixng
100
101    async def create(
102        self, system_name: str, *, user_name: str = "",
103        enabled: bool = True,
104    ) -> LogixNG:
105        """Create an empty LogixNG program at *system_name*. Idempotent
106        by name — returns the existing program if one already lives
107        there (matches how the base bean managers behave)."""
108        if system_name in self._by_name:
109            return self._by_name[system_name]
110        prog = LogixNG(user_name=user_name, enabled=enabled)
111        return await self.put(system_name, prog)
112
113    async def delete(self, system_name: str) -> bool:
114        """Remove the program at *system_name*. Returns True if one
115        was found. Fires ``removed``."""
116        prog = self._by_name.pop(system_name, None)
117        if prog is None:
118            return False
119        await self.removed.emit(prog)
120        return True
121
122    # ---- disk ----------------------------------------------------------
123
124    def save_all(self, root: pathlib.Path) -> None:
125        """Write every program to ``root/<system_name>.yaml``.
126
127        Creates ``root`` if absent. Overwrites any existing files;
128        stale files (programs deleted since the last save) are
129        removed so the on-disk set matches memory.
130        """
131        root.mkdir(parents=True, exist_ok=True)
132        keep: set[pathlib.Path] = set()
133        for system_name, prog in self._by_name.items():
134            path = root / f"{system_name}.yaml"
135            save_yaml(prog, path)
136            keep.add(path)
137        for existing in root.glob("*.yaml"):
138            if existing not in keep:
139                try:
140                    existing.unlink()
141                except OSError as e:
142                    _log.warning(
143                        "logixng: couldn't remove stale %s (%s)", existing, e,
144                    )
145
146    def save_one(self, root: pathlib.Path, system_name: str) -> None:
147        """Persist a single program without touching siblings. Cheaper
148        than :meth:`save_all` when only one entry changed."""
149        prog = self._by_name.get(system_name)
150        if prog is None:
151            return
152        root.mkdir(parents=True, exist_ok=True)
153        save_yaml(prog, root / f"{system_name}.yaml")
154
155    def delete_on_disk(self, root: pathlib.Path, system_name: str) -> None:
156        """Remove ``<root>/<system_name>.yaml`` if it exists. Used by
157        the service's ``delete`` handler so the file goes away when the
158        in-memory entry does."""
159        path = root / f"{system_name}.yaml"
160        if path.exists():
161            try:
162                path.unlink()
163            except OSError as e:
164                _log.warning(
165                    "logixng: couldn't unlink %s (%s)", path, e,
166                )
167
168    async def load_all(self, root: pathlib.Path) -> int:
169        """Populate the manager from ``root/*.yaml``.
170
171        Returns the number of programs loaded. Files that fail to
172        parse are logged and skipped — one broken file doesn't
173        prevent the rest from loading.
174        """
175        if not root.is_dir():
176            return 0
177        count = 0
178        for yaml_path in sorted(root.glob("*.yaml")):
179            system_name = yaml_path.stem
180            if not _SAFE_NAME_RE.match(system_name):
181                _log.warning(
182                    "logixng: skipping %s (unsafe filename)", yaml_path,
183                )
184                continue
185            try:
186                node = load_yaml(yaml_path)
187            except Exception as e:  # noqa: BLE001 — one bad file mustn't block the rest
188                # YAML parse errors, decoder failures, IO errors — all
189                # get logged and skipped. The manager keeps loading
190                # siblings so one broken file doesn't nuke the whole
191                # set.
192                _log.warning(
193                    "logixng: skipping %s (%s)", yaml_path, e,
194                )
195                continue
196            if not isinstance(node, LogixNG):
197                _log.warning(
198                    "logixng: %s isn't a LogixNG root, skipping", yaml_path,
199                )
200                continue
201            await self.put(system_name, node)
202            count += 1
203        return count
204
205    async def load_yaml_text(
206        self, system_name: str, text: str,
207    ) -> LogixNG:
208        """Parse *text* as a LogixNG YAML program and register it at
209        *system_name*. Convenient for JSON-service uploads that ship
210        the tree inline."""
211        node = from_yaml(text)
212        if not isinstance(node, LogixNG):
213            raise ValueError("payload is not a LogixNG root")
214        return await self.put(system_name, node)