jmri_server.loco_cv_store

  1"""``LocoCvStore`` — lazy per-locomotive CV-state cache backed by disk.
  2
  3Two on-disk layouts can carry a loco's CV state:
  4
  5* **JMRI-canonical**: the ``<values>`` block embedded in
  6  ``<profile>/roster/<id>.xml`` (the locomotive-config file itself).
  7  This is what JMRI Java reads and writes.
  8* **Legacy proprietary**: ``<profile>/rosters/<id>.xml`` with our own
  9  ``<locomotive-cv-values>`` root. Used by earlier alphas.
 10
 11On read we try the canonical location first; if it has no ``<values>``
 12we fall back to the legacy one. When *both* have CV data, we raise
 13:class:`AmbiguousCvSourceError` — the UI catches that and prompts the
 14user to pick which source is authoritative. Once resolved (or on a fresh
 15loco), writes go to the location we loaded from, defaulting to the
 16JMRI-canonical embedded ``<values>`` block for interop.
 17
 18Storage policy is otherwise unchanged:
 19
 20* In-memory state is the source of truth during a session.
 21* Files are the source of truth across server restarts.
 22* Writes are atomic (tmp + rename).
 23* Failures are logged, not raised, so a permissions issue doesn't
 24  crash the programmer page mid-session.
 25"""
 26from __future__ import annotations
 27
 28import logging
 29import pathlib
 30import re
 31import shutil
 32from collections.abc import Callable
 33from dataclasses import dataclass, field
 34from enum import StrEnum
 35
 36from jmri_core.beans.loco_cv_state import LocoCvState
 37from jmri_core.persistence.loco_cv_xml import (
 38    load_cv_values_from_locomotive_config,
 39    load_loco_cv,
 40    save_cv_values_into_locomotive_config,
 41)
 42from jmri_core.persistence.profile import Profile
 43
 44# ``(loco_id) → {cv_number_str: default_value_str}``. Consulted by
 45# :meth:`LocoCvStore.set_cv` when deciding what to persist: any CV whose
 46# in-memory value equals its catalog default is *dropped* from the
 47# roster ``<values>`` block on write, per the spec (roster stores only
 48# user customisations). Returning an empty dict disables filtering for
 49# that loco (catalog not yet loaded etc.).
 50DefaultLookup = Callable[[str], dict[str, str]]
 51
 52_log = logging.getLogger(__name__)
 53_SAFE_FILENAME = re.compile(r"[^A-Za-z0-9._-]+")
 54
 55
 56class CvSource(StrEnum):
 57    """Where a loco's CV state is / should be persisted."""
 58    ROSTER   = "roster"    # JMRI-canonical: <profile>/roster/<id>.xml <values>
 59    LEGACY   = "legacy"    # our layout:      <profile>/rosters/<id>.xml
 60
 61
 62class AmbiguousCvSourceError(Exception):
 63    """Both source locations have CV data for the same loco.
 64
 65    The UI catches this and prompts the user which source is authoritative
 66    before proceeding. Once chosen, the picked source is remembered on
 67    :class:`LocoCvStore` so future accesses don't re-prompt.
 68    """
 69
 70    def __init__(
 71        self, loco_id: str,
 72        roster_path: pathlib.Path, legacy_path: pathlib.Path,
 73    ) -> None:
 74        super().__init__(
 75            f"CV state for loco {loco_id!r} exists in both "
 76            f"{roster_path} and {legacy_path}; caller must pick one via "
 77            f"LocoCvStore.set_preferred_source()",
 78        )
 79        self.loco_id = loco_id
 80        self.roster_path = roster_path
 81        self.legacy_path = legacy_path
 82
 83
 84def _safe_filename(loco_id: str) -> str:
 85    """Sanitize a loco id into a filesystem-safe filename stem."""
 86    cleaned = _SAFE_FILENAME.sub("_", loco_id).strip("_") or "entry"
 87    return f"{cleaned}.xml"
 88
 89
 90def _resolve_filename(loco_id: str, file_name: str | None) -> str:
 91    """Pick the CV-state filename for *loco_id*.
 92
 93    JMRI's roster stores a ``fileName`` attribute on each locomotive
 94    (often a small integer like ``1.xml``) — when set, that's the file
 95    JMRI uses for decoder state, and that's the file our programmer
 96    must read/write. Falls back to a sanitised ``<id>.xml`` when the
 97    roster doesn't pin a filename.
 98    """
 99    if file_name:
100        # Only the basename; never let a caller traverse out of the dir.
101        return pathlib.Path(file_name).name
102    return _safe_filename(loco_id)
103
104
105@dataclass
106class LocoCvStore:
107    """Lazy disk-backed map of loco-id → :class:`LocoCvState`.
108
109    ``default_lookup`` is optional; when supplied, writes filter out any
110    CV whose value matches the catalog default so ``roster/<id>.xml``
111    stays a minimal delta. See :data:`DefaultLookup` for the callable's
112    signature. Set to ``None`` for legacy behaviour (write every CV).
113    """
114
115    profile: Profile
116    default_lookup: DefaultLookup | None = None
117    _by_id: dict[str, LocoCvState] = field(default_factory=dict)
118    # loco_id → chosen source. Set via ``set_preferred_source`` when the
119    # UI resolves an ambiguity, or automatically when only one source
120    # has data.
121    _source_by_id: dict[str, CvSource] = field(default_factory=dict)
122    # Roster files we've already backed up during this session.
123    # Absolute paths so switching a loco's ``fileName`` still triggers
124    # a fresh backup on the new file. See :meth:`_backup_once`.
125    _backed_up_paths: set[pathlib.Path] = field(default_factory=set)
126
127    def __post_init__(self) -> None:
128        # Both locations are created lazily on first write. Not
129        # pre-scanned — the programmer opens for one loco at a time.
130        self.profile.rosters_cv_dir.mkdir(parents=True, exist_ok=True)
131        self.profile.roster_dir.mkdir(parents=True, exist_ok=True)
132
133    # ----- source resolution ------------------------------------------
134
135    def _roster_path(self, loco_id: str, file_name: str | None) -> pathlib.Path:
136        """The JMRI-canonical file: ``<profile>/roster/<filename>``."""
137        return self.profile.roster_dir / _resolve_filename(loco_id, file_name)
138
139    def _legacy_path(self, loco_id: str, file_name: str | None) -> pathlib.Path:
140        """Our proprietary file: ``<profile>/rosters/<filename>``."""
141        return self.profile.rosters_cv_dir / _resolve_filename(loco_id, file_name)
142
143    def path_for(
144        self, loco_id: str, file_name: str | None = None,
145        source: CvSource | None = None,
146    ) -> pathlib.Path:
147        """Resolve the file path for *loco_id* under *source*.
148
149        Backwards-compat with older callers that assumed the legacy
150        location: when *source* is None, use the loco's preferred source
151        if known, else default to the legacy path.
152        """
153        source = source or self._source_by_id.get(loco_id, CvSource.LEGACY)
154        if source is CvSource.ROSTER:
155            return self._roster_path(loco_id, file_name)
156        return self._legacy_path(loco_id, file_name)
157
158    def _backup_once(self, roster_path: pathlib.Path) -> None:
159        """Copy *roster_path* to ``<name>.xml.bak`` the first time the
160        session touches it.
161
162        No-op if:
163
164        * the path has already been backed up during this store's life;
165        * the source file doesn't exist (nothing to back up);
166        * the copy raises :class:`OSError` (permissions / filesystem
167          full) — logged so a follow-up sync can see the failure, but
168          the write itself proceeds.
169        """
170        abs_path = roster_path.resolve() if roster_path.exists() else roster_path
171        if abs_path in self._backed_up_paths:
172            return
173        # Mark first so a failure to copy doesn't cause a retry storm
174        # on every subsequent write.
175        self._backed_up_paths.add(abs_path)
176        if not roster_path.exists():
177            return
178        backup_path = roster_path.with_suffix(roster_path.suffix + ".bak")
179        try:
180            shutil.copy2(roster_path, backup_path)
181            _log.info(
182                "LocoCvStore: backed up %s → %s", roster_path, backup_path,
183            )
184        except OSError as e:
185            _log.warning(
186                "LocoCvStore: backup %s → %s failed: %s",
187                roster_path, backup_path, e,
188            )
189
190    def set_preferred_source(
191        self, loco_id: str, source: CvSource,
192    ) -> None:
193        """Record which source the UI wants us to use for this loco.
194
195        Clears any cached in-memory state so the next :meth:`get` reads
196        the freshly-picked source from disk.
197        """
198        self._source_by_id[loco_id] = source
199        self._by_id.pop(loco_id, None)
200
201    # ----- read --------------------------------------------------------
202
203    def get(self, loco_id: str, file_name: str | None = None) -> LocoCvState:
204        """Return the loco's CV state, loading from disk on first hit.
205
206        Resolution order:
207
208        1. In-memory cache
209        2. Preferred source set via :meth:`set_preferred_source`
210        3. JMRI-canonical ``roster/<id>.xml`` ``<values>`` block
211        4. Legacy ``rosters/<id>.xml``
212        5. Empty state if none of the above have CVs
213
214        Raises :class:`AmbiguousCvSourceError` when both canonical and
215        legacy locations have CV data and the preferred source hasn't
216        been pinned yet.
217        """
218        if loco_id in self._by_id:
219            return self._by_id[loco_id]
220
221        pref = self._source_by_id.get(loco_id)
222        if pref is CvSource.ROSTER:
223            state = self._read_roster(loco_id, file_name) or LocoCvState(loco_id=loco_id)
224        elif pref is CvSource.LEGACY:
225            state = self._read_legacy(loco_id, file_name) or LocoCvState(loco_id=loco_id)
226        else:
227            roster_state = self._read_roster(loco_id, file_name)
228            legacy_state = self._read_legacy(loco_id, file_name)
229            if roster_state and legacy_state and (
230                roster_state.cv_values and legacy_state.cv_values
231            ):
232                raise AmbiguousCvSourceError(
233                    loco_id,
234                    self._roster_path(loco_id, file_name),
235                    self._legacy_path(loco_id, file_name),
236                )
237            if roster_state and roster_state.cv_values:
238                self._source_by_id[loco_id] = CvSource.ROSTER
239                state = roster_state
240            elif legacy_state:
241                self._source_by_id[loco_id] = CvSource.LEGACY
242                state = legacy_state
243            elif roster_state is not None:
244                # Roster entry file exists but has no <values> — write
245                # to JMRI-canonical on next set_cv for best interop.
246                self._source_by_id[loco_id] = CvSource.ROSTER
247                state = roster_state
248            else:
249                # Neither exists yet — default new locos to canonical.
250                self._source_by_id[loco_id] = CvSource.ROSTER
251                state = LocoCvState(loco_id=loco_id)
252
253        # File-stored ids might not match what the caller passed (JMRI
254        # ``fileName`` decouples filename from loco id). Canonicalise.
255        if state.loco_id != loco_id:
256            state = LocoCvState(loco_id=loco_id, cv_values=dict(state.cv_values))
257        self._by_id[loco_id] = state
258        return state
259
260    def _read_roster(
261        self, loco_id: str, file_name: str | None,
262    ) -> LocoCvState | None:
263        """Try to read CV state from ``roster/<id>.xml`` (JMRI-canonical)."""
264        path = self._roster_path(loco_id, file_name)
265        if not path.exists():
266            return None
267        try:
268            return load_cv_values_from_locomotive_config(path, loco_id=loco_id)
269        except Exception as e:  # noqa: BLE001
270            _log.warning("LocoCvStore: failed to read %s: %s", path, e)
271            return None
272
273    def _read_legacy(
274        self, loco_id: str, file_name: str | None,
275    ) -> LocoCvState | None:
276        """Try to read CV state from ``rosters/<id>.xml`` (our layout)."""
277        path = self._legacy_path(loco_id, file_name)
278        if not path.exists():
279            return None
280        try:
281            return load_loco_cv(path)
282        except Exception as e:  # noqa: BLE001
283            _log.warning("LocoCvStore: failed to read %s: %s", path, e)
284            return None
285
286    # ----- write -------------------------------------------------------
287
288    def record_read_result(
289        self, loco_id: str, cv: str, value: int, file_name: str | None = None,
290    ) -> LocoCvState:
291        """Record a freshly-read CV in the in-memory cache. Does NOT
292        persist to disk and does NOT trigger a ``.bak``.
293
294        JMRI Java treats reads as transient until the user explicitly
295        saves. Mirroring that (rather than our older persist-every-read
296        behaviour) means:
297
298        * The ``.bak`` created by :meth:`_backup_once` captures the
299          pre-modification state — a Read pane that touches 28 CVs
300          doesn't burn the backup on a read that leaves the roster
301          logically unchanged.
302        * SD cards on Pi setups don't take a write per CV read (the
303          Read pane action does ~30 reads sequentially).
304        * A restart discards the read cache — but that's exactly the
305          "reads are ephemeral" contract JMRI Java exposes.
306
307        Callers who want the read to persist (e.g. a "Save read
308        results" button, or a batch that logically commits reads) can
309        call :meth:`set_cv` with the same value.
310        """
311        state = self.get(loco_id, file_name)
312        new_cvs = dict(state.cv_values)
313        new_cvs[str(cv)] = int(value)
314        new_state = LocoCvState(loco_id=loco_id, cv_values=new_cvs)
315        self._by_id[loco_id] = new_state
316        return new_state
317
318    def set_cv(
319        self, loco_id: str, cv: str, value: int, file_name: str | None = None,
320    ) -> LocoCvState:
321        """Set one CV value; persist to the JMRI-canonical
322        ``roster/<id>.xml`` ``<values>`` block, filtering CVs equal to
323        the catalog default. Returns the updated state.
324
325        Filtering behaviour depends on :attr:`default_lookup`:
326
327        * When supplied and it returns a non-empty dict for this loco,
328          any CV whose value matches the default is *dropped* from what
329          gets written — the roster file stays a minimal delta.
330        * When missing or empty, every in-memory CV is written (backward
331          compatible with the pre-refactor behaviour).
332
333        For interop with JMRI Java, writes always target the embedded
334        ``<values>`` block inside ``roster/<id>.xml``. The old
335        proprietary ``rosters/<id>.xml`` remains read-only for the
336        migration window; explicit ``CvSource.LEGACY`` preferences from
337        the ambiguity chooser also read from there but write forward to
338        the canonical location.
339        """
340        state = self.get(loco_id, file_name)
341        new_cvs = dict(state.cv_values)
342        new_cvs[str(cv)] = int(value)
343        new_state = LocoCvState(loco_id=loco_id, cv_values=new_cvs)
344        self._by_id[loco_id] = new_state
345
346        # Apply the default-diff filter: drop any CV whose stringified
347        # value equals the catalog default. Comparison is done on
348        # strings so we sidestep numeric edge cases (e.g. "0"/"00").
349        defaults: dict[str, str] = {}
350        if self.default_lookup is not None:
351            try:
352                defaults = self.default_lookup(loco_id) or {}
353            except Exception as e:  # noqa: BLE001
354                _log.warning(
355                    "LocoCvStore: default lookup failed for %r: %s", loco_id, e,
356                )
357        to_persist = {
358            cv_name: cv_val for cv_name, cv_val in new_cvs.items()
359            if str(cv_val) != defaults.get(cv_name)
360        }
361
362        # Write to JMRI-canonical roster/<id>.xml <values> block.
363        roster_path = self._roster_path(loco_id, file_name)
364        # JMRI Java behaviour: back up the pristine roster file to
365        # ``<id>.xml.bak`` once, before the first write of the session.
366        # Repeated writes to the same file in the same session don't
367        # re-backup (otherwise the backup would race behind current
368        # state and lose the pre-session baseline). Backup is
369        # best-effort — a failure here logs but doesn't block the
370        # write, so a read-only ``.bak`` (e.g. permissions on an SD
371        # card) can't prevent programming.
372        self._backup_once(roster_path)
373        try:
374            save_cv_values_into_locomotive_config(
375                roster_path, to_persist, loco_id=loco_id,
376            )
377        except OSError as e:
378            _log.warning("LocoCvStore: failed to write %s: %s", roster_path, e)
379
380        # Future reads should hit the roster path so the just-written
381        # values come back on a re-open even if legacy still has stale
382        # data from a previous alpha.
383        self._source_by_id[loco_id] = CvSource.ROSTER
384        return new_state