jmri_core.persistence.signal_systems

  1"""``SignalSystemLoader`` — parse JMRI's ``xml/signals/`` corpus.
  2
  3JMRI ships ~50 signal systems under ``<install>/xml/signals/<name>/``
  4— each is a directory containing an ``aspects.xml`` (the aspect
  5table for the system) plus one or more ``appearance-*.xml`` files
  6(per-mast-type appearance tables). This loader handles the first
  7half: it reads every ``aspects.xml`` and returns a
  8:class:`SignalSystem` per directory.
  9
 10Appearance files, images, and the ``aspectMappings`` transition
 11graphs are deferred with Panel Editor (Tier 1 #3). What we ship
 12here is enough to:
 13
 14* enumerate the systems the operator has available so mobile
 15  clients / a picker UI can present a valid choice;
 16* return the valid aspect names for a given system so a
 17  :class:`~jmri_core.beans.signal_mast.SignalMast` can validate
 18  ``set_aspect`` calls;
 19* render the aspect table in a read-only web view.
 20
 21XML shape (checked against JMRI's ``xml/schema/aspecttable.xsd`` and
 22sampled files from ``basic``, ``AAR-1946``, ``B&O-1957``)::
 23
 24    <aspecttable>
 25      <name>...</name>
 26      <reference>...</reference>
 27      <aspects>
 28        <aspect>
 29          <name>Clear</name>
 30          <indication>Proceed</indication>
 31          <speed>Normal</speed>
 32          <speed2>Normal</speed2>          <!-- optional -->
 33          <route>Diverging</route>         <!-- optional -->
 34        </aspect>
 35        ...
 36      </aspects>
 37    </aspecttable>
 38"""
 39from __future__ import annotations
 40
 41import logging
 42import pathlib
 43import xml.etree.ElementTree as ET
 44from dataclasses import dataclass, field
 45
 46_log = logging.getLogger(__name__)
 47
 48
 49@dataclass(frozen=True, slots=True)
 50class SignalAspect:
 51    """One aspect within a signal system.
 52
 53    ``route`` is JMRI's flag distinguishing diverging aspects from
 54    straight-route ones. Real XML uses the value ``Diverging`` when
 55    set; other values are preserved as-is.
 56    """
 57
 58    name: str
 59    indication: str = ""
 60    speed: str = ""
 61    speed2: str = ""
 62    route: str = ""
 63
 64
 65@dataclass(frozen=True, slots=True)
 66class SignalSystem:
 67    """One signal system loaded from ``aspects.xml``.
 68
 69    ``id`` is the directory name (e.g. ``"basic"`` or ``"AAR-1946"``);
 70    ``name`` and ``reference`` come from the XML. ``aspects`` is
 71    presented in file order — JMRI callers depend on the order
 72    matching what a signaller expects (Clear first, Stop somewhere
 73    down the list, etc.).
 74    """
 75
 76    id: str
 77    name: str = ""
 78    reference: str = ""
 79    aspects: tuple[SignalAspect, ...] = field(default_factory=tuple)
 80
 81
 82class SignalSystemLoader:
 83    """Scans a signal-XML root once and caches the result.
 84
 85    Typical use::
 86
 87        loader = SignalSystemLoader(pathlib.Path("/opt/JMRI/xml/signals"))
 88        systems = loader.systems()
 89        basic = loader.get("basic")
 90
 91    ``root`` may be ``None`` (unset env var, missing install) — in
 92    that case :meth:`systems` returns an empty tuple and :meth:`get`
 93    returns ``None`` without raising. That's the state the
 94    ``signalSystem`` service reflects as 503 to clients.
 95    """
 96
 97    def __init__(self, root: pathlib.Path | None) -> None:
 98        self._root = root
 99        self._cache: dict[str, SignalSystem] | None = None
100
101    @property
102    def root(self) -> pathlib.Path | None:
103        return self._root
104
105    def systems(self) -> tuple[SignalSystem, ...]:
106        return tuple(self._loaded().values())
107
108    def get(self, system_id: str) -> SignalSystem | None:
109        return self._loaded().get(system_id)
110
111    def reload(self) -> None:
112        """Drop the cache so the next access re-reads the XML.
113
114        Useful for operators editing signal XML in place; not called
115        automatically because the cache should be stable for
116        performance reasons.
117        """
118        self._cache = None
119
120    # ---- internals ------------------------------------------------------
121
122    def _loaded(self) -> dict[str, SignalSystem]:
123        cache = self._cache
124        if cache is not None:
125            return cache
126        cache = {}
127        root = self._root
128        if root is not None and root.is_dir():
129            for system_dir in sorted(
130                p for p in root.iterdir() if p.is_dir()
131            ):
132                system = _parse_system(system_dir)
133                if system is not None:
134                    cache[system.id] = system
135        self._cache = cache
136        return cache
137
138
139# ---- XML parsing --------------------------------------------------------
140
141
142def _parse_system(system_dir: pathlib.Path) -> SignalSystem | None:
143    """Read ``<system_dir>/aspects.xml``. Missing / malformed → None
144    with a warning logged. Namespaced elements are ignored (docbook
145    ``copyright`` etc.); we only care about the aspect table itself.
146    """
147    aspects_xml = system_dir / "aspects.xml"
148    if not aspects_xml.is_file():
149        return None
150    try:
151        tree = ET.parse(aspects_xml)
152    except (ET.ParseError, OSError) as e:
153        _log.warning("signal system %s: %s", aspects_xml, e)
154        return None
155    root = tree.getroot()
156    if _localname(root.tag) != "aspecttable":
157        _log.warning(
158            "signal system %s: root is <%s>, expected <aspecttable>",
159            aspects_xml, root.tag,
160        )
161        return None
162
163    name = _first_text(root, "name")
164    reference = _first_text(root, "reference")
165    aspects = tuple(_parse_aspects(root))
166    return SignalSystem(
167        id=system_dir.name,
168        name=name,
169        reference=reference,
170        aspects=aspects,
171    )
172
173
174def _parse_aspects(root: ET.Element) -> list[SignalAspect]:
175    """Extract each ``<aspect>`` element from the ``<aspects>`` container.
176
177    Missing sub-elements default to empty strings — JMRI itself is
178    lax about them (some legacy files omit ``speed2``, some pre-2011
179    files omit ``route``).
180    """
181    out: list[SignalAspect] = []
182    aspects_el = _find_child(root, "aspects")
183    if aspects_el is None:
184        return out
185    for aspect_el in aspects_el:
186        if _localname(aspect_el.tag) != "aspect":
187            continue
188        aspect_name = _first_text(aspect_el, "name")
189        if not aspect_name:
190            # An aspect without a name is meaningless; skip rather
191            # than emit a nameless entry that the UI can't render.
192            continue
193        out.append(SignalAspect(
194            name=aspect_name,
195            indication=_first_text(aspect_el, "indication"),
196            speed=_first_text(aspect_el, "speed"),
197            speed2=_first_text(aspect_el, "speed2"),
198            route=_first_text(aspect_el, "route"),
199        ))
200    return out
201
202
203def _localname(tag: str) -> str:
204    """Strip an XML namespace prefix. ``{http://ns}name`` → ``name``."""
205    if tag.startswith("{"):
206        return tag.split("}", 1)[1]
207    return tag
208
209
210def _find_child(parent: ET.Element, name: str) -> ET.Element | None:
211    for child in parent:
212        if _localname(child.tag) == name:
213            return child
214    return None
215
216
217def _first_text(parent: ET.Element, name: str) -> str:
218    child = _find_child(parent, name)
219    if child is None or child.text is None:
220        return ""
221    return child.text.strip()