scanengine-3/tools/helios_lds_probe.py

#!/usr/bin/env python3
"""Raw-wire probe for the Helios pump-diode current (LDS).

Why this exists: the laser panel reports a diode current of 32 mA that no
Set will change — and 32 is also what a status register reads with bit 5
set (LER: "Over voltage laser diode", LCE: "Door switch open", CCE:
"Q-switch under/over temperature").  So either the controller really holds
LDS = 32 and is refusing to take a new value, or the line the driver reads
as LDS's answer belongs to some other query.  Only the wire can say which,
and the driver cannot show it: it parses replies, and parsing is the thing
in question.

Nothing here reuses the driver's reply matching.  Every byte the controller
sends is printed as it arrives, with the command that preceded it, so the
transcript answers "what does LDS actually reply?" directly.

Usage:
    python3 tools/helios_lds_probe.py --port /dev/ttyUSB0
    python3 tools/helios_lds_probe.py --port /dev/ttyUSB0 --current 900
    python3 tools/helios_lds_probe.py --port /dev/ttyUSB0 --read-only

Safety: LDS sets the pump diode's pulse current.  It does not start
emission — that needs LDO 1 — and this probe never writes LDO.  If it finds
the laser already enabled it refuses to write anything unless --force is
given, since changing the current under emission changes the output.
"""

from __future__ import annotations

import argparse
import sys
import time
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parent.parent))

from hardware.helios_registers import (      # noqa: E402
    CCE_FLAGS, LCE_FLAGS, LER_FLAGS, SEVERITY_LABEL, decode_register,
)
from hardware.serial_util import open_8n1    # noqa: E402

REGISTERS = {"LER": LER_FLAGS, "LCE": LCE_FLAGS, "CCE": CCE_FLAGS}

# Reads that should not change anything, in the order the panel's poll
# issues them, plus the two the panel never asks for: LDG (pulse mode, which
# is what decides whether LDS is applied at all) and LMA — whose unit column
# in the manual says mA, but which this controller answers in m°C, so it is
# a resonator temperature and not a second current reading.
READ_SWEEP = ["LER", "LCE", "CCE", "LDO", "LDG", "LDF", "LDS", "LMA", "HTR"]

QUIET_S = 0.4       # a reply is over once the line is idle this long
LISTEN_S = 2.5      # ...but never wait longer than this for one


def exchange(ser, command: str, quiet_s: float = QUIET_S) -> list[tuple[float, bytes]]:
    """Send `command` and return every chunk that comes back, with timings.

    No parsing, no line matching: the point is to see what the controller
    sends, including anything the driver would have discarded.
    """
    ser.reset_input_buffer()
    ser.reset_output_buffer()
    t0 = time.monotonic()
    ser.write((command + "\r").encode("ascii"))
    ser.flush()

    chunks: list[tuple[float, bytes]] = []
    last = time.monotonic()
    while True:
        now = time.monotonic()
        if now - t0 >= LISTEN_S:
            break
        waiting = ser.in_waiting
        if waiting:
            chunks.append((now - t0, ser.read(waiting)))
            last = time.monotonic()
        elif now - last >= quiet_s:
            break
        else:
            time.sleep(0.01)
    return chunks


def show(command: str, chunks) -> str:
    """Print one exchange and return the reply as text."""
    raw = b"".join(c for _, c in chunks)
    print(f"\n  > {command}")
    if not raw:
        print("    (no reply)")
        return ""
    for offset, chunk in chunks:
        print(f"    +{offset * 1000:6.0f} ms  {chunk!r}")
    text = raw.decode("ascii", errors="replace")
    lines = [ln.strip() for ln in text.replace("\r", "\n").split("\n") if ln.strip()]
    for line in lines:
        print(f"    line: {line!r}")
    return text


def answers_for(command: str, reply: str) -> bool:
    """True if some line of `reply` names `command` — i.e. it is its answer."""
    mnemonic = command.split()[0].upper()
    for line in reply.replace("\r", "\n").split("\n"):
        head = line.strip().split("=")[0].split()
        if head and head[0].upper() == mnemonic:
            return True
    return False


def value_of(command: str, reply: str) -> int | None:
    """The integer this reply reports for `command`, if it reports one."""
    mnemonic = command.split()[0].upper()
    for line in reply.replace("\r", "\n").split("\n"):
        head, sep, tail = line.strip().partition("=")
        if not sep or head.split()[:1] != [mnemonic]:
            continue
        fields = tail.split()
        if fields:
            try:
                return int(fields[0])
            except ValueError:
                return None
    return None


def decode(name: str, value: int | None):
    if value is None:
        print(f"    {name}: no numeric value in the reply")
        return
    active = decode_register(REGISTERS[name], value)
    print(f"    {name} = {value} (0x{value:04X})"
          + ("  — no flags set" if not active else ""))
    for bit, sev, desc, comment in active:
        print(f"      bit {bit:>2} ({1 << bit:>5}) {SEVERITY_LABEL.get(sev, '[    ]')} "
              f"{desc} — {comment}")


def main() -> int:
    ap = argparse.ArgumentParser(description=__doc__,
                                 formatter_class=argparse.RawDescriptionHelpFormatter)
    ap.add_argument("--port", required=True, help="serial device, e.g. /dev/ttyUSB0")
    ap.add_argument("--current", type=int, default=900,
                    help="LDS value to try writing (mA, default 900)")
    ap.add_argument("--read-only", action="store_true",
                    help="query only; write nothing")
    ap.add_argument("--force", action="store_true",
                    help="write LDS even if the laser reports itself enabled")
    args = ap.parse_args()

    ser = open_8n1(args.port, baudrate=9600, timeout=1.0)
    time.sleep(0.2)
    ser.reset_input_buffer()
    print(f"Helios probe on {args.port} — 9600 8N1\n")
    print("=" * 70)
    print("READ SWEEP — what each query actually answers")
    print("=" * 70)

    replies: dict[str, str] = {}
    for command in READ_SWEEP:
        replies[command] = show(command, exchange(ser, command))
        time.sleep(0.1)

    print("\n" + "=" * 70)
    print("STATUS REGISTERS")
    print("=" * 70)
    before = {}
    for name in REGISTERS:
        before[name] = value_of(name, replies[name])
        decode(name, before[name])

    lds_before = value_of("LDS", replies["LDS"])
    print("\n" + "=" * 70)
    print("LDS")
    print("=" * 70)
    if not replies["LDS"]:
        print("    LDS answered nothing — it may be write-only on this firmware,")
        print("    and the panel's read-back is coming from somewhere else.")
    elif not answers_for("LDS", replies["LDS"]):
        print("    The reply to LDS does not name LDS.  That line belongs to")
        print("    another command: the read-back is misaligned, not the laser.")
        print(f"    Reply was: {replies['LDS']!r}")
    else:
        print(f"    LDS reads back as {lds_before} mA, and the reply names LDS,")
        print("    so this is the controller's own value — not a stray line.")

    if args.read_only:
        ser.close()
        return 0

    enabled = value_of("LDO", replies["LDO"])
    if enabled == 1 and not args.force:
        print("\nLDO reads 1 — the laser is enabled and emitting.  Not writing")
        print("LDS; re-run with --force if changing the current now is intended.")
        ser.close()
        return 1

    print("\n" + "=" * 70)
    print(f"WRITE TEST — LDS {args.current}")
    print("=" * 70)
    show(f"LDS {args.current}", exchange(ser, f"LDS {args.current}"))
    time.sleep(0.3)
    after_reply = show("LDS", exchange(ser, "LDS"))
    lds_after = value_of("LDS", after_reply)

    print("\n  Registers after the write (bit 2 = command error, bit 15 = range error):")
    for name in REGISTERS:
        value = value_of(name, show(name, exchange(ser, name)))
        decode(name, value)
        if before[name] is not None and value is not None and value != before[name]:
            print(f"      ^ changed from {before[name]} — the write set this")

    print("\n" + "=" * 70)
    print("VERDICT")
    print("=" * 70)
    if lds_after == args.current:
        print(f"    The controller took {args.current} mA.  If the panel still shows")
        print("    the old value, the problem is in the GUI, not on the wire.")
    elif lds_after == lds_before:
        print(f"    The controller kept {lds_before} mA and ignored the write.")
        print("    Check the flags above: a latched critical error (reset with")
        print("    CCE 0 / LCE 0 / LER 0) or a pulse mode that does not apply a")
        print("    pulse current are the two documented reasons for that.")
    else:
        print(f"    LDS went from {lds_before} to {lds_after} — neither the old")
        print(f"    value nor the {args.current} mA that was asked for.")

    ser.close()
    return 0


if __name__ == "__main__":
    sys.exit(main())