jmri_withrottle.protocol

  1"""WiThrottle line protocol — parsing + encoding.
  2
  3WiThrottle is a line-based ASCII protocol on TCP. Newline-terminated
  4messages flow in both directions, with mostly self-describing first
  5characters:
  6
  7* ``VN`` — protocol version
  8* ``HT`` — server (railroad) name
  9* ``RL`` — roster list
 10* ``PPA`` — track power: ``PPA1`` on, ``PPA0`` off
 11* ``PFT`` — fast clock
 12* ``M{T,S}…`` — throttle multiplex commands; ``T`` is the primary
 13  throttle, ``S`` is the secondary on the same client
 14* ``*`` — heartbeat tick from client
 15* ``Q`` — quit
 16
 17We implement enough of the spec to drive a loco from Engine Driver:
 18acquire/release a throttle, set speed/direction/functions, e-stop,
 19heartbeat. The fancier bits (consist building, signal cues,
 20turnout/route control over WiThrottle, panel data, fast-clock sync)
 21land in follow-up sprints.
 22
 23References:
 24* https://www.jmri.org/help/en/package/jmri/jmrit/withrottle/Protocol.shtml
 25* Engine Driver's source (Android) for real-world client behaviour.
 26"""
 27from __future__ import annotations
 28
 29from dataclasses import dataclass
 30
 31
 32# Default heartbeat interval the server requests of clients. Engine
 33# Driver's default poll interval matches this.
 34DEFAULT_HEARTBEAT_S = 10
 35
 36# Maximum function number we acknowledge. WiThrottle supports F0..F28.
 37MAX_FUNCTION = 28
 38
 39# Default function labels sent when no roster labels are supplied.
 40# Engine Driver's manual-address acquire path uses its *own* client-side
 41# defaults ("Light", "Bell", "Horn" for F0..F2, generic Fn thereafter);
 42# roster-based acquires take the labels from the wire. Matching those
 43# three keeps both acquire paths cosmetically consistent.
 44_DEFAULT_FUNCTION_LABELS = (
 45    ["Light", "Bell", "Horn"]
 46    + [f"F{i}" for i in range(3, MAX_FUNCTION + 1)]
 47)
 48
 49
 50# ----- typed inbound messages ----------------------------------------------
 51
 52
 53@dataclass(slots=True)
 54class HelloName:
 55    """Client sent its display name. ``N<name>``."""
 56    name: str
 57
 58
 59@dataclass(slots=True)
 60class HelloHardwareId:
 61    """Client sent its hardware id. ``HU<id>``."""
 62    hardware_id: str
 63
 64
 65@dataclass(slots=True)
 66class Heartbeat:
 67    """Client sent the periodic ``*`` tick."""
 68
 69
 70@dataclass(slots=True)
 71class Quit:
 72    """Client is disconnecting (``Q``)."""
 73
 74
 75@dataclass(slots=True)
 76class AcquireLoco:
 77    """Client requested a loco on a throttle (``M{T,S}+<addr><;><label>``).
 78
 79    *throttle* is the single-character throttle id (typically ``T``
 80    or ``S``). *address_token* is the raw address string from the wire,
 81    e.g. ``L3601`` (long 3601) or ``S3`` (short 3) or ``3601`` (legacy).
 82    """
 83    throttle: str
 84    address_token: str
 85    label: str
 86
 87
 88@dataclass(slots=True)
 89class ReleaseLoco:
 90    """Client released a loco (``M{T,S}-<addr><;>``)."""
 91    throttle: str
 92    address_token: str
 93
 94
 95@dataclass(slots=True)
 96class ThrottleAction:
 97    """A V/R/F/X action on an acquired loco.
 98
 99    ``M{T,S}A<address_token><;><action><value>``:
100
101    * ``V<0..126>`` — set speed (126 = full, 1 = e-stop sentinel, 0 = stop)
102    * ``R<0|1>`` — direction (1 = forward)
103    * ``F<state><fn>`` — function: state 1 = press, 0 = release
104    * ``f<state><fn>`` — function force-state (set explicitly)
105    * ``X`` — e-stop
106    * ``I`` — idle (speed 0)
107    * ``L`` — set speed steps; we ignore for alpha
108    """
109    throttle: str
110    address_token: str
111    action: str   # one of: "V", "R", "F", "f", "X", "I", "L"
112    value: str    # remainder of the action; may be empty (e.g. for "X")
113
114
115@dataclass(slots=True)
116class TurnoutAction:
117    """Client asked us to change a turnout: ``PTA<action><sysName>``.
118
119    * ``action == "T"`` — throw
120    * ``action == "C"`` — close
121    * ``action == "2"`` — toggle
122    """
123    action: str          # "T" | "C" | "2"
124    system_name: str
125
126
127@dataclass(slots=True)
128class Unknown:
129    """An inbound line we don't recognise — preserve it for logging."""
130    raw: str
131
132
133# ----- turnout state codes (match jmri.Turnout Java constants) -------------
134
135TURNOUT_UNKNOWN      = 1
136TURNOUT_CLOSED       = 2
137TURNOUT_THROWN       = 4
138TURNOUT_INCONSISTENT = 8
139
140
141# ----- decoder -------------------------------------------------------------
142
143
144def decode_line(line: str) -> object:
145    """Parse one WiThrottle inbound line. Returns one of the typed messages."""
146    line = line.rstrip("\r\n")
147    if not line:
148        return Unknown(raw="")
149    head = line[0]
150    if head == "*":
151        return Heartbeat()
152    if head == "Q":
153        return Quit()
154    if head == "N" and len(line) > 1:
155        return HelloName(name=line[1:])
156    if head == "H" and line.startswith("HU"):
157        return HelloHardwareId(hardware_id=line[2:])
158    if head == "M":
159        return _decode_throttle(line)
160    if line.startswith("PTA") and len(line) >= 4:
161        return TurnoutAction(action=line[3], system_name=line[4:])
162    return Unknown(raw=line)
163
164
165def _decode_throttle(line: str) -> object:
166    """``M{T,S}<op><addr><;><payload>``."""
167    if len(line) < 4:
168        return Unknown(raw=line)
169    throttle = line[1]
170    op = line[2]
171    rest = line[3:]
172    # The wire uses "<;>" (the literal three characters) as a separator.
173    sep = "<;>"
174    if sep in rest:
175        addr_token, payload = rest.split(sep, 1)
176    else:
177        addr_token, payload = rest, ""
178    if op == "+":
179        return AcquireLoco(throttle=throttle, address_token=addr_token, label=payload)
180    if op == "-":
181        return ReleaseLoco(throttle=throttle, address_token=addr_token)
182    if op == "A":
183        if not payload:
184            return Unknown(raw=line)
185        action = payload[0]
186        value = payload[1:]
187        return ThrottleAction(
188            throttle=throttle, address_token=addr_token,
189            action=action, value=value,
190        )
191    return Unknown(raw=line)
192
193
194# ----- address-token parsing ----------------------------------------------
195
196
197def parse_address_token(token: str) -> tuple[int, bool]:
198    """Decode WiThrottle's address-token format.
199
200    * ``L<n>`` — explicit long address ``n``
201    * ``S<n>`` — explicit short address ``n``
202    * bare digits — NMRA convention: addresses ``1..127`` are short,
203      ``≥128`` are long. Older clients (and some legacy code paths in
204      Engine Driver itself) drop the prefix; matching JMRI's behaviour,
205      we infer the type from the numeric range so high-address locos
206      get the correct DCC encoding.
207
208    Returns ``(address, is_long)``.
209    """
210    if not token:
211        raise ValueError("empty address token")
212    if token[0] in ("L", "l"):
213        return int(token[1:]), True
214    if token[0] in ("S", "s"):
215        return int(token[1:]), False
216    addr = int(token)
217    # NMRA: short DCC range is 1..127; anything higher must be long.
218    return addr, addr >= 128
219
220
221def format_address_token(address: int, long_address: bool) -> str:
222    """Encode an address as a WiThrottle wire token."""
223    return f"{'L' if long_address else 'S'}{address}"
224
225
226# ----- encoders (server → client) -----------------------------------------
227
228
229def greeting(*,
230    name: str = "Python JMRI",
231    heartbeat_s: int = DEFAULT_HEARTBEAT_S,
232    web_port: int = 8000,
233) -> list[str]:
234    """Initial handshake lines a server sends right after accept."""
235    return [
236        "VN2.0",                  # protocol version 2.0
237        f"HT{name}",              # railroad name
238        f"HtPython JMRI {name}",  # display title
239        f"PW{web_port}",          # JSON server port (optional advertisement)
240        f"*{heartbeat_s}",        # request heartbeat at this cadence
241        "PPA0",                   # initial power state (clients refresh later)
242    ]
243
244
245def power_state(on: bool) -> str:
246    return f"PPA{1 if on else 0}"
247
248
249def roster_list(entries: list[dict]) -> str:
250    """``RL<count>]\\[<id><}|><address>]\\[<long|short>]\\[<id>...``.
251
252    *entries* are roster-service dicts (id, dccAddress, longAddress).
253    """
254    parts: list[str] = [f"RL{len(entries)}"]
255    for e in entries:
256        eid = str(e.get("id") or "")
257        addr = str(e.get("dccAddress") or "")
258        long = "L" if e.get("longAddress") else "S"
259        # Wire form per JMRI's WiThrottle docs.
260        parts.append(f"{eid}}}|{{{addr}}}|{{{long}")
261    # JMRI uses "]\\[" (the literal three characters) between entries.
262    return "]\\[".join(parts)
263
264
265def throttle_acquired(throttle: str, address_token: str) -> str:
266    """Acknowledge an M+ request."""
267    return f"M{throttle}+{address_token}<;>"
268
269
270def throttle_released(throttle: str, address_token: str) -> str:
271    return f"M{throttle}-{address_token}<;>"
272
273
274def throttle_speed(throttle: str, address_token: str, speed: int) -> str:
275    return f"M{throttle}A{address_token}<;>V{speed}"
276
277
278def throttle_direction(throttle: str, address_token: str, forward: bool) -> str:
279    return f"M{throttle}A{address_token}<;>R{1 if forward else 0}"
280
281
282def throttle_function(throttle: str, address_token: str, fn: int, on: bool) -> str:
283    """``F<state><fn>`` — note state is *before* the function number."""
284    return f"M{throttle}A{address_token}<;>F{1 if on else 0}{fn}"
285
286
287def throttle_speed_step_mode(
288    throttle: str, address_token: str, mode: int = 1,
289) -> str:
290    """``M<mux>A<addr><;>s<mode>`` — 1=128, 2=28, 4=27, 8=14, 16=Motorola.
291
292    JMRI's ``MultiThrottleController.encodeSpeedStepMode``. Sent after
293    the function states so Engine Driver / iOS know how to scale the
294    slider (128-step by default on modern DCC).
295    """
296    return f"M{throttle}A{address_token}<;>s{mode}"
297
298
299def throttle_function_labels(
300    throttle: str, address_token: str, labels: list[str] | None = None,
301) -> str:
302    """``M<mux>L<addr><;>]\\[label0]\\[label1]...]\\[label28]``.
303
304    JMRI's ``MultiThrottleController.sendFunctionLabels`` — this is the
305    line Engine Driver and WiThrottle iOS need after acquire so they can
306    render the F0..F28 button bar.
307
308    The label values matter more than they look:
309
310    * **Empty** labels are interpreted by Engine Driver as "no function
311      configured" and the bar stays hidden.
312    * **Missing line** works only for manual (``L<addr>``) acquires; for
313      roster (``E<id>``) acquires the client expects labels and hides
314      the bar when none arrive.
315
316    So we always send the line with non-empty defaults (``F0``..``F28``).
317    Pass explicit *labels* when you have roster data — anything falsy in
318    the list (empty string, ``None``) falls back to the default for that
319    slot.
320    """
321    if labels is None:
322        wire_labels = list(_DEFAULT_FUNCTION_LABELS)
323    else:
324        # Merge caller-supplied labels over defaults; empty slots keep
325        # the default so the client always sees something.
326        wire_labels = list(_DEFAULT_FUNCTION_LABELS)
327        for i, lbl in enumerate(labels[: MAX_FUNCTION + 1]):
328            if lbl:
329                wire_labels[i] = lbl
330    return f"M{throttle}L{address_token}<;>" + "".join(
331        f"]\\[{lbl}" for lbl in wire_labels
332    )
333
334
335def fast_clock(local_unix_seconds: int, rate: float) -> str:
336    """``PFT<local-unix-seconds><;><rate>``.
337
338    JMRI's ``FastClockController.sendFastTimeAndRate`` sends
339    ``(UTC_millis + TZ_offset) / 1000`` — the fast-clock time as if
340    it were a unix timestamp *in local time* (i.e. UTC unix seconds
341    biased by the local timezone offset so ``client_hour == local_hour``
342    when the client renders it as UTC).
343
344    A rate of ``0.0`` is the "stopped" signal — JMRI sends a second
345    ``PFT<time><;>0.0`` after the normal broadcast when the clock is
346    held; Engine Driver / WiThrottle iOS use that to pause their own
347    display.
348    """
349    return f"PFT{int(local_unix_seconds)}<;>{rate}"
350
351
352def turnout_titles() -> str:
353    """``PTT`` — turnout state-label header.
354
355    Sent once per session, before the turnout list. Tells the client
356    which display labels to attach to each numeric state code so the
357    UI can render "Closed"/"Thrown" (or a translation) instead of
358    raw digits.
359
360    Format matches JMRI's ``TurnoutController.sendTitles``:
361    ``PTT]\\[<menu-title>}|{Turnout]\\[Closed}|{2]\\[Thrown}|{4]\\[Unknown}|{1]\\[Inconsistent}|{8``
362    """
363    return (
364        "PTT"
365        + "]\\[Turnouts}|{Turnout"
366        + f"]\\[Closed}}|{{{TURNOUT_CLOSED}"
367        + f"]\\[Thrown}}|{{{TURNOUT_THROWN}"
368        + f"]\\[Unknown}}|{{{TURNOUT_UNKNOWN}"
369        + f"]\\[Inconsistent}}|{{{TURNOUT_INCONSISTENT}"
370    )
371
372
373def turnout_list(entries: list[dict]) -> str:
374    """``PTL]\\[<sysName>}|{<userName>}|{<state>]\\[...``
375
376    JMRI's ``TurnoutController.sendList``. Each entry is a dict with:
377
378    * ``system_name`` — required, e.g. ``NT12``
379    * ``user_name`` — optional friendly label; empty string if absent
380    * ``state`` — one of :data:`TURNOUT_UNKNOWN/CLOSED/THROWN/INCONSISTENT`
381    """
382    parts: list[str] = ["PTL"]
383    for e in entries:
384        sys_name = str(e.get("system_name") or "")
385        user_name = str(e.get("user_name") or "")
386        state = int(e.get("state") or TURNOUT_UNKNOWN)
387        parts.append(f"]\\[{sys_name}}}|{{{user_name}}}|{{{state}")
388    return "".join(parts)
389
390
391def turnout_state(system_name: str, state: int) -> str:
392    """``PTA<state><sysName>`` — broadcast a turnout state change to the client.
393
394    JMRI's ``TurnoutController.sendTurnoutState``. Called whenever a
395    turnout's ``KnownState`` changes — either from a client action, a
396    layout cab, or an internal script.
397    """
398    return f"PTA{state}{system_name}"
399
400
401def fast_clock_local_unix_seconds(
402    hours: int, minutes: int, seconds: int = 0,
403    *, now: object = None,
404) -> int:
405    """Convert an ``HH:MM:SS`` fast-clock time to the local-unix value
406    JMRI's ``PFT`` payload expects.
407
408    We fold today's date onto the given HH:MM:SS to build a full
409    datetime, then bias UTC unix seconds by the current timezone offset
410    (matching JMRI's ``(fastClockTime + timeZoneOffset) / 1000L``).
411    """
412    import datetime  # noqa: PLC0415
413    import time as _time  # noqa: PLC0415
414    ref = now if isinstance(now, datetime.datetime) else datetime.datetime.now()
415    ref = ref.replace(hour=hours, minute=minutes, second=seconds, microsecond=0)
416    utc_unix = ref.timestamp()
417    # `time.timezone` is seconds west of UTC (positive); we want the
418    # opposite sign. `altzone` applies when DST is in effect.
419    if _time.localtime(utc_unix).tm_isdst > 0 and _time.daylight:
420        offset = -_time.altzone
421    else:
422        offset = -_time.timezone
423    return int(utc_unix + offset)