jmri_decoderpro.session

  1"""Programming session — binds a decoder definition + programmer.
  2
  3A :class:`ProgrammingSession` is the top-level handle DecoderPro hands
  4to clients. It knows:
  5
  6* which :class:`DecoderDefinition` is on the programming track (or the
  7  ops-mode loco being modified);
  8* which :class:`Programmer` to drive;
  9* the current cache of CV values read so far.
 10
 11Two flavors:
 12
 13* :meth:`from_global` — service-mode session against the programming
 14  track. Reads work on any decoder.
 15* :meth:`from_ops` — ops-mode session against a running loco; reads
 16  only succeed with RailCom-capable hardware.
 17"""
 18from __future__ import annotations
 19
 20import asyncio
 21import logging
 22from dataclasses import dataclass, field
 23
 24from jmri_core.beans.programmer import Programmer, ProgrammingMode
 25from jmri_core.hal.connection import SystemConnection
 26
 27from jmri_decoderpro.catalog import DecoderDefinition
 28
 29_log = logging.getLogger(__name__)
 30
 31
 32@dataclass
 33class CvCacheEntry:
 34    """One observed CV value + the time it was read."""
 35
 36    value: int
 37    read_at: float       # monotonic seconds when we last read it
 38    dirty: bool = False  # True if local writes haven't been pushed yet
 39
 40
 41@dataclass
 42class ProgrammingSession:
 43    """Active CV programming session."""
 44
 45    definition: DecoderDefinition
 46    programmer: Programmer
 47    cv_cache: dict[int, CvCacheEntry] = field(default_factory=dict)
 48
 49    @property
 50    def mode(self) -> ProgrammingMode:
 51        return self.programmer.mode
 52
 53    # ----- construction ----------------------------------------------------
 54
 55    @classmethod
 56    def from_global(
 57        cls, connection: SystemConnection, definition: DecoderDefinition
 58    ) -> ProgrammingSession:
 59        """Service-mode session — uses the connection's global programmer."""
 60        programmer = connection.programmers.global_programmer()
 61        if programmer is None:
 62            raise RuntimeError(
 63                f"connection {connection.name!r} has no global programmer"
 64            )
 65        return cls(definition=definition, programmer=programmer)
 66
 67    @classmethod
 68    def from_ops(
 69        cls,
 70        connection: SystemConnection,
 71        definition: DecoderDefinition,
 72        *,
 73        address: int,
 74        long_address: bool = False,
 75    ) -> ProgrammingSession:
 76        """Ops-mode session for the loco at *address*."""
 77        programmer = connection.programmers.ops_programmer(
 78            address, long_address=long_address
 79        )
 80        if programmer is None:
 81            raise RuntimeError(
 82                f"connection {connection.name!r} has no ops-mode programmer"
 83            )
 84        return cls(definition=definition, programmer=programmer)
 85
 86    # ----- CV access -------------------------------------------------------
 87
 88    async def read_cv(self, cv: int) -> int:
 89        """Read one CV, caching the result for later lookups."""
 90        value = await self.programmer.read_cv(cv)
 91        self.cv_cache[cv] = CvCacheEntry(
 92            value=value, read_at=asyncio.get_event_loop().time(), dirty=False,
 93        )
 94        return value
 95
 96    async def write_cv(self, cv: int, value: int) -> None:
 97        """Write one CV, also updating the local cache."""
 98        if not 0 <= value <= 255:
 99            raise ValueError(f"CV value must be 0..255, got {value}")
100        await self.programmer.write_cv(cv, value)
101        self.cv_cache[cv] = CvCacheEntry(
102            value=value, read_at=asyncio.get_event_loop().time(), dirty=False,
103        )
104
105    async def read_all(self) -> dict[int, int]:
106        """Read every CV the decoder definition exposes. Returns a {cv: value} dict.
107
108        Variables with non-integer CV numbers (e.g., "17.18" for long-address
109        composite) are skipped — those need dedicated split-CV handling.
110        """
111        out: dict[int, int] = {}
112        for variable in self.definition.variables:
113            try:
114                cv_num = int(variable.cv)
115            except ValueError:
116                continue
117            if cv_num in out:
118                continue
119            try:
120                out[cv_num] = await self.read_cv(cv_num)
121            except Exception as e:  # noqa: BLE001 — keep reading the rest
122                _log.warning("read_all: CV%d failed (%s)", cv_num, e)
123        return out
124
125    # ----- identity --------------------------------------------------------
126
127    async def read_identity(self) -> tuple[int, int]:
128        """Read CV7 (version) and CV8 (manufacturer). Common for "detect decoder"."""
129        return await self.programmer.read_cv(7), await self.programmer.read_cv(8)
130
131    async def read_identity_extended(self) -> tuple[int, int, int | None]:
132        """Full identity: CV7 (version) + CV8 (mfg) + productID probe.
133
134        The productID probe is per-manufacturer — see
135        :mod:`jmri_decoderpro.probes`. Returns ``product_id=None``
136        when no probe is registered for the mfg, when the probe
137        determined this model doesn't populate the productID CVs,
138        or when a wire error interrupted the probe. Callers pass
139        the returned productID to
140        :meth:`~jmri_decoderpro.catalog.DecoderCatalog.identify_by_cv_extended`
141        to narrow the identify picker beyond CV7/CV8 alone.
142        """
143        from jmri_decoderpro.probes import read_product_id  # noqa: PLC0415
144
145        cv7 = await self.programmer.read_cv(7)
146        cv8 = await self.programmer.read_cv(8)
147        product_id = await read_product_id(self, mfg_id=cv8, model_id=cv7)
148        return cv7, cv8, product_id