jmri_server.config_store

  1"""``ConfigStore`` — server-owned ``Configuration.xml`` bean persistence.
  2
  3Holds the parsed bean tables (turnouts, sensors, lights) from the
  4``layout-config`` Configuration.xml file in the active profile. Other
  5sections (signal heads, memories, routes, blocks, panels, LogixNG)
  6are preserved verbatim so users can keep their existing JMRI configs
  7without losing untouched data when our web UI saves changes.
  8
  9Storage policy:
 10
 11* One Configuration.xml at ``Profile.config_file``.
 12* On startup: read the file once, populate in-memory state.
 13* On every bean mutation: rewrite the whole file.
 14* Atomic write (write to ``.tmp``, rename) so a crash mid-write can
 15  never lose data.
 16* In-memory state is the source of truth during a server session;
 17  the XML file is the source of truth at startup.
 18"""
 19from __future__ import annotations
 20
 21import logging
 22import os
 23import pathlib
 24from collections.abc import Iterable
 25from dataclasses import dataclass, field
 26
 27from jmri_core.persistence import (
 28    ConfigFile,
 29    LightRow,
 30    LightSection,
 31    SensorRow,
 32    SensorSection,
 33    TurnoutRow,
 34    TurnoutSection,
 35    empty_config,
 36    light_mgr_class_for_prefix,
 37    load_config,
 38    save_config_bytes,
 39    sensor_mgr_class_for_prefix,
 40    turnout_mgr_class_for_prefix,
 41)
 42from jmri_core.persistence.profile import Profile
 43
 44_log = logging.getLogger(__name__)
 45
 46# Rows are grouped by system-name prefix at flush time so each JMRI-style
 47# ``<turnouts class="...">`` block carries a *concrete* driver-specific
 48# manager class (e.g. ``NceTurnoutManagerXml``). The abstract class names
 49# (``AbstractTurnoutManagerConfigXML`` etc.) that used to live here are
 50# schema-invalid — they end in uppercase ``L`` and JMRI Java refuses to
 51# parse them at startup.
 52
 53
 54def _system_prefix(system_name: str) -> str:
 55    """The JMRI system-prefix character — first letter of the system name."""
 56    return system_name[:1].upper() if system_name else "I"
 57
 58
 59def _group_by_prefix[R](rows: Iterable[R]) -> list[tuple[str, list[R]]]:
 60    """Group *rows* by their system-name prefix.
 61
 62    Returns ``[(prefix, sorted_rows), ...]`` with stable ordering
 63    (prefixes alphabetical, rows sorted by system_name within each
 64    group). One returned tuple maps to one ``<turnouts>`` / ``<sensors>``
 65    / ``<lights>`` block in the emitted XML — JMRI writes one block per
 66    connection prefix, and matching that layout keeps files loadable in
 67    JMRI Java.
 68    """
 69    buckets: dict[str, list[R]] = {}
 70    for row in rows:
 71        buckets.setdefault(_system_prefix(row.system_name), []).append(row)  # type: ignore[attr-defined]
 72    return [
 73        (prefix, sorted(buckets[prefix], key=lambda r: r.system_name))  # type: ignore[attr-defined]
 74        for prefix in sorted(buckets)
 75    ]
 76
 77
 78@dataclass
 79class ConfigStore:
 80    """In-memory + on-disk Configuration.xml."""
 81
 82    profile: Profile
 83    _turnouts: dict[str, TurnoutRow] = field(default_factory=dict)
 84    _sensors: dict[str, SensorRow] = field(default_factory=dict)
 85    _lights: dict[str, LightRow] = field(default_factory=dict)
 86    _preserved_doctype: str | None = None
 87    _preserved_root_attribs: dict[str, str] = field(default_factory=dict)
 88    _preserved_elements: list = field(default_factory=list)
 89
 90    def __post_init__(self) -> None:
 91        self.profile.ensure()
 92        self._load_existing()
 93
 94    def _load_existing(self) -> None:
 95        """Load the on-disk Configuration.xml into memory, if it exists."""
 96        path = self.profile.config_file
 97        if not path.exists():
 98            cfg = empty_config()
 99            self._preserved_doctype = cfg.doctype
100            return
101        try:
102            cfg = load_config(path)
103        except Exception as e:  # noqa: BLE001
104            _log.warning(
105                "ConfigStore: failed to parse %s (%s); starting empty", path, e
106            )
107            self._preserved_doctype = empty_config().doctype
108            return
109        for row in cfg.all_turnouts():
110            self._turnouts[row.system_name] = row
111        for row in cfg.all_sensors():
112            self._sensors[row.system_name] = row
113        for row in cfg.all_lights():
114            self._lights[row.system_name] = row
115        self._preserved_doctype = cfg.doctype
116        self._preserved_root_attribs = dict(cfg.root_attribs)
117        self._preserved_elements = list(cfg.preserved)
118
119    # ----- read --------------------------------------------------------
120
121    def turnouts(self) -> Iterable[TurnoutRow]:
122        return self._turnouts.values()
123
124    def sensors(self) -> Iterable[SensorRow]:
125        return self._sensors.values()
126
127    def lights(self) -> Iterable[LightRow]:
128        return self._lights.values()
129
130    # ----- write -------------------------------------------------------
131
132    def upsert_turnout(self, row: TurnoutRow) -> None:
133        # Preserve unmodeled extras (locked / lockMode / decoder etc.)
134        # from any existing row — the bean services don't carry these
135        # fields and shouldn't accidentally erase them on edit.
136        prev = self._turnouts.get(row.system_name)
137        if prev and not row.extra_attribs and prev.extra_attribs:
138            row.extra_attribs = dict(prev.extra_attribs)
139        if prev and row.comment is None and prev.comment:
140            row.comment = prev.comment
141        self._turnouts[row.system_name] = row
142        self._flush()
143
144    def upsert_sensor(self, row: SensorRow) -> None:
145        prev = self._sensors.get(row.system_name)
146        if prev and not row.extra_attribs and prev.extra_attribs:
147            row.extra_attribs = dict(prev.extra_attribs)
148        if prev and row.comment is None and prev.comment:
149            row.comment = prev.comment
150        self._sensors[row.system_name] = row
151        self._flush()
152
153    def upsert_light(self, row: LightRow) -> None:
154        prev = self._lights.get(row.system_name)
155        if prev and not row.extra_attribs and prev.extra_attribs:
156            row.extra_attribs = dict(prev.extra_attribs)
157        if prev and row.comment is None and prev.comment:
158            row.comment = prev.comment
159        self._lights[row.system_name] = row
160        self._flush()
161
162    def remove_turnout(self, system_name: str) -> bool:
163        existed = self._turnouts.pop(system_name, None) is not None
164        if existed:
165            self._flush()
166        return existed
167
168    def remove_sensor(self, system_name: str) -> bool:
169        existed = self._sensors.pop(system_name, None) is not None
170        if existed:
171            self._flush()
172        return existed
173
174    def remove_light(self, system_name: str) -> bool:
175        existed = self._lights.pop(system_name, None) is not None
176        if existed:
177            self._flush()
178        return existed
179
180    # ----- I/O ---------------------------------------------------------
181
182    def _build_config(self) -> ConfigFile:
183        return ConfigFile(
184            root_attribs=dict(self._preserved_root_attribs),
185            doctype=self._preserved_doctype,
186            turnout_sections=self._build_turnout_sections(),
187            sensor_sections=self._build_sensor_sections(),
188            light_sections=self._build_light_sections(),
189            preserved=list(self._preserved_elements),
190        )
191
192    def _build_turnout_sections(self) -> list[TurnoutSection]:
193        return [
194            TurnoutSection(
195                manager_class=turnout_mgr_class_for_prefix(prefix),
196                rows=rows,
197            )
198            for prefix, rows in _group_by_prefix(self._turnouts.values())
199        ]
200
201    def _build_sensor_sections(self) -> list[SensorSection]:
202        return [
203            SensorSection(
204                manager_class=sensor_mgr_class_for_prefix(prefix),
205                rows=rows,
206            )
207            for prefix, rows in _group_by_prefix(self._sensors.values())
208        ]
209
210    def _build_light_sections(self) -> list[LightSection]:
211        return [
212            LightSection(
213                manager_class=light_mgr_class_for_prefix(prefix),
214                rows=rows,
215            )
216            for prefix, rows in _group_by_prefix(self._lights.values())
217        ]
218
219    def _flush(self) -> None:
220        """Atomically rewrite the file. Failures are logged, not raised."""
221        path = self.profile.config_file
222        try:
223            data = save_config_bytes(self._build_config())
224            tmp = path.with_suffix(path.suffix + ".tmp")
225            tmp.write_bytes(data)
226            os.replace(tmp, path)
227        except OSError as e:
228            _log.warning("ConfigStore: failed to write %s: %s", path, e)
229
230    # ----- introspection ----------------------------------------------
231
232    def __len__(self) -> int:
233        return len(self._turnouts) + len(self._sensors) + len(self._lights)
234
235    def path(self) -> pathlib.Path:
236        return self.profile.config_file