jmri_decoderpro.probes

  1"""Per-manufacturer productID probes.
  2
  3CV7 (version) + CV8 (mfg) alone typically narrow a decoder identity
  4to a family, not a specific model. Each manufacturer that ships more
  5than a couple of decoders per family encodes a **productID** across
  6one or more extra CVs — the exact set differs per manufacturer, and
  7sometimes per model within a manufacturer. This module implements
  8the same probe logic JMRI Java's
  9``jmri.jmrit.decoderdefn.IdentifyDecoder`` runs, translated to the
 10async programmer API here.
 11
 12Verified against::
 13
 14    ~/Documents/jmri.jar.src/jmri/jmrit/decoderdefn/IdentifyDecoder.java
 15
 16**NMRA manufacturer IDs** used below (from that file's
 17:class:`Manufacturer` enum, lines 105..117 in the decompiled source)::
 18
 19    DIETZ        = 115
 20    DIY          =  13
 21    DOEHLER      =  97
 22    ESU          = 151
 23    HARMAN       =  98
 24    HORNBY       =  48
 25    QSI          = 113
 26    SOUNDTRAXX   = 141
 27    TCS          = 153
 28    TRAINOMATIC  =  78
 29    ZIMO         = 145
 30
 31**What this module currently implements** (highest-value set for
 32North-American users, matching what real decoders on the workbench
 33here can exercise):
 34
 35* :func:`probe_soundtraxx` (CV8 = 141) — CV253/255/256 bit-packing.
 36* :func:`probe_esu`        (CV8 = 151) — CV31/32 page/index writes
 37  plus 4-byte little-endian read from CV261..264.
 38* :func:`probe_zimo`       (CV8 = 145) — single CV250 read.
 39* :func:`probe_tcs`        (CV8 = 153) — CV249 with branching to
 40  248/111/110 for extended IDs.
 41
 42Others (Hornby, Dietz, Harman, DIY, Doehler, Trainomatic, QSI) are
 43tracked in ``README.md`` as future adds. NCE + Digitrax are NOT
 44probed even in JMRI Java — CV7/CV8 alone identifies them.
 45
 46Probes are registered in :data:`PROBES` keyed by the CV8 mfg-id;
 47dispatch happens via :func:`read_product_id`. Adding a new probe is
 48a one-line registry entry plus a standalone async function; the
 49identify flow doesn't need to change.
 50"""
 51from __future__ import annotations
 52
 53import logging
 54from collections.abc import Awaitable, Callable
 55from typing import TYPE_CHECKING
 56
 57if TYPE_CHECKING:
 58    from jmri_decoderpro.session import ProgrammingSession
 59
 60_log = logging.getLogger(__name__)
 61
 62
 63# Mfg IDs — kept as module-level constants so tests and future probes
 64# can reference them without magic numbers.
 65MFG_SOUNDTRAXX = 141
 66MFG_ESU = 151
 67MFG_ZIMO = 145
 68MFG_TCS = 153
 69
 70
 71ProbeFn = Callable[
 72    ["ProgrammingSession", int, int],
 73    Awaitable[int | None],
 74]
 75
 76
 77# ---- SoundTraxx --------------------------------------------------------
 78
 79
 80async def probe_soundtraxx(
 81    session: ProgrammingSession, mfg_id: int, model_id: int,
 82) -> int | None:
 83    """SoundTraxx productID probe.
 84
 85    Only fires for Tsunami2 family (``model_id`` in 70..72); older
 86    Tsunami / Econami decoders don't populate the productID CVs.
 87
 88    Byte-packing (from ``IdentifyDecoder.java`` lines 262..380)::
 89
 90        highest = CV253
 91        low     = CV256
 92        high    = CV255
 93        productID = low | ((high & 0x7) << 8) | (highest << 11)
 94    """
 95    if not (70 <= model_id <= 72):
 96        return None
 97    highest = await session.programmer.read_cv(253)
 98    low = await session.programmer.read_cv(256)
 99    high = await session.programmer.read_cv(255)
100    return low | ((high & 0x7) << 8) | (highest << 11)
101
102
103# ---- ESU ---------------------------------------------------------------
104
105
106async def probe_esu(
107    session: ProgrammingSession, mfg_id: int, model_id: int,
108) -> int | None:
109    """ESU (Lok family) productID probe.
110
111    Only fires for LokPilot v4+ / LokSound v4+ (``model_id == 255``);
112    older ESU decoders don't have the extended-ID CVs.
113
114    Two index CVs (CV31 / CV32) select the "extended-identifier"
115    page, then 4 bytes at CV261..264 make up a little-endian
116    productID (from ``IdentifyDecoder.java`` lines 331..458)::
117
118        write CV31 = 0
119        write CV32 = 255
120        productID = CV261 | (CV262 << 8) | (CV263 << 16) | (CV264 << 24)
121    """
122    if model_id != 255:
123        return None
124    # These writes route subsequent reads through the extended
125    # identifier page. Failing writes are non-fatal — some ops-mode
126    # programmers can't do the page-switch and the caller falls
127    # back to family-level ID.
128    await session.programmer.write_cv(31, 0)
129    await session.programmer.write_cv(32, 255)
130    b0 = await session.programmer.read_cv(261)
131    b1 = await session.programmer.read_cv(262)
132    b2 = await session.programmer.read_cv(263)
133    b3 = await session.programmer.read_cv(264)
134    return b0 | (b1 << 8) | (b2 << 16) | (b3 << 24)
135
136
137# ---- Zimo --------------------------------------------------------------
138
139
140async def probe_zimo(
141    session: ProgrammingSession, mfg_id: int, model_id: int,
142) -> int | None:
143    """Zimo productID probe — CV250 holds a single-byte decoder ID."""
144    return await session.programmer.read_cv(250)
145
146
147# ---- TCS ---------------------------------------------------------------
148
149
150async def probe_tcs(
151    session: ProgrammingSession, mfg_id: int, model_id: int,
152) -> int | None:
153    """Train Control Systems (TCS) productID probe.
154
155    TCS uses CV249 as the primary ID; older decoders (< 129) stop
156    there. Newer decoders extend the ID across CV248 / CV111 / CV110
157    with a family-specific packing formula (from
158    ``IdentifyDecoder.java`` lines 227..426).
159
160    Short form (CV249 < 129)::
161
162        productID = CV249
163
164    Extended form (CV249 ≥ 129)::
165
166        lowest = CV249; low = CV248; high = CV111; highest = CV110
167        # Which packing applies depends on ``model_id`` (from CV7) and
168        # the observed lowest byte — matches the branching in
169        # ``IdentifyDecoder.test7`` verbatim.
170    """
171    lowest = await session.programmer.read_cv(249)
172    if lowest < 129:
173        return lowest
174
175    low = await session.programmer.read_cv(248)
176    high = await session.programmer.read_cv(111)
177    highest = await session.programmer.read_cv(110)
178
179    # See ``IdentifyDecoder.test7`` (Java lines 414..425) for the
180    # authoritative branching. Kept as a nested if/else to make the
181    # correspondence with the Java source obvious to a reader who
182    # cross-checks; a table-driven form would obscure the mapping.
183    if (
184        (129 <= lowest <= 135 and low == 5)
185        or model_id >= 5
186    ):
187        if lowest == 180 and model_id == 5:
188            return lowest + low * 256
189        return (
190            lowest + low * 256 + high * 65536 + highest * 16777216
191        )
192    if (
193        (129 <= lowest <= 135)
194        or (170 <= lowest <= 172)
195        or lowest == 180
196    ) and model_id == 4:
197        return lowest + low * 256
198    return lowest
199
200
201# ---- registry + dispatch -----------------------------------------------
202
203
204PROBES: dict[int, ProbeFn] = {
205    MFG_SOUNDTRAXX: probe_soundtraxx,
206    MFG_ESU: probe_esu,
207    MFG_ZIMO: probe_zimo,
208    MFG_TCS: probe_tcs,
209}
210
211
212async def read_product_id(
213    session: ProgrammingSession, mfg_id: int, model_id: int,
214) -> int | None:
215    """Look up the probe for *mfg_id* and run it.
216
217    Returns ``None`` when:
218
219    * no probe is registered for the manufacturer, or
220    * the probe determined the model doesn't populate productID CVs
221      (e.g. older SoundTraxx decoders), or
222    * a wire-level error prevented any CV from being read — the
223      caller falls back to family-level identify in that case.
224
225    The error path treats a raised exception during a probe as a
226    non-fatal miss: identify continues with just CV7+CV8. That
227    matches JMRI Java's behaviour of continuing after an optional
228    CV read fails.
229    """
230    probe = PROBES.get(mfg_id)
231    if probe is None:
232        return None
233    try:
234        return await probe(session, mfg_id, model_id)
235    except Exception as e:  # noqa: BLE001
236        _log.warning(
237            "productID probe for mfg=%d model=%d failed: %s",
238            mfg_id, model_id, e,
239        )
240        return None