jmri_server.services.programmer

  1"""``ProgrammerService`` — JSON wire interface for CV programming.
  2
  3JMRI's stock JSON API doesn't expose decoder programming, so we ship a
  4small dedicated topic at ``type: "programmer"``. Wire shapes:
  5
  6* ``method=get``, ``data={"cv": N, "mode": "service"|"ops", "address"?: N}``
  7  → reads the CV, returns ``{"cv": N, "value": 0..255}``.
  8* ``method=post``, ``data={"cv": N, "value": V, ...}`` → writes the CV.
  9* ``method=list``, ``data={"mode": "service"|"ops"}`` → returns
 10  ``{"mode": ..., "available": true|false, "ops_address": ...}``.
 11
 12The service uses the active connection's ``ProgrammerManager``. If the
 13connection has no programmer (default in-memory connections), every
 14request returns ``available: false``.
 15"""
 16from __future__ import annotations
 17
 18from typing import Any, cast
 19
 20from jmri_core.beans.programmer import Programmer, ProgrammingMode
 21
 22from jmri_server.service import JsonService, JsonServiceError
 23
 24# CV value range. JMRI ships CVs in the standard NMRA 8-bit range.
 25CV_VALUE_MIN = 0
 26CV_VALUE_MAX = 255
 27
 28
 29def _resolve_programmer(ctx_connection: Any, data: dict[str, Any]) -> Programmer:
 30    """Pick a service or ops-mode programmer from the request data."""
 31    mode_str = data.get("mode", "service")
 32    if mode_str == "service":
 33        programmer = ctx_connection.programmers.global_programmer()
 34        if programmer is None:
 35            raise JsonServiceError(
 36                503,
 37                f"connection {ctx_connection.name!r} has no service-mode programmer",
 38            )
 39        return cast("Programmer", programmer)
 40    if mode_str == "ops":
 41        address = data.get("address")
 42        if not isinstance(address, int):
 43            raise JsonServiceError(400, "ops-mode programming requires an 'address'")
 44        long_address = bool(data.get("isLongAddress", False))
 45        programmer = ctx_connection.programmers.ops_programmer(
 46            address, long_address=long_address
 47        )
 48        if programmer is None:
 49            raise JsonServiceError(
 50                503,
 51                f"connection {ctx_connection.name!r} has no ops-mode programmer",
 52            )
 53        return cast("Programmer", programmer)
 54    raise JsonServiceError(400, f"unknown programmer mode {mode_str!r}")
 55
 56
 57def _loco_cv_store(ctx: Any) -> Any:
 58    """Lazily attach a :class:`LocoCvStore` rooted at the active profile.
 59
 60    Wires a ``default_lookup`` callable that maps ``loco_id`` →
 61    ``{cv_number: default_str}`` by consulting the roster (for the loco's
 62    family/model) + the decoder catalog (for that model's variable
 63    defaults). When the catalog isn't loaded or the loco has no decoder
 64    metadata, returns ``{}`` so :meth:`LocoCvStore.set_cv` writes every
 65    CV (equivalent to pre-refactor behaviour).
 66    """
 67    store = getattr(ctx, "_loco_cv_store", None)
 68    if store is not None:
 69        return store
 70    from jmri_core.persistence.profile import Profile  # noqa: PLC0415
 71
 72    from jmri_server.loco_cv_store import LocoCvStore  # noqa: PLC0415
 73    profile = getattr(ctx, "_roster_profile", None) or Profile.default()
 74
 75    def _defaults_for(loco_id: str) -> dict[str, str]:
 76        """Look up catalog defaults for *loco_id*'s decoder model.
 77
 78        Failure modes (missing catalog, unknown model, no roster entry)
 79        all resolve to an empty dict so the write path just skips the
 80        filter — losing the "diff-only" optimisation but preserving
 81        correctness (every CV in memory gets written).
 82        """
 83        catalog = getattr(ctx, "_decoder_catalog", None)
 84        if catalog is None:
 85            return {}
 86        # Resolve family/model via the roster service. This runs
 87        # synchronously inside LocoCvStore.set_cv; the roster service's
 88        # own read path is sync-safe.
 89        try:
 90            from jmri_server.services.roster import (  # noqa: PLC0415
 91                _roster_service_get_sync,
 92            )
 93            entry = _roster_service_get_sync(ctx, loco_id)
 94        except Exception:  # noqa: BLE001
 95            return {}
 96        family = (entry or {}).get("decoderFamily") or ""
 97        model = (entry or {}).get("decoderModel") or ""
 98        if not family or not model:
 99            return {}
100        defn = catalog.find(family, model)
101        if defn is None:
102            return {}
103        # Compose per-CV *default bytes* by applying each variable's
104        # human-default value through ``variable_apply_to_state``. Bit-
105        # mask variables sharing a CV (Address Format + Speed Steps +
106        # Power Source Conv on CV 29) each fold their bits into the
107        # same byte via read-modify-write against the running snapshot.
108        # Split-CV variables (long address = CV 17 + CV 18) produce
109        # entries for every touched CV. Skip variable kinds that don't
110        # participate in byte-mask composition (SPEED_TABLE, COMPOSITE,
111        # SPLIT_DATETIME, CONSTANT).
112        from jmri_core.persistence.decoder_xml import (  # noqa: PLC0415
113            VariableKind,
114            variable_apply_to_state,
115            variable_is_indexed,
116            variable_store_key,
117        )
118        # String-keyed so plain and indexed CV defaults coexist
119        # (``275`` vs ``16.1.275``). Indexed variables get their
120        # default bytes stored per compound key; plain variables get
121        # str(cv) as before.
122        result: dict[str, str] = {}
123        int_view: dict[int, int] = {}
124        for var in defn.variables:
125            if var.default is None:
126                continue
127            if var.kind in (
128                VariableKind.SPEED_TABLE,
129                VariableKind.COMPOSITE,
130                VariableKind.SPLIT_DATETIME,
131                VariableKind.CONSTANT,
132            ):
133                continue
134            if variable_is_indexed(var):
135                # Indexed: emit one entry per byte under compound key.
136                assert var.split is not None
137                base = var.split.primary_cv
138                for i in range(var.split.byte_count):
139                    key = variable_store_key(var, base + i)
140                    shift = 8 * (var.split.byte_count - 1 - i)
141                    result[key] = str((int(var.default) >> shift) & 0xFF)
142                continue
143            try:
144                updates = variable_apply_to_state(var, var.default, int_view)
145            except Exception:  # noqa: BLE001
146                continue
147            int_view.update(updates)
148            for cv_int, byte in updates.items():
149                result[str(cv_int)] = str(byte)
150        return result
151
152    store = LocoCvStore(profile=profile, default_lookup=_defaults_for)
153    ctx._loco_cv_store = store
154    return store
155
156
157class ProgrammerService(JsonService):
158    topic = "programmer"
159
160    async def on_get(self, data: dict[str, Any] | None) -> dict[str, Any]:
161        data = data or {}
162        cv = data.get("cv")
163        if not isinstance(cv, int):
164            raise JsonServiceError(400, "programmer.get requires integer 'cv'")
165        programmer = _resolve_programmer(self.ctx.connection, data)
166        try:
167            value = await programmer.read_cv(cv)
168        except TimeoutError:
169            raise JsonServiceError(504, f"timeout reading CV {cv}") from None
170        except NotImplementedError as e:
171            # e.g. NCE ops-mode CV read — no RailCom on PowerPro.
172            raise JsonServiceError(501, str(e)) from None
173        except Exception as e:  # noqa: BLE001
174            # Connection-specific errors (NceCommandError, LocoNet CS
175            # NACKs, DCC-EX parse failures). Surface as 502 so the UI
176            # can distinguish from timeouts/protocol errors upstream.
177            raise JsonServiceError(502, f"CV {cv} read failed: {e}") from None
178        # Record the freshly-read value in the in-memory cache so the
179        # next pane render (and subsequent variable_extract_from_state
180        # calls) sees it. JMRI Java treats reads as transient — no
181        # disk persistence, no ``.bak`` — until the user explicitly
182        # saves. We mirror that via ``record_read_result`` rather than
183        # ``set_cv``. A Read pane over 28 CVs used to translate into
184        # 28 atomic disk writes (rough on SD cards); now it's zero.
185        loco_id = data.get("locoId") or data.get("loco_id")
186        if isinstance(loco_id, str) and loco_id:
187            file_name = data.get("fileName") or data.get("file_name")
188            if not isinstance(file_name, str):
189                file_name = None
190            _loco_cv_store(self.ctx).record_read_result(
191                loco_id, str(cv), value, file_name,
192            )
193        return {"cv": cv, "value": value, "mode": programmer.mode.value}
194
195    async def on_post(self, data: dict[str, Any] | None) -> dict[str, Any]:
196        data = data or {}
197        cv = data.get("cv")
198        value = data.get("value")
199        if not isinstance(cv, int):
200            raise JsonServiceError(400, "programmer.post requires integer 'cv'")
201        if not isinstance(value, int) or not CV_VALUE_MIN <= value <= CV_VALUE_MAX:
202            raise JsonServiceError(
203                400,
204                f"programmer.post requires integer 'value' {CV_VALUE_MIN}..{CV_VALUE_MAX}",
205            )
206        programmer = _resolve_programmer(self.ctx.connection, data)
207        try:
208            await programmer.write_cv(cv, value)
209        except TimeoutError:
210            raise JsonServiceError(504, f"timeout writing CV {cv}") from None
211        except NotImplementedError as e:
212            raise JsonServiceError(501, str(e)) from None
213        except Exception as e:  # noqa: BLE001
214            raise JsonServiceError(502, f"CV {cv} write failed: {e}") from None
215        # Persist the committed write to the per-loco CV file if a loco
216        # was identified — keeps the on-disk state in sync with what
217        # was actually programmed. JMRI's roster ``fileName`` attribute
218        # decouples loco-id from filename (e.g. id=459 fileName=1.xml).
219        loco_id = data.get("locoId") or data.get("loco_id")
220        if isinstance(loco_id, str) and loco_id:
221            file_name = data.get("fileName") or data.get("file_name")
222            if not isinstance(file_name, str):
223                file_name = None
224            _loco_cv_store(self.ctx).set_cv(loco_id, str(cv), value, file_name)
225        return {"cv": cv, "value": value, "mode": programmer.mode.value}
226
227    async def on_list(self, data: dict[str, Any] | None) -> list[dict[str, Any]]:
228        """Report what programmers are available on the active connection."""
229        out: list[dict[str, Any]] = []
230        global_prog = self.ctx.connection.programmers.global_programmer()
231        out.append({
232            "mode": ProgrammingMode.SERVICE_DIRECT.value,
233            "available": global_prog is not None,
234        })
235        # Ops-mode availability per-address; we report just the manager-level capability.
236        try:
237            ops_capable = (
238                self.ctx.connection.programmers.ops_programmer(3) is not None
239            )
240        except Exception:  # noqa: BLE001
241            ops_capable = False
242        out.append({
243            "mode": ProgrammingMode.OPS_MODE.value,
244            "available": ops_capable,
245        })
246        return out