jmri_decoderpro.catalog

  1"""Decoder catalog — lightweight index + lazy full-definition loading.
  2
  3Loading JMRI's 2000+ decoder XML files is too slow when done eagerly
  4(parsing every variable, pane, and enum choice up front). Instead the
  5catalog holds only a ``(family, model) → source_file`` metadata index
  6built by :func:`iter_decoder_index`, and calls :func:`load_decoder` on
  7demand the first time a specific definition is needed.
  8
  9Trade-offs:
 10
 11* ``__iter__`` yields :class:`DecoderIndexEntry` metadata records —
 12  fast enumeration for enumerate/filter/list-models workflows.
 13* :meth:`find` returns a :class:`DecoderDefinition` — triggers the
 14  lazy full-file parse for that entry (once; cached).
 15* :meth:`identify_by_cv` iterates metadata + calls ``find`` for each
 16  candidate. Rare user action, so the load-per-candidate is acceptable.
 17
 18The ``from_files`` classmethod is retained for tests that supply
 19already-parsed :class:`DecoderFile` objects — those populate the cache
 20immediately so no lazy load fires.
 21"""
 22from __future__ import annotations
 23
 24import logging
 25import pathlib
 26from collections.abc import Iterator
 27from dataclasses import dataclass
 28
 29from lxml import etree
 30
 31from jmri_core.persistence.decoder_xml import (
 32    DecoderFamily,
 33    DecoderFile,
 34    DecoderIndexEntry,
 35    DecoderModel,
 36    DecoderPane,
 37    DecoderVariable,
 38    iter_decoder_index,
 39    load_decoder,
 40)
 41from jmri_core.persistence.programmer_xml import (
 42    ProgrammerConfig,
 43    iter_programmer_configs,
 44)
 45
 46_log = logging.getLogger(__name__)
 47
 48
 49@dataclass(frozen=True)
 50class DecoderKey:
 51    """Family + model identity, used as a stable lookup key."""
 52
 53    family: str
 54    model: str
 55
 56    def __str__(self) -> str:
 57        return f"{self.family} / {self.model}"
 58
 59
 60@dataclass(frozen=True)
 61class DecoderDefinition:
 62    """One decoder model's fully parsed definition.
 63
 64    Constructed on demand by :meth:`DecoderCatalog.find` from the source
 65    XML file recorded in the catalog's metadata index. ``panes`` carries
 66    the decoder-specific tabs; the standard tabs are layered on top at
 67    UI-render time.
 68    """
 69
 70    family: DecoderFamily
 71    model: DecoderModel
 72    variables: tuple[DecoderVariable, ...]
 73    panes: tuple[DecoderPane, ...] = ()
 74    source_file: pathlib.Path | None = None
 75
 76    @property
 77    def key(self) -> DecoderKey:
 78        return DecoderKey(family=self.family.name, model=self.model.model)
 79
 80    @property
 81    def mfg(self) -> str:
 82        return self.family.mfg
 83
 84
 85class DecoderCatalog:
 86    """Fast metadata index + lazy full-definition loader.
 87
 88    Two internal maps:
 89
 90    * ``_metadata`` — ``DecoderKey → DecoderIndexEntry`` from the fast scan
 91    * ``_cache``    — ``DecoderKey → DecoderDefinition`` filled on demand
 92
 93    ``programmers`` holds the standard programmer definitions
 94    (``Comprehensive.xml``, ``Basic.xml``, ...) parsed from the sibling
 95    ``programmers/`` dir. Those are small and load eagerly.
 96    """
 97
 98    def __init__(self) -> None:
 99        self._metadata: dict[DecoderKey, DecoderIndexEntry] = {}
100        self._cache: dict[DecoderKey, DecoderDefinition] = {}
101        self.programmers: dict[str, ProgrammerConfig] = {}
102        # NMRA CV8 manufacturer ID → mfg name, loaded from
103        # ``decoderIndex.xml``'s ``<mfgList>`` at :meth:`from_directory`.
104        # Falls back to :data:`_NMRA_MFG_IDS` when the catalog wasn't
105        # loaded from a JMRI xml tree (tests, standalone probes).
106        self.mfg_ids: dict[int, str] = {}
107
108    # ----- construction ---------------------------------------------------
109
110    @classmethod
111    def from_directory(cls, root: str | pathlib.Path) -> DecoderCatalog:
112        """Build the catalog by scanning *root* for decoder XML files.
113
114        Only the identity fields are extracted at this stage — variables
115        + panes are parsed lazily via :meth:`find`. The sibling
116        ``programmers/`` directory (if present) is loaded eagerly since
117        it's small.
118        """
119        catalog = cls()
120        for entry in iter_decoder_index(root):
121            catalog._metadata[DecoderKey(entry.family, entry.model)] = entry
122        # decoderIndex.xml sits at the ``xml/`` root next to the
123        # ``decoders/`` dir and carries the authoritative NMRA
124        # manufacturer table (currently 179 entries). Load it if
125        # present so identify can map any CV8 the decoder returns.
126        index_path = pathlib.Path(root).parent / "decoderIndex.xml"
127        if index_path.is_file():
128            catalog.mfg_ids = _load_mfg_ids(index_path)
129        programmers_dir = pathlib.Path(root).parent / "programmers"
130        if programmers_dir.is_dir():
131            catalog.programmers = {
132                pc.name: pc for pc in iter_programmer_configs(programmers_dir)
133            }
134            _log.info(
135                "DecoderCatalog: indexed %d entries + %d programmer configs",
136                len(catalog), len(catalog.programmers),
137            )
138        else:
139            _log.info("DecoderCatalog: indexed %d entries", len(catalog))
140        return catalog
141
142    @classmethod
143    def from_files(cls, files: list[DecoderFile]) -> DecoderCatalog:
144        """Build the catalog from already-parsed :class:`DecoderFile`
145        objects. Used by tests + callers that want eager loading.
146
147        Populates both the metadata index and the definition cache so
148        subsequent :meth:`find` calls don't trigger any file I/O.
149        """
150        catalog = cls()
151        for f in files:
152            catalog._absorb(f)
153        return catalog
154
155    def add_index_entry(self, entry: DecoderIndexEntry) -> None:
156        """Register a scanned :class:`DecoderIndexEntry` in the metadata
157        index. Full-definition parse is deferred to first :meth:`find`.
158
159        Used by the web catalog-load worker so it can drive the
160        thermometer at scan time rather than parse time.
161        """
162        self._metadata[DecoderKey(entry.family, entry.model)] = entry
163
164    def _absorb(self, df: DecoderFile) -> None:
165        """Populate both metadata and cache from an already-parsed file."""
166        if df.family is None:
167            return
168        for model in df.family.models:
169            key = DecoderKey(df.family.name, model.model)
170            self._metadata[key] = DecoderIndexEntry(
171                family=df.family.name,
172                mfg=df.family.mfg,
173                model=model.model,
174                source_file=df.path or pathlib.Path(""),
175                # Propagate the ``<model productID="…"/>`` attribute
176                # so identify_by_cv_extended can narrow past the
177                # family level. Missing on tests / older files stays
178                # empty string, which the extended-identify path
179                # treats as "no productID match, keep in fallback".
180                product_id=getattr(model, "product_id", ""),
181            )
182            self._cache[key] = DecoderDefinition(
183                family=df.family,
184                model=model,
185                variables=tuple(df.variables),
186                panes=tuple(df.panes),
187                source_file=df.path,
188            )
189
190    # ----- lookup ---------------------------------------------------------
191
192    def __len__(self) -> int:
193        return len(self._metadata)
194
195    def __iter__(self) -> Iterator[DecoderIndexEntry]:
196        """Iterate metadata entries — fast, no full parse.
197
198        Callers that need variables/panes should use :meth:`find` on the
199        entry's key.
200        """
201        return iter(self._metadata.values())
202
203    def get(self, key: DecoderKey) -> DecoderDefinition | None:
204        return self._lazy_load(key)
205
206    def find(self, family: str, model: str) -> DecoderDefinition | None:
207        return self._lazy_load(DecoderKey(family=family, model=model))
208
209    def _lazy_load(self, key: DecoderKey) -> DecoderDefinition | None:
210        """Return the cached definition, parsing the source file on first hit."""
211        if key not in self._metadata:
212            return None
213        cached = self._cache.get(key)
214        if cached is not None:
215            return cached
216        meta = self._metadata[key]
217        source = meta.source_file
218        try:
219            df = load_decoder(source)
220        except Exception as e:  # noqa: BLE001
221            _log.warning(
222                "DecoderCatalog: lazy-load %s (%s/%s) failed: %s",
223                source, key.family, key.model, e,
224            )
225            return None
226        # One decoder file typically declares many models — cache them
227        # all, so a load triggered by find(F, M1) also serves find(F, M2)
228        # without re-parsing the file.
229        self._absorb(df)
230        return self._cache.get(key)
231
232    # ----- fast metadata queries -----------------------------------------
233
234    def manufacturers(self) -> list[str]:
235        return sorted({e.mfg for e in self._metadata.values() if e.mfg})
236
237    def families(self, mfg: str | None = None) -> list[str]:
238        return sorted({
239            e.family for e in self._metadata.values()
240            if e.family and (mfg is None or e.mfg == mfg)
241        })
242
243    def models(self, family: str) -> list[str]:
244        return sorted({
245            e.model for e in self._metadata.values() if e.family == family
246        })
247
248    # ----- identification -------------------------------------------------
249
250    def identify_by_cv(
251        self, cv7_version: int, cv8_mfg: int,
252    ) -> list[DecoderIndexEntry]:
253        """Return metadata entries consistent with CV7/CV8 values.
254
255        Metadata-only (no lazy load): iterating a full catalog and
256        parsing every candidate file is prohibitively slow on modest
257        hardware — the JMRI decoder tree has ~2000 files, each with
258        xi:include fragments to resolve. Instead we filter by mfg alone
259        (from :attr:`mfg_ids` when the catalog was loaded from a JMRI
260        xml tree, or the small stopgap :data:`_NMRA_MFG_IDS` when it
261        wasn't) and let the caller pick a specific entry to fully load
262        via :meth:`find`.
263
264        Raises :class:`UnknownDecoderIdentityError` when *cv8_mfg* isn't
265        a known NMRA manufacturer ID — that includes the NCE NoAck
266        sentinel byte (``0x30`` = 48) that a failed programming-track
267        read returns. Surfacing the error is much friendlier than
268        silently returning "no matches" or, worse, iterating the whole
269        catalog when the read failed.
270        """
271        mfg_name = self.mfg_ids.get(cv8_mfg) or _NMRA_MFG_IDS.get(cv8_mfg)
272        if mfg_name is None:
273            raise UnknownDecoderIdentityError(cv7_version, cv8_mfg)
274        # Match case-insensitively AND on the leading token so
275        # "SoundTraxx (Throttle-Up)" matches entries stored as either
276        # form — decoderIndex.xml uses the parenthesised name; some
277        # older decoder files use the short form.
278        needle = mfg_name.split(" (")[0].lower()
279        # Filter by CV7 within the model's declared version range.
280        # JMRI decoder XMLs use ``lowVersionID`` / ``highVersionID`` on
281        # ``<family>`` (and less often on ``<model>``) to gate which
282        # decoder revisions a definition applies to. Models with no
283        # declared range stay wildcard (0..255) so old files without
284        # the attributes aren't hidden from the identify list.
285        return [
286            entry for entry in self._metadata.values()
287            if needle in entry.mfg.lower()
288            and entry.low_version_id <= cv7_version <= entry.high_version_id
289        ]
290
291    def identify_by_cv_extended(
292        self, cv7_version: int, cv8_mfg: int, product_id: int | None,
293    ) -> list[DecoderIndexEntry]:
294        """Same as :meth:`identify_by_cv`, then narrow by *product_id*.
295
296        When *product_id* is ``None`` (no probe run, or the probe
297        determined the model doesn't populate the productID CVs),
298        this returns the same list as :meth:`identify_by_cv` —
299        the caller shows the family-level picker and the user
300        chooses the model.
301
302        When *product_id* is an int, entries whose
303        :attr:`~jmri_core.persistence.decoder_xml.DecoderIndexEntry.product_id`
304        (the ``<model productID="…"/>`` XML attribute) matches are
305        preferred. JMRI stores comma-separated productID lists on
306        one model when the same XML applies to several variants —
307        we split and match against each. Entries without a
308        productID attribute are always kept in the fallback list
309        so no CV7/CV8 match is ever lost.
310
311        If any entry has a productID match, only those matches are
312        returned. If none do, we fall back to the CV7/CV8-only list
313        — this handles cases like a catalog scan that missed newer
314        productIDs the operator's decoder actually reports.
315        """
316        base = self.identify_by_cv(cv7_version, cv8_mfg)
317        if product_id is None:
318            return base
319        product_id_str = str(product_id)
320        matches = [
321            entry for entry in base
322            if _model_matches_product_id(entry.product_id, product_id_str)
323        ]
324        return matches if matches else base
325
326
327def _model_matches_product_id(entry_field: str, needle: str) -> bool:
328    """Return True when *needle* matches any productID token in
329    *entry_field*. JMRI decoder XML stores multi-productID models as
330    a comma-separated list (``"12345,12346"``); a single-token match
331    is enough."""
332    if not entry_field:
333        return False
334    return any(
335        token.strip() == needle for token in entry_field.split(",")
336    )
337
338
339class UnknownDecoderIdentityError(Exception):
340    """CV7/CV8 read back values we can't map to a known decoder.
341
342    Distinct from "we know the mfg but have no matching decoder in the
343    catalog" (which returns an empty list). This is raised when we
344    can't even interpret the identity bytes — most often because the
345    read failed on the wire and the CS returned its NoAck sentinel.
346    """
347
348    # CV8 values that almost always mean "the physical read failed"
349    # rather than a real (but unrecognised) mfg ID:
350    #   0   — NMRA reserved / decoder didn't respond
351    #   48  — NCE PowerPro NoAck sentinel byte (0x30)
352    #   255 — 0xFF, common "read timed out" fill
353    _WIRE_FAIL_SENTINELS = {0, 48, 255}
354
355    def __init__(self, cv7: int, cv8: int) -> None:
356        if cv8 in self._WIRE_FAIL_SENTINELS:
357            hint = (
358                f"CV8={cv8} looks like a failed wire read (0/48/255 are the "
359                "common no-decoder / no-ack sentinels). Retry, or verify the "
360                "decoder is on the programming track and powered."
361            )
362        else:
363            hint = (
364                f"CV8={cv8} is not a known NMRA manufacturer ID. The catalog "
365                "may not be loaded (Load catalog above), or this is a very "
366                "new / obscure decoder."
367            )
368        super().__init__(f"CV7={cv7} CV8={cv8} — {hint}")
369        self.cv7 = cv7
370        self.cv8 = cv8
371
372
373def _matches_cv7(defn: DecoderDefinition, target: int) -> bool:
374    """Return True if the decoder's CV7 'Decoder Version' default == *target*.
375
376    Many decoder definitions don't pin CV7 (they declare it as a
377    read-only identity variable); in those cases we accept the match
378    rather than excluding the candidate. The CV8 manufacturer match is
379    the more reliable signal.
380    """
381    for variable in defn.variables:
382        if variable.cv != "7":
383            continue
384        if variable.default is None:
385            return True
386        return variable.default == target
387    return True   # CV7 unspecified — leave open
388
389
390# Top-N NMRA manufacturer IDs from the public registry. Used only as a
391# best-effort filter; missing IDs just fall through to the unfiltered
392# (CV7-only) match. When the catalog is loaded from a JMRI xml tree
393# via :meth:`DecoderCatalog.from_directory`, the authoritative table
394# (179 manufacturers) is loaded from ``decoderIndex.xml`` and used in
395# preference to this stopgap. IDs and names below match the JMRI
396# ``<mfgList>`` — do not "correct" without cross-checking that file.
397_NMRA_MFG_IDS: dict[int, str] = {
398    1:   "CML Systems",
399    11:  "North Coast Engineering",
400    54:  "TCH Technology",
401    97:  "Doehler und Haass",
402    99:  "Lenz",
403    101: "Bachmann Trains",
404    113: "QSIndustries",
405    127: "Atlas",
406    129: "Digitrax",
407    131: "Trix Modelleisenbahn",
408    141: "SoundTraxx (Throttle-Up)",
409    143: "MRC",
410    145: "Zimo",
411    151: "Electronic Solutions Ulm GmbH",
412    153: "Train Control Systems",
413    161: "Roco Modelspielwaren",
414    165: "MERG",
415}
416
417
418def _load_mfg_ids(index_path: pathlib.Path) -> dict[int, str]:
419    """Parse the ``<mfgList>`` block of ``decoderIndex.xml``.
420
421    Returns ``{mfgID → mfg name}`` for every ``<manufacturer>`` entry.
422    Malformed IDs are skipped rather than raised — JMRI's file has
423    been solid for years, but keeping the loader tolerant means a
424    single bad entry can't blank the whole identify feature.
425    """
426    try:
427        root = etree.parse(str(index_path)).getroot()
428    except (etree.XMLSyntaxError, OSError) as e:
429        _log.warning("failed to load %s: %s", index_path, e)
430        return {}
431    ids: dict[int, str] = {}
432    for entry in root.iter("manufacturer"):
433        raw_id = entry.get("mfgID", "")
434        name = entry.get("mfg", "")
435        if not name:
436            continue
437        try:
438            ids[int(raw_id)] = name
439        except ValueError:
440            continue
441    _log.info(
442        "DecoderCatalog: loaded %d NMRA manufacturer entries from %s",
443        len(ids), index_path.name,
444    )
445    return ids