jmri_decoderpro

 62@dataclass(frozen=True)
 63class ComposedPane:
 64    """One entry in the composed tab list.
 65
 66    * ``name`` — display label (English default)
 67    * ``slug`` — URL-safe id used in ``/programmer/loco/<id>/pane/<slug>``
 68    * ``pane`` — the underlying :class:`DecoderPane` (None for the
 69      synthetic Roster Entry tab)
 70    * ``is_synthetic`` — True for tabs the framework injects rather than
 71      pulls from XML (currently just Roster Entry)
 72    * ``source`` — where the pane came from: ``"roster"`` |
 73      ``"programmer"`` | ``"decoder"``
 74    * ``items_available`` — variable ``item`` names the loaded decoder
 75      actually provides. A subset of ``pane.items`` filtered against the
 76      decoder's variable table.
 77    """
 78
 79    name: str
 80    slug: str
 81    source: str
 82    pane: DecoderPane | None = None
 83    is_synthetic: bool = False
 84    items_available: tuple[str, ...] = field(default_factory=tuple)
 85
 86    @property
 87    def is_empty(self) -> bool:
 88        """True when the pane has no variables the decoder can serve.
 89
 90        The Roster Entry tab is never empty (it always has roster fields).
 91        A pane with ``<cvtable/>`` or ``<dccaddress/>`` is also never
 92        empty — those widgets don't depend on named variables.
 93        """
 94        if self.is_synthetic:
 95            return False
 96        if self.pane is not None and (
 97            self.pane.has_cvtable
 98            or self.pane.has_dcc_address
 99            or self.pane.has_speed_table
100            or self.pane.has_fnmapping
101        ):
102            return False
103        return not self.items_available

33@dataclass
34class CvCacheEntry:
35    """One observed CV value + the time it was read."""
36
37    value: int
38    read_at: float       # monotonic seconds when we last read it
39    dirty: bool = False  # True if local writes haven't been pushed yet

 86class DecoderCatalog:
 87    """Fast metadata index + lazy full-definition loader.
 88
 89    Two internal maps:
 90
 91    * ``_metadata`` — ``DecoderKey → DecoderIndexEntry`` from the fast scan
 92    * ``_cache``    — ``DecoderKey → DecoderDefinition`` filled on demand
 93
 94    ``programmers`` holds the standard programmer definitions
 95    (``Comprehensive.xml``, ``Basic.xml``, ...) parsed from the sibling
 96    ``programmers/`` dir. Those are small and load eagerly.
 97    """
 98
 99    def __init__(self) -> None:
100        self._metadata: dict[DecoderKey, DecoderIndexEntry] = {}
101        self._cache: dict[DecoderKey, DecoderDefinition] = {}
102        self.programmers: dict[str, ProgrammerConfig] = {}
103        # NMRA CV8 manufacturer ID → mfg name, loaded from
104        # ``decoderIndex.xml``'s ``<mfgList>`` at :meth:`from_directory`.
105        # Falls back to :data:`_NMRA_MFG_IDS` when the catalog wasn't
106        # loaded from a JMRI xml tree (tests, standalone probes).
107        self.mfg_ids: dict[int, str] = {}
108
109    # ----- construction ---------------------------------------------------
110
111    @classmethod
112    def from_directory(cls, root: str | pathlib.Path) -> DecoderCatalog:
113        """Build the catalog by scanning *root* for decoder XML files.
114
115        Only the identity fields are extracted at this stage — variables
116        + panes are parsed lazily via :meth:`find`. The sibling
117        ``programmers/`` directory (if present) is loaded eagerly since
118        it's small.
119        """
120        catalog = cls()
121        for entry in iter_decoder_index(root):
122            catalog._metadata[DecoderKey(entry.family, entry.model)] = entry
123        # decoderIndex.xml sits at the ``xml/`` root next to the
124        # ``decoders/`` dir and carries the authoritative NMRA
125        # manufacturer table (currently 179 entries). Load it if
126        # present so identify can map any CV8 the decoder returns.
127        index_path = pathlib.Path(root).parent / "decoderIndex.xml"
128        if index_path.is_file():
129            catalog.mfg_ids = _load_mfg_ids(index_path)
130        programmers_dir = pathlib.Path(root).parent / "programmers"
131        if programmers_dir.is_dir():
132            catalog.programmers = {
133                pc.name: pc for pc in iter_programmer_configs(programmers_dir)
134            }
135            _log.info(
136                "DecoderCatalog: indexed %d entries + %d programmer configs",
137                len(catalog), len(catalog.programmers),
138            )
139        else:
140            _log.info("DecoderCatalog: indexed %d entries", len(catalog))
141        return catalog
142
143    @classmethod
144    def from_files(cls, files: list[DecoderFile]) -> DecoderCatalog:
145        """Build the catalog from already-parsed :class:`DecoderFile`
146        objects. Used by tests + callers that want eager loading.
147
148        Populates both the metadata index and the definition cache so
149        subsequent :meth:`find` calls don't trigger any file I/O.
150        """
151        catalog = cls()
152        for f in files:
153            catalog._absorb(f)
154        return catalog
155
156    def add_index_entry(self, entry: DecoderIndexEntry) -> None:
157        """Register a scanned :class:`DecoderIndexEntry` in the metadata
158        index. Full-definition parse is deferred to first :meth:`find`.
159
160        Used by the web catalog-load worker so it can drive the
161        thermometer at scan time rather than parse time.
162        """
163        self._metadata[DecoderKey(entry.family, entry.model)] = entry
164
165    def _absorb(self, df: DecoderFile) -> None:
166        """Populate both metadata and cache from an already-parsed file."""
167        if df.family is None:
168            return
169        for model in df.family.models:
170            key = DecoderKey(df.family.name, model.model)
171            self._metadata[key] = DecoderIndexEntry(
172                family=df.family.name,
173                mfg=df.family.mfg,
174                model=model.model,
175                source_file=df.path or pathlib.Path(""),
176                # Propagate the ``<model productID="…"/>`` attribute
177                # so identify_by_cv_extended can narrow past the
178                # family level. Missing on tests / older files stays
179                # empty string, which the extended-identify path
180                # treats as "no productID match, keep in fallback".
181                product_id=getattr(model, "product_id", ""),
182            )
183            self._cache[key] = DecoderDefinition(
184                family=df.family,
185                model=model,
186                variables=tuple(df.variables),
187                panes=tuple(df.panes),
188                source_file=df.path,
189            )
190
191    # ----- lookup ---------------------------------------------------------
192
193    def __len__(self) -> int:
194        return len(self._metadata)
195
196    def __iter__(self) -> Iterator[DecoderIndexEntry]:
197        """Iterate metadata entries — fast, no full parse.
198
199        Callers that need variables/panes should use :meth:`find` on the
200        entry's key.
201        """
202        return iter(self._metadata.values())
203
204    def get(self, key: DecoderKey) -> DecoderDefinition | None:
205        return self._lazy_load(key)
206
207    def find(self, family: str, model: str) -> DecoderDefinition | None:
208        return self._lazy_load(DecoderKey(family=family, model=model))
209
210    def _lazy_load(self, key: DecoderKey) -> DecoderDefinition | None:
211        """Return the cached definition, parsing the source file on first hit."""
212        if key not in self._metadata:
213            return None
214        cached = self._cache.get(key)
215        if cached is not None:
216            return cached
217        meta = self._metadata[key]
218        source = meta.source_file
219        try:
220            df = load_decoder(source)
221        except Exception as e:  # noqa: BLE001
222            _log.warning(
223                "DecoderCatalog: lazy-load %s (%s/%s) failed: %s",
224                source, key.family, key.model, e,
225            )
226            return None
227        # One decoder file typically declares many models — cache them
228        # all, so a load triggered by find(F, M1) also serves find(F, M2)
229        # without re-parsing the file.
230        self._absorb(df)
231        return self._cache.get(key)
232
233    # ----- fast metadata queries -----------------------------------------
234
235    def manufacturers(self) -> list[str]:
236        return sorted({e.mfg for e in self._metadata.values() if e.mfg})
237
238    def families(self, mfg: str | None = None) -> list[str]:
239        return sorted({
240            e.family for e in self._metadata.values()
241            if e.family and (mfg is None or e.mfg == mfg)
242        })
243
244    def models(self, family: str) -> list[str]:
245        return sorted({
246            e.model for e in self._metadata.values() if e.family == family
247        })
248
249    # ----- identification -------------------------------------------------
250
251    def identify_by_cv(
252        self, cv7_version: int, cv8_mfg: int,
253    ) -> list[DecoderIndexEntry]:
254        """Return metadata entries consistent with CV7/CV8 values.
255
256        Metadata-only (no lazy load): iterating a full catalog and
257        parsing every candidate file is prohibitively slow on modest
258        hardware — the JMRI decoder tree has ~2000 files, each with
259        xi:include fragments to resolve. Instead we filter by mfg alone
260        (from :attr:`mfg_ids` when the catalog was loaded from a JMRI
261        xml tree, or the small stopgap :data:`_NMRA_MFG_IDS` when it
262        wasn't) and let the caller pick a specific entry to fully load
263        via :meth:`find`.
264
265        Raises :class:`UnknownDecoderIdentityError` when *cv8_mfg* isn't
266        a known NMRA manufacturer ID — that includes the NCE NoAck
267        sentinel byte (``0x30`` = 48) that a failed programming-track
268        read returns. Surfacing the error is much friendlier than
269        silently returning "no matches" or, worse, iterating the whole
270        catalog when the read failed.
271        """
272        mfg_name = self.mfg_ids.get(cv8_mfg) or _NMRA_MFG_IDS.get(cv8_mfg)
273        if mfg_name is None:
274            raise UnknownDecoderIdentityError(cv7_version, cv8_mfg)
275        # Match case-insensitively AND on the leading token so
276        # "SoundTraxx (Throttle-Up)" matches entries stored as either
277        # form — decoderIndex.xml uses the parenthesised name; some
278        # older decoder files use the short form.
279        needle = mfg_name.split(" (")[0].lower()
280        # Filter by CV7 within the model's declared version range.
281        # JMRI decoder XMLs use ``lowVersionID`` / ``highVersionID`` on
282        # ``<family>`` (and less often on ``<model>``) to gate which
283        # decoder revisions a definition applies to. Models with no
284        # declared range stay wildcard (0..255) so old files without
285        # the attributes aren't hidden from the identify list.
286        return [
287            entry for entry in self._metadata.values()
288            if needle in entry.mfg.lower()
289            and entry.low_version_id <= cv7_version <= entry.high_version_id
290        ]
291
292    def identify_by_cv_extended(
293        self, cv7_version: int, cv8_mfg: int, product_id: int | None,
294    ) -> list[DecoderIndexEntry]:
295        """Same as :meth:`identify_by_cv`, then narrow by *product_id*.
296
297        When *product_id* is ``None`` (no probe run, or the probe
298        determined the model doesn't populate the productID CVs),
299        this returns the same list as :meth:`identify_by_cv` —
300        the caller shows the family-level picker and the user
301        chooses the model.
302
303        When *product_id* is an int, entries whose
304        :attr:`~jmri_core.persistence.decoder_xml.DecoderIndexEntry.product_id`
305        (the ``<model productID="…"/>`` XML attribute) matches are
306        preferred. JMRI stores comma-separated productID lists on
307        one model when the same XML applies to several variants —
308        we split and match against each. Entries without a
309        productID attribute are always kept in the fallback list
310        so no CV7/CV8 match is ever lost.
311
312        If any entry has a productID match, only those matches are
313        returned. If none do, we fall back to the CV7/CV8-only list
314        — this handles cases like a catalog scan that missed newer
315        productIDs the operator's decoder actually reports.
316        """
317        base = self.identify_by_cv(cv7_version, cv8_mfg)
318        if product_id is None:
319            return base
320        product_id_str = str(product_id)
321        matches = [
322            entry for entry in base
323            if _model_matches_product_id(entry.product_id, product_id_str)
324        ]
325        return matches if matches else base

61@dataclass(frozen=True)
62class DecoderDefinition:
63    """One decoder model's fully parsed definition.
64
65    Constructed on demand by :meth:`DecoderCatalog.find` from the source
66    XML file recorded in the catalog's metadata index. ``panes`` carries
67    the decoder-specific tabs; the standard tabs are layered on top at
68    UI-render time.
69    """
70
71    family: DecoderFamily
72    model: DecoderModel
73    variables: tuple[DecoderVariable, ...]
74    panes: tuple[DecoderPane, ...] = ()
75    source_file: pathlib.Path | None = None
76
77    @property
78    def key(self) -> DecoderKey:
79        return DecoderKey(family=self.family.name, model=self.model.model)
80
81    @property
82    def mfg(self) -> str:
83        return self.family.mfg

50@dataclass(frozen=True)
51class DecoderKey:
52    """Family + model identity, used as a stable lookup key."""
53
54    family: str
55    model: str
56
57    def __str__(self) -> str:
58        return f"{self.family} / {self.model}"

47class ProgrammerConfig(BaseModel):
48    """One loaded programmer definition (e.g. Comprehensive.xml).
49
50    * ``panes`` — the standard programmer panes in include order (Basic,
51      Motor, Basic Speed Control, ...). One entry per successful
52      ``xi:include`` resolution.
53    * ``unresolved_includes`` — hrefs that couldn't be resolved to a
54      local file, preserved for diagnostics. An unresolved include just
55      means that pane doesn't appear in the tab list; the rest of the
56      programmer still works.
57    """
58    model_config = ConfigDict(arbitrary_types_allowed=True)
59
60    name: str = ""
61    panes: list[DecoderPane] = Field(default_factory=list)
62    unresolved_includes: list[str] = Field(default_factory=list)
63    source_file: pathlib.Path | None = None
64    # ``<programmer decoderFilePanes="yes" showEmptyPanes="default">``
65    # attributes. ``decoderFilePanes="yes"`` means: also render any
66    # decoder-declared panes in addition to the standard ones.
67    # ``showEmptyPanes="default"`` = respect per-model showEmptyPanes;
68    # ``"yes"`` = always show; ``"no"`` = always hide.
69    decoder_file_panes: bool = True
70    show_empty_panes: str = "default"

 42@dataclass
 43class ProgrammingSession:
 44    """Active CV programming session."""
 45
 46    definition: DecoderDefinition
 47    programmer: Programmer
 48    cv_cache: dict[int, CvCacheEntry] = field(default_factory=dict)
 49
 50    @property
 51    def mode(self) -> ProgrammingMode:
 52        return self.programmer.mode
 53
 54    # ----- construction ----------------------------------------------------
 55
 56    @classmethod
 57    def from_global(
 58        cls, connection: SystemConnection, definition: DecoderDefinition
 59    ) -> ProgrammingSession:
 60        """Service-mode session — uses the connection's global programmer."""
 61        programmer = connection.programmers.global_programmer()
 62        if programmer is None:
 63            raise RuntimeError(
 64                f"connection {connection.name!r} has no global programmer"
 65            )
 66        return cls(definition=definition, programmer=programmer)
 67
 68    @classmethod
 69    def from_ops(
 70        cls,
 71        connection: SystemConnection,
 72        definition: DecoderDefinition,
 73        *,
 74        address: int,
 75        long_address: bool = False,
 76    ) -> ProgrammingSession:
 77        """Ops-mode session for the loco at *address*."""
 78        programmer = connection.programmers.ops_programmer(
 79            address, long_address=long_address
 80        )
 81        if programmer is None:
 82            raise RuntimeError(
 83                f"connection {connection.name!r} has no ops-mode programmer"
 84            )
 85        return cls(definition=definition, programmer=programmer)
 86
 87    # ----- CV access -------------------------------------------------------
 88
 89    async def read_cv(self, cv: int) -> int:
 90        """Read one CV, caching the result for later lookups."""
 91        value = await self.programmer.read_cv(cv)
 92        self.cv_cache[cv] = CvCacheEntry(
 93            value=value, read_at=asyncio.get_event_loop().time(), dirty=False,
 94        )
 95        return value
 96
 97    async def write_cv(self, cv: int, value: int) -> None:
 98        """Write one CV, also updating the local cache."""
 99        if not 0 <= value <= 255:
100            raise ValueError(f"CV value must be 0..255, got {value}")
101        await self.programmer.write_cv(cv, value)
102        self.cv_cache[cv] = CvCacheEntry(
103            value=value, read_at=asyncio.get_event_loop().time(), dirty=False,
104        )
105
106    async def read_all(self) -> dict[int, int]:
107        """Read every CV the decoder definition exposes. Returns a {cv: value} dict.
108
109        Variables with non-integer CV numbers (e.g., "17.18" for long-address
110        composite) are skipped — those need dedicated split-CV handling.
111        """
112        out: dict[int, int] = {}
113        for variable in self.definition.variables:
114            try:
115                cv_num = int(variable.cv)
116            except ValueError:
117                continue
118            if cv_num in out:
119                continue
120            try:
121                out[cv_num] = await self.read_cv(cv_num)
122            except Exception as e:  # noqa: BLE001 — keep reading the rest
123                _log.warning("read_all: CV%d failed (%s)", cv_num, e)
124        return out
125
126    # ----- identity --------------------------------------------------------
127
128    async def read_identity(self) -> tuple[int, int]:
129        """Read CV7 (version) and CV8 (manufacturer). Common for "detect decoder"."""
130        return await self.programmer.read_cv(7), await self.programmer.read_cv(8)
131
132    async def read_identity_extended(self) -> tuple[int, int, int | None]:
133        """Full identity: CV7 (version) + CV8 (mfg) + productID probe.
134
135        The productID probe is per-manufacturer — see
136        :mod:`jmri_decoderpro.probes`. Returns ``product_id=None``
137        when no probe is registered for the mfg, when the probe
138        determined this model doesn't populate the productID CVs,
139        or when a wire error interrupted the probe. Callers pass
140        the returned productID to
141        :meth:`~jmri_decoderpro.catalog.DecoderCatalog.identify_by_cv_extended`
142        to narrow the identify picker beyond CV7/CV8 alone.
143        """
144        from jmri_decoderpro.probes import read_product_id  # noqa: PLC0415
145
146        cv7 = await self.programmer.read_cv(7)
147        cv8 = await self.programmer.read_cv(8)
148        product_id = await read_product_id(self, mfg_id=cv8, model_id=cv7)
149        return cv7, cv8, product_id

340class UnknownDecoderIdentityError(Exception):
341    """CV7/CV8 read back values we can't map to a known decoder.
342
343    Distinct from "we know the mfg but have no matching decoder in the
344    catalog" (which returns an empty list). This is raised when we
345    can't even interpret the identity bytes — most often because the
346    read failed on the wire and the CS returned its NoAck sentinel.
347    """
348
349    # CV8 values that almost always mean "the physical read failed"
350    # rather than a real (but unrecognised) mfg ID:
351    #   0   — NMRA reserved / decoder didn't respond
352    #   48  — NCE PowerPro NoAck sentinel byte (0x30)
353    #   255 — 0xFF, common "read timed out" fill
354    _WIRE_FAIL_SENTINELS = {0, 48, 255}
355
356    def __init__(self, cv7: int, cv8: int) -> None:
357        if cv8 in self._WIRE_FAIL_SENTINELS:
358            hint = (
359                f"CV8={cv8} looks like a failed wire read (0/48/255 are the "
360                "common no-decoder / no-ack sentinels). Retry, or verify the "
361                "decoder is on the programming track and powered."
362            )
363        else:
364            hint = (
365                f"CV8={cv8} is not a known NMRA manufacturer ID. The catalog "
366                "may not be loaded (Load catalog above), or this is a very "
367                "new / obscure decoder."
368            )
369        super().__init__(f"CV7={cv7} CV8={cv8} — {hint}")
370        self.cv7 = cv7
371        self.cv8 = cv8

106def compose_panes(
107    variables: list[DecoderVariable] | tuple[DecoderVariable, ...],
108    programmer: ProgrammerConfig | None,
109    decoder_panes: list[DecoderPane] | tuple[DecoderPane, ...] = (),
110    *,
111    show_empty: bool | None = None,
112) -> list[ComposedPane]:
113    """Build the ordered tab list for a loco's programmer view.
114
115    Behaviour:
116
117    1. Synthetic ``Roster Entry`` tab always first.
118    2. Standard tabs from *programmer*, filtered against
119       ``_STANDARD_TAB_ORDER``. Any standard tab the programmer doesn't
120       declare is skipped.
121    3. Any decoder-declared pane from *decoder_panes* that isn't already
122       covered by name lands after the standard set.
123    4. Empty panes are removed unless *show_empty* forces them in
124       (mirrors JMRI's ``showEmptyPanes`` semantics; default follows the
125       programmer's own setting when *show_empty* is None).
126    """
127    if show_empty is None and programmer is not None:
128        show_empty = programmer.show_empty_panes.lower() == "yes"
129    if show_empty is None:
130        show_empty = False
131
132    decoder_items = {v.item for v in variables if v.item}
133
134    out: list[ComposedPane] = [
135        ComposedPane(
136            name=ROSTER_ENTRY_NAME,
137            slug=ROSTER_ENTRY_SLUG,
138            source="roster",
139            is_synthetic=True,
140        )
141    ]
142
143    # Decoder-declared panes take priority over the standard programmer
144    # pane with the same name — SoundTraxx Tsunami2 ships its own
145    # "Function Map" pane (Extended Function Map) that replaces
146    # Comprehensive.xml's generic ``<fnmapping/>`` matrix, ESU/Zimo do
147    # the same for their model-specific tabs, etc. Index the
148    # decoder-declared panes by name up front so the standard-pane pass
149    # can substitute them in place, preserving the standard tab order.
150    decoder_by_name: dict[str, DecoderPane] = {}
151    include_decoder_panes = programmer is None or programmer.decoder_file_panes
152    if include_decoder_panes:
153        for pane in decoder_panes:
154            if pane.name and pane.name not in _DROPPED_TAB_NAMES:
155                decoder_by_name.setdefault(pane.name, pane)
156
157    # Standard panes — declared by the programmer, ordered by spec.
158    programmer_by_name: dict[str, DecoderPane] = {}
159    if programmer is not None:
160        for pane in programmer.panes:
161            if pane.name and pane.name not in _DROPPED_TAB_NAMES:
162                programmer_by_name.setdefault(pane.name, pane)
163
164    seen_slugs: set[str] = {ROSTER_ENTRY_SLUG}
165    for standard_name in _STANDARD_TAB_ORDER:
166        pane = programmer_by_name.pop(standard_name, None)
167        if pane is None:
168            continue
169        # Prefer the decoder-declared variant when it exists.
170        override = decoder_by_name.pop(standard_name, None)
171        if override is not None:
172            composed = _compose_one(override, "decoder", decoder_items)
173        else:
174            composed = _compose_one(pane, "programmer", decoder_items)
175        if composed.is_empty and not show_empty:
176            continue
177        seen_slugs.add(composed.slug)
178        out.append(composed)
179
180    # Programmer-declared panes that don't match the fixed spec order —
181    # append after the standard block so downstream users still see them.
182    for pane in programmer_by_name.values():
183        override = decoder_by_name.pop(pane.name, None)
184        if override is not None:
185            composed = _compose_one(override, "decoder", decoder_items)
186        else:
187            composed = _compose_one(pane, "programmer", decoder_items)
188        if composed.is_empty and not show_empty:
189            continue
190        if composed.slug in seen_slugs:
191            continue
192        seen_slugs.add(composed.slug)
193        out.append(composed)
194
195    # Remaining decoder-declared panes — ones with no matching standard
196    # tab. Append after the standard block.
197    seen_names = {c.name for c in out}
198    for pane in decoder_by_name.values():
199        if pane.name in seen_names:
200            continue
201        composed = _compose_one(pane, "decoder", decoder_items)
202        if composed.is_empty and not show_empty:
203            continue
204        if composed.slug in seen_slugs:
205            continue
206        seen_slugs.add(composed.slug)
207        seen_names.add(composed.name)
208        out.append(composed)
209
210    return out

252def find_pane_by_slug(
253    panes: list[ComposedPane], slug: str,
254) -> ComposedPane | None:
255    for p in panes:
256        if p.slug == slug:
257            return p
258    return None