jmri_loconet.programmer

  1"""LocoNet service-mode CV programmer (slot 124).
  2
  3LocoNet programming on the programming track works by writing the
  4programmer-mode command into "slot 124" (``PROGRAMMING_SLOT``) via
  5``OPC_WR_SL_DATA`` and waiting for the command station to reply with
  6either:
  7
  8* ``OPC_LONG_ACK`` (write complete, or read failed), **or**
  9* ``OPC_SL_RD_DATA`` for slot 124 carrying the read result.
 10
 11The 14-byte slot-124 frame layout (per Digitrax's published spec):
 12
 13```
 14[0]  OPC_WR_SL_DATA (0xEF)
 15[1]  0x0E   (length = 14)
 16[2]  0x7C   (slot 124)
 17[3]  STAT   programming status (usually 0)
 18[4]  PCMD   programming command — see PCMD_* constants
 19[5]  0x00   reserved
 20[6]  HOPSA  ops-mode address high byte (0 for service mode)
 21[7]  LOPSA  ops-mode address low byte  (0 for service mode)
 22[8]  TRK    track power state (0)
 23[9]  CVH    bit 0=CV7, bit 1=CV8, bit 2=CV9, bit 5=DATA bit 7
 24[10] CVL    CV bits 0..6
 25[11] DATA7  data bits 0..6
 26[12] ID1    session id (0)
 27[13] checksum (XOR, added by the codec)
 28```
 29
 30For ops-mode programming the same frame is used with PCMD's ops-mode bit
 31set and HOPSA/LOPSA carrying the loco address.
 32
 33For Stage 9.1 we ship service-mode-direct only (the modern, decoder-
 34universal mode). Service-mode paged/register and the bit-level operations
 35are deferred to Stage 11.
 36"""
 37from __future__ import annotations
 38
 39import asyncio
 40import logging
 41
 42from jmri_core.beans.programmer import (
 43    Programmer,
 44    ProgrammerManager,
 45    ProgrammingMode,
 46)
 47
 48from jmri_loconet.messages import (
 49    LocoNetMessage,
 50    OpcLongAck,
 51    OpcSlRdData,
 52    OpcWrSlData,
 53)
 54from jmri_loconet.opcodes import OPC_WR_SL_DATA
 55from jmri_loconet.traffic import LocoNetTrafficController
 56
 57_log = logging.getLogger(__name__)
 58
 59PROGRAMMING_SLOT = 124
 60
 61# Programming command (PCMD) bit layout:
 62#   bit 6 = byte vs bit mode  (1 = byte)
 63#   bit 5 = read vs write     (0 = write, 1 = read — JMRI convention)
 64#   bit 4 = ops vs service    (1 = ops mode)
 65#   bits 3..0 = sub-mode
 66# Direct-mode write byte: 0b0110_1000 = 0x68 (byte, write, service, direct)
 67# Direct-mode read byte:  0b0010_1000 = 0x28 (byte, read,  service, direct)
 68# These differ from one popular doc by the JMRI convention; we match JMRI.
 69PCMD_SERVICE_DIRECT_WRITE = 0x68
 70PCMD_SERVICE_DIRECT_READ  = 0x28
 71PCMD_OPS_MODE_WRITE       = 0x6C   # byte-write, ops mode
 72PCMD_OPS_MODE_READ        = 0x2C   # byte-read,  ops mode
 73
 74LONG_ACK_WRITE_DONE = 0x40   # successful write
 75LONG_ACK_READ_FAIL  = 0x00   # decoder didn't respond — read failed
 76
 77
 78# ---- frame builders --------------------------------------------------------
 79
 80
 81def _pack_cv_and_data(cv: int, value: int) -> tuple[int, int, int]:
 82    """Pack the 1-based CV# + 8-bit value into the LocoNet (CVH, CVL, DATA7) triple."""
 83    if not 1 <= cv <= 1024:
 84        raise ValueError(f"CV must be 1..1024, got {cv}")
 85    if not 0 <= value <= 255:
 86        raise ValueError(f"value must be 0..255, got {value}")
 87    cv0 = cv - 1
 88    cvl = cv0 & 0x7F
 89    # CV high bits go into CVH bits 0..2; data bit 7 goes into CVH bit 5.
 90    cvh = ((cv0 >> 7) & 0x07) | ((value & 0x80) >> 2)
 91    data7 = value & 0x7F
 92    return cvh, cvl, data7
 93
 94
 95def _unpack_data(cvh: int, data7: int) -> int:
 96    """Reverse of :func:`_pack_cv_and_data` for the 8-bit data byte."""
 97    return ((cvh & 0x20) << 2) | (data7 & 0x7F)
 98
 99
100def _build_prog_slot_message(
101    pcmd: int,
102    cv: int,
103    value: int,
104    *,
105    ops_address: int = 0,
106) -> OpcWrSlData:
107    """Construct a slot-124 programming command as an :class:`OpcWrSlData`.
108
109    The codec emits this as a 14-byte ``OPC_WR_SL_DATA`` frame; the dataclass
110    fields below intentionally don't map 1:1 to throttle-slot fields, but
111    the wire encoder works because every byte is at the correct offset.
112    """
113    cvh, cvl, data7 = _pack_cv_and_data(cv, value)
114    # Field assignment per the slot-124 layout:
115    #   stat  → [3]    (programming status; 0 for outgoing requests)
116    #   address low 7  → [4] ADR (== PCMD goes here per JMRI spec)
117    #   speed → [5]    (always 0)
118    #   dirf  → [6]    (HOPSA)
119    #   trk   → [7]    (LOPSA — but in OpcWrSlData encode this is bytes[7])
120    #   addr high 7    → [9] ADR2 (== CVH)
121    #   snd   → [10]   (CVL)
122    # This is bending the dataclass to a different slot layout. We document
123    # the byte positions explicitly so reviewers can map them.
124    return OpcWrSlData(
125        slot=PROGRAMMING_SLOT,
126        stat=0,
127        address=cvl | (cvh << 7),   # ADR=cvl, ADR2=cvh per OpcWrSlData encode
128        speed=pcmd,                  # actually placed at body[4] (PCMD)
129        dirf=(ops_address >> 7) & 0x7F,   # HOPSA
130        trk=ops_address & 0x7F,           # LOPSA
131        snd=data7,                         # DATA7
132    )
133
134
135# ---- main programmer ------------------------------------------------------
136
137
138class LnProgrammer:
139    """Service- and ops-mode CV access against a LocoNet command station."""
140
141    def __init__(self, tc: LocoNetTrafficController) -> None:
142        self._tc = tc
143        # One waiter at a time (LocoNet programming is sequential).
144        self._busy = asyncio.Lock()
145
146    async def write_cv_direct(self, cv: int, value: int, *, timeout: float = 6.0) -> None:
147        """Service-mode-direct CV write. Blocks until OPC_LONG_ACK arrives."""
148        async with self._busy:
149            await self._send_and_wait_ack(
150                _build_prog_slot_message(PCMD_SERVICE_DIRECT_WRITE, cv, value),
151                timeout=timeout,
152            )
153
154    async def read_cv_direct(self, cv: int, *, timeout: float = 6.0) -> int:
155        """Service-mode-direct CV read. Returns the value byte."""
156        async with self._busy:
157            return await self._send_and_wait_read(
158                _build_prog_slot_message(PCMD_SERVICE_DIRECT_READ, cv, 0),
159                timeout=timeout,
160            )
161
162    async def write_cv_ops(
163        self, address: int, cv: int, value: int, *, timeout: float = 2.0,
164    ) -> None:
165        """Ops-mode CV write to *address*. No reply needed; we fire-and-forget."""
166        async with self._busy:
167            await self._tc.send(_build_prog_slot_message(
168                PCMD_OPS_MODE_WRITE, cv, value, ops_address=address,
169            ))
170
171    async def read_cv_ops(
172        self, address: int, cv: int, *, timeout: float = 6.0,
173    ) -> int:
174        """Ops-mode CV read — requires RailCom-capable decoder + readers."""
175        async with self._busy:
176            return await self._send_and_wait_read(
177                _build_prog_slot_message(
178                    PCMD_OPS_MODE_READ, cv, 0, ops_address=address,
179                ),
180                timeout=timeout,
181            )
182
183    # ----- internals -------------------------------------------------------
184
185    async def _send_and_wait_ack(self, msg: OpcWrSlData, *, timeout: float) -> None:
186        future: asyncio.Future[None] = asyncio.get_event_loop().create_future()
187
188        async def listener(inbound: LocoNetMessage) -> None:
189            if future.done():
190                return
191            if isinstance(inbound, OpcLongAck) and inbound.responding_opcode == OPC_WR_SL_DATA:
192                if inbound.ack_code == LONG_ACK_WRITE_DONE:
193                    future.set_result(None)
194                else:
195                    future.set_exception(
196                        RuntimeError(f"LocoNet programming write failed: ack={inbound.ack_code:#x}")
197                    )
198
199        disconnect = self._tc.inbound.connect(listener)
200        try:
201            await self._tc.send(msg)
202            await asyncio.wait_for(future, timeout=timeout)
203        finally:
204            disconnect()
205
206    async def _send_and_wait_read(self, msg: OpcWrSlData, *, timeout: float) -> int:
207        future: asyncio.Future[int] = asyncio.get_event_loop().create_future()
208
209        async def listener(inbound: LocoNetMessage) -> None:
210            if future.done():
211                return
212            if isinstance(inbound, OpcSlRdData) and inbound.slot == PROGRAMMING_SLOT:
213                # The slot-rd frame carries (CVH, DATA7) at the same byte
214                # positions our OpcSlRdData decoded as (address_high, snd).
215                cvh = (inbound.address >> 7) & 0x7F
216                value = _unpack_data(cvh, inbound.snd)
217                future.set_result(value)
218            elif (
219                isinstance(inbound, OpcLongAck)
220                and inbound.responding_opcode == OPC_WR_SL_DATA
221                and inbound.ack_code == LONG_ACK_READ_FAIL
222            ):
223                future.set_exception(
224                    RuntimeError("LocoNet programming read failed (decoder did not respond)")
225                )
226
227        disconnect = self._tc.inbound.connect(listener)
228        try:
229            await self._tc.send(msg)
230            return await asyncio.wait_for(future, timeout=timeout)
231        finally:
232            disconnect()
233
234
235# ---- ABC adapters ---------------------------------------------------------
236
237
238class LnServiceProgrammer(Programmer):
239    """Service-mode-direct adapter satisfying the core ABC."""
240
241    def __init__(self, inner: LnProgrammer) -> None:
242        self._inner = inner
243
244    @property
245    def mode(self) -> ProgrammingMode:
246        return ProgrammingMode.SERVICE_DIRECT
247
248    async def read_cv(self, cv: int) -> int:
249        return await self._inner.read_cv_direct(cv)
250
251    async def write_cv(self, cv: int, value: int) -> None:
252        await self._inner.write_cv_direct(cv, value)
253
254
255class LnOpsProgrammer(Programmer):
256    """Ops-mode adapter scoped to a specific DCC address."""
257
258    def __init__(self, inner: LnProgrammer, address: int, *, long_address: bool) -> None:
259        self._inner = inner
260        self._address = address
261        self._long = long_address
262
263    @property
264    def mode(self) -> ProgrammingMode:
265        return ProgrammingMode.OPS_MODE
266
267    @property
268    def address(self) -> int:
269        return self._address
270
271    async def read_cv(self, cv: int) -> int:
272        return await self._inner.read_cv_ops(self._address, cv)
273
274    async def write_cv(self, cv: int, value: int) -> None:
275        await self._inner.write_cv_ops(self._address, cv, value)
276
277
278class LnProgrammerManager(ProgrammerManager):
279    """Hands out LocoNet service/ops programmers backed by one :class:`LnProgrammer`."""
280
281    def __init__(self, inner: LnProgrammer) -> None:
282        self._inner = inner
283        self._service_singleton: LnServiceProgrammer | None = None
284
285    def global_programmer(self) -> LnServiceProgrammer:
286        if self._service_singleton is None:
287            self._service_singleton = LnServiceProgrammer(self._inner)
288        return self._service_singleton
289
290    def ops_programmer(self, address: int, *, long_address: bool = False) -> LnOpsProgrammer:
291        return LnOpsProgrammer(self._inner, address, long_address=long_address)
292
293    @property
294    def inner(self) -> LnProgrammer:
295        return self._inner