Helios panel: refresh about every 0.65 s, and yield the port to the operator

Three things were making the panel feel slow, all of them waiting rather
than working.

The trailing quiet window is waited out once per query, so it set the pace
of the whole sweep: at 50 ms that was 400 ms of a 590 ms poll spent
listening to silence. A reply streams at the baud rate — ~1 ms between
bytes, no measurable gap between its lines — and the deadline restarts on
every line read, so 20 ms outlasts the gap it exists for twenty times over.
A tail that still arrives late is caught by _discard_input(), which is what
actually protects the next query. Replayed against the rig transcript, a
sweep goes from 590 ms to 356 ms.

The poll interval was 1 s on top of that, so a value could be 1.6 s stale.
At 0.3 s the panel comes round about every 0.65 s.

And a poll held the port for its whole sweep, so a button pressed during
one waited for all eight queries. _work_pending() on the worker base lets a
poll drop what is left as soon as the operator queues something: a click
now waits ~135 ms for the register read in progress instead of the full
sweep, and the rest is picked up next time round.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

KNOWN_ISSUES.md

@@ -25,11 +25,12 @@ anywhere in this repo's protocol notes. `helios_test_app.py` called it
 anyway and raised `AttributeError` into a popup; the button is now disabled
 and the handler reports the gap instead.
 
-**Bench check:** find the power-read command in the Helios manual (the
+**Answered (2026-09-08):** the mnemonic is `HMP` (Table 6-3, "Laser Power
-other reads are three-letter mnemonics like `LDO`, `LDS`, `LTA`). If one
+Monitor", read-only, 0-5000 mW) — but the table adds "not available on all
-exists, add `get_power_mw()` to `hardware/helios_laser.py` using
+models", and the operator confirms this rig's laser has no power meter.  So
-`_query_int`, then re-enable the button. If it doesn't, delete the Power
+the button stays disabled; what remains is to delete the Power Monitoring
-Monitoring group from the test app and fix the README.
+group from `helios_test_app.py` and drop the power-monitoring claims from
+`docs/hardware/HELIOS_DRIVER_README.md`.
 
 ## Genesis laser: forked protocol implementations disagree
 

gui/qt_workers.py

@@ -44,6 +44,18 @@ class QueueWorker(QObject):
     def stop_worker(self):
         self._cmd_q.put(_STOP)
 
+    # ── Worker-side helpers ───────────────────────────────────────────────────
+
+    def _work_pending(self) -> bool:
+        """True if the operator is waiting on something.
+
+        A poll is one queue item that can hold the port for hundreds of
+        milliseconds; a button pressed during one should not have to wait
+        for the whole sweep to finish.  A poll that checks this between
+        reads gives the port up and picks the rest up next time round.
+        """
+        return not self._cmd_q.empty()
+
     # ── Worker loop ───────────────────────────────────────────────────────────
 
     @pyqtSlot()

hardware/helios_laser.py

@@ -34,6 +34,7 @@ class HeliosLaser:
         self.timeout = timeout
         self.serial = None
         self.is_connected = False
+        self._rx = bytearray()      # bytes read off the port, not yet a line
 
     @staticmethod
     def list_available_ports() -> List[str]:
@@ -52,6 +53,7 @@ class HeliosLaser:
 
         try:
             self.serial = open_8n1(self.port, baudrate=9600, timeout=self.timeout)
+            self._rx.clear()
             time.sleep(0.1)  # Allow time for connection to stabilize
             self.is_connected = True
             logger.info(f"Connected to Helios laser on {self.port}")
@@ -92,44 +94,210 @@ class HeliosLaser:
             logger.error(f"Failed to send command '{command}': {e}")
             return False
 
+    # A reply can run to more than one line.  Every status-register query
+    # answers with the value and then a decode line:
+    #
+    #     LCE = 2
+    #     Bit 15..0: 0000 0000 0000 0010
+    #
+    # At 9600 baud those trailing ~30 characters are still on the wire when
+    # read_until() returns the first line, so reset_input_buffer() cannot
+    # drop them.  Left there they become the next query's "answer", and
+    # every reply after that is one line behind — a register read reported
+    # as a "Bit 15..0" string, and the reads around it timing out on a
+    # leading blank line.  So: match a reply to the command that asked for
+    # it, and read off the rest of it before the next command goes out.
+    # How long the line has to stay silent before a reply counts as over.
+    # It is waited out once per query, so it sets the pace of the whole
+    # status poll: at 50 ms that was 400 ms of a 590 ms poll spent listening
+    # to nothing.  A reply streams at the baud rate — ~1 ms between bytes,
+    # no measurable gap between its lines — and the deadline restarts on
+    # every line, so 20 ms is twenty times the gap it has to outlast.  A
+    # tail that still arrives late is caught by _discard_input() rather
+    # than by waiting longer here.
+    TRAILING_QUIET_S = 0.02
+    MAX_REPLY_LINES = 8
+
+    def _discard_input(self):
+        """Drop anything unread, on the wire and already taken off it."""
+        self._rx.clear()
+        self.serial.reset_input_buffer()
+
+    def _read_line(self) -> Optional[str]:
+        """One line, however it is framed; None if nothing came in time.
+
+        The controller ends every line with CRLF and pads a reply with a
+        blank line or two:
+
+            b'LDS =    100 mA\r\n\r\n'
+
+        Reading up to CR alone leaves the trailing LF behind, and the next
+        read then waits out the whole port timeout for a CR that will not
+        come until some later command is answered.  That was a second of
+        dead air per query — a status poll took ~8.6 s against the 1 s
+        interval that schedules it — and worse, a query that spends its
+        deadline blocked gives up while its own reply is still arriving.
+        The next query then flushes the port mid-line, and the fragment it
+        reads is a bare number: "LCE =     32" cut after the "=" is where a
+        diode current of 32 mA came from.
+        """
+        deadline = time.monotonic() + self.timeout
+        while True:
+            cut = min((i for i in (self._rx.find(b'\r'), self._rx.find(b'\n'))
+                       if i >= 0), default=-1)
+            if cut >= 0:
+                line = bytes(self._rx[:cut])
+                # CRLF is one terminator, not an empty line between two.
+                end = cut + (2 if self._rx[cut:cut + 2] == b'\r\n' else 1)
+                del self._rx[:end]
+                return line.decode('ascii', errors='replace').strip()
+            if time.monotonic() >= deadline:
+                return None
+            chunk = self.serial.read(self.serial.in_waiting or 1)
+            if not chunk:
+                return None             # port timeout: nothing more is coming
+            self._rx += chunk
+
+    def _read_pending_lines(self) -> List[str]:
+        """Every further line the controller sends before the line goes quiet."""
+        lines: List[str] = []
+        deadline = time.monotonic() + self.TRAILING_QUIET_S
+        while True:
+            # What has already arrived is read whatever the quiet window
+            # says: the window is for deciding when to stop waiting, not
+            # for leaving a line in the buffer to confuse the next query.
+            if self._rx or self.serial.in_waiting:
+                line = self._read_line()
+                if line is None:
+                    return lines            # a partial line, nothing behind it
+                if line:
+                    lines.append(line)
+                deadline = time.monotonic() + self.TRAILING_QUIET_S
+                continue
+            if time.monotonic() >= deadline:
+                return lines
+            time.sleep(0.005)
+
+    # The only replies that come back without naming what they answer.
+    # Every other line has to identify itself: an unlabelled number is not
+    # evidence that it is *this* register's number, and taking one on faith
+    # is how a status register's value ends up displayed as a diode current.
+    UNLABELLED_REPLIES = frozenset({"CSR", "HSR"})
+
+    @classmethod
+    def _value_in(cls, line: str, mnemonic: str) -> Optional[str]:
+        """The value `line` holds for `mnemonic`, or None if it isn't its reply.
+
+        The controller answers "LDF = 20000 ns".  A line naming a different
+        mnemonic is the tail of an earlier reply, and "Bit 15..0: ..." is a
+        status register's decode line; neither is an answer to this query.
+        A line naming nothing counts only for the serial numbers, which is
+        the one reply known to come back bare.
+        """
+        head, sep, tail = line.partition('=')
+        if sep:
+            named = head.split()
+            if named and named[0].upper() != mnemonic:
+                return None
+            fields = tail.split()   # drop the unit suffix ("ns", "mA", "m°C")
+            return fields[0] if fields else None
+        if line.lower().startswith("bit"):
+            return None
+        fields = line.split()
+        if fields and fields[0].upper() == mnemonic:
+            return fields[1] if len(fields) > 1 else None
+        if mnemonic in cls.UNLABELLED_REPLIES:
+            return line
+        return None
+
     def _query(self, command: str) -> Optional[str]:
         """Send a query and return the value from its response.
 
-        Reads until the CR terminator rather than sleeping a fixed interval:
+        Reads until this command's reply arrives rather than sleeping a fixed
-        the device usually answers in a few ms, so the old unconditional
+        interval: the device usually answers in a few ms, so the old
-        0.05 + 0.2 s cost ~250 ms per query and made an 8-query status poll
+        unconditional 0.05 + 0.2 s cost ~250 ms per query and made an 8-query
-        take ~2 s — longer than the 1 s interval that scheduled it.
+        status poll take ~2 s — longer than the 1 s interval that scheduled
+        it.
         """
+        fields = command.split()
+        mnemonic = fields[0].upper() if fields else ""
         try:
-            # Clear any stale bytes so a previous timed-out reply can't be
+            # Anything volunteered while the port was idle answers no command.
-            # mistaken for this command's response.
+            self._discard_input()
-            self.serial.reset_input_buffer()
 
             if not self._send_command(command):
                 return None
 
-            response = self.serial.read_until(b'\r').decode('ascii', errors='replace').strip()
+            deadline = time.monotonic() + self.timeout
-            if not response:
+            for _ in range(self.MAX_REPLY_LINES):
+                line = self._read_line()
+                if line is None:
+                    break               # nothing arrived within the timeout
+                if line:
+                    logger.debug(f"Query '{command}' line: {line!r}")
+                    value = self._value_in(line, mnemonic)
+                    if value is not None:
+                        for extra in self._read_pending_lines():
+                            logger.debug(f"Query '{command}' trailing: {extra!r}")
+                        return value
+                if time.monotonic() >= deadline:
+                    break
+
             logger.warning(f"Query '{command}' timed out after {self.timeout}s")
+            self._read_pending_lines()
             return None
-            logger.debug(f"Query '{command}' response: {response}")
-
-            # Helios format: "COMMAND = VALUE UNIT" — take just the value
-            if '=' in response:
-                parts = response.split('=')
-                if len(parts) >= 2:
-                    value_part = parts[1].strip()
-                    # Strip the unit suffix if present (e.g. "ns", "mA", "mW")
-                    fields = value_part.split()
-                    if fields:
-                        return fields[0]
-
-            return response
 
         except Exception as e:
             logger.error(f"Failed to read response for '{command}': {e}")
             return None
 
+    def _write_command(self, command: str) -> bool:
+        """Send a command with no value to read back, and clear whatever the
+        controller prints in acknowledgement — left in the buffer, that is
+        what the next query would read as its own answer.
+        """
+        if not self._send_command(command):
+            return False
+        try:
+            for line in self._read_pending_lines():
+                logger.debug(f"Command '{command}' reply: {line!r}")
+        except Exception as e:
+            # The command went out; only the tidy-up failed.
+            logger.error(f"Failed to read the reply to '{command}': {e}")
+        return True
+
+    # Section 6 of the operator's manual, under Syntax:
+    #
+    #     Commands or set values can be discarded by the controller
+    #     unintentionally.  It is recommended to query the set value after
+    #     the command is entered to confirm the actual value.
+    #
+    # (The command table repeats it: "Query the command to confirm it was
+    # accepted.")  A setter that only writes therefore cannot report whether
+    # it worked, and the panel's next status poll reads back the old value —
+    # which looks exactly like the GUI refusing the operator's number.
+    SET_RETRIES = 3
+    SET_SETTLE_S = 0.02         # let the controller store it before reading
+
+    def _write_verified(self, mnemonic: str, value: int) -> bool:
+        """Write `value` to `mnemonic`, and confirm the controller took it.
+
+        Returns False if the read-back never matches, leaving the controller
+        holding whatever value it kept — the caller is expected to say so
+        rather than let the discarded write pass for a successful one.
+        """
+        for attempt in range(1, self.SET_RETRIES + 1):
+            if not self._write_command(f"{mnemonic} {value}"):
+                return False
+            time.sleep(self.SET_SETTLE_S)
+            readback = self._query_int(mnemonic)
+            if readback == value:
+                return True
+            logger.warning(
+                f"'{mnemonic} {value}' not accepted: controller reports "
+                f"{readback} (attempt {attempt}/{self.SET_RETRIES})")
+        return False
+
     def _query_int(self, command: str) -> Optional[int]:
         """Query a value that should parse as an int; None if absent/unparseable."""
         raw = self._query(command)
@@ -155,27 +323,35 @@ class HeliosLaser:
             logger.error(f"Period {period_ns} ns out of range (8000-60000)")
             return False
 
-        command = f"LDF {period_ns}"
+        return self._write_verified("LDF", period_ns)
-        return self._send_command(command)
 
     def set_current_ma(self, current: int) -> bool:
-        """Set pump diode current in mA."""
+        """Set pump diode pulse current (LDS) in mA.
+
+        False means the controller did not take the value — see
+        _write_verified.  The manual's range for LDS is 0-7000 mA; the
+        2000 mA ceiling here is this rig's limit, not the protocol's.
+        """
         if not (0 <= current <= 2000):
             logger.error(f"Current {current} mA out of range (0-2000)")
             return False
 
-        command = f"LDS {current}"
+        return self._write_verified("LDS", current)
-        return self._send_command(command)
 
     def set_pulse_mode(self, mode: PulseMode) -> bool:
-        """Set pulse mode."""
+        """Set pulse mode.
+
+        Note from the manual's LDG entry: "LDF has to be set again after LDG
+        is changed, except for single pulse triggering" — so a caller that
+        changes the mode has to re-send the frequency.
+        """
         command = f"LDG {mode.value}"
-        return self._send_command(command)
+        return self._write_command(command)
 
     def set_laser_enable(self, enable: bool) -> bool:
         """Enable or disable laser emission."""
         command = f"LDO {1 if enable else 0}"
-        success = self._send_command(command)
+        success = self._write_command(command)
 
         if success:
             state = "enabled" if enable else "disabled"
@@ -243,11 +419,11 @@ class HeliosLaser:
             True if all three commands sent successfully
         """
         ok = True
-        ok = self._send_command("CCE 0") and ok
+        ok = self._write_command("CCE 0") and ok
         time.sleep(0.1)
-        ok = self._send_command("LCE 0") and ok
+        ok = self._write_command("LCE 0") and ok
         time.sleep(0.1)
-        ok = self._send_command("LER 0") and ok
+        ok = self._write_command("LER 0") and ok
         if ok:
             logger.info("Fault reset sequence sent")
         return ok
@@ -258,18 +434,22 @@ class HeliosLaser:
         return None if value is None else value == 1
 
     def send_raw_command(self, command: str) -> Optional[str]:
-        """Send a raw command and return the unparsed response (diagnostics)."""
+        """Send a raw command and return its whole unparsed reply (diagnostics).
+
+        Every line comes back, the "Bit 15..0: ..." decode line included:
+        seeing the entire reply is the point of the raw console.
+        """
         if not self.is_connected or not self.serial:
             logger.error("Not connected to laser")
             return None
         try:
-            self.serial.reset_input_buffer()
+            self._discard_input()
             if not self._send_command(command):
                 return None
-            raw = self.serial.read_until(b'\r')
+            first = self._read_line()
-            if not raw:
+            lines = [first] if first else []
-                raw = self.serial.read(self.serial.in_waiting)
+            lines += self._read_pending_lines()
-            return raw.decode('ascii', errors='replace').strip()
+            return "\n".join(lines)
         except Exception as e:
             logger.error(f"send_raw_command error: {e}")
             return None
@@ -277,7 +457,7 @@ class HeliosLaser:
     def set_remote_enable(self, enable: bool) -> bool:
         """Set the remote enable state (LRE - utility connector pin 8)."""
         command = f"LRE {1 if enable else 0}"
-        return self._send_command(command)
+        return self._write_command(command)
 
     # No __del__: it used to call disconnect(), which disables the laser and
     # writes to the serial port from the garbage collector at an

hardware/helios_registers.py

@@ -0,0 +1,102 @@
+"""Helios status-register bit definitions (Tables 8-1, 8-2, 8-3).
+
+Kept out of the driver and out of the Qt apps so that a command-line
+diagnostic can decode a register without importing either.
+
+Each entry: bit_number -> (severity, description, comment)
+  severity: 'C' = critical error, 'S' = status, 'I' = input error, '' = none
+"""
+
+LER_FLAGS = {
+    0:  ('C', 'Controller temperature failure (resonator/SHG/q-switch)',
+         'Check CCE register for details'),
+    1:  ('S', 'Trigger input active',
+         'High when trigger signal applied or laser in continuous pulsing'),
+    2:  ('I', 'Command error',
+         'Unknown command sent to controller'),
+    3:  ('C', 'Laser disable pin open (utility connector)',
+         'Shuts down pump diodes; reset LER 0 required to restart'),
+    4:  ('C', 'Internal hardware failure',
+         'Contact Coherent'),
+    5:  ('C', 'Over voltage laser diode',
+         'Check for open circuit or voltage spikes'),
+    6:  ('C', 'Internal hardware failure',
+         'Contact Coherent'),
+    7:  ('C', 'Controller temperature failure at pump diodes',
+         'Check LCE register for details'),
+    8:  ('S', 'Laser start delay (60 s warmup)',
+         'Laser cannot be started yet; status error LED flashing'),
+    9:  ('C', 'Internal hardware failure',
+         'Check environment for strong EMI; contact Coherent'),
+    10: ('C', 'Internal hardware failure',
+         'Check environment for strong EMI; contact Coherent'),
+    11: ('C', 'Internal hardware failure',
+         'Check environment for strong EMI; contact Coherent'),
+    12: ('S', 'Slave controller error (remote input)',
+         'Valid only for master controller coupled with a slave'),
+    13: ('',  'Laserhead not found',
+         'Head not connected / not found; check EMI; ignore for double-electronic slave'),
+    14: ('',  'Laserhead I\u00b2C acknowledge error',
+         'Check environment for strong EMI; ignore for double-electronic slave'),
+    15: ('S', 'Range-Error (not critical)',
+         'Input value out of range'),
+}
+
+LCE_FLAGS = {
+    0:  ('C', 'Pump diode over/under temperature',
+         'Limit exceeded (<10\u00b0C or >60\u00b0C); check head cooling'),
+    1:  ('C', 'Internal hardware failure',
+         'Contact Coherent'),
+    2:  ('C', 'Pump diode temperature out of range',
+         'Actual temp >2\u00b0C off setpoint for >1 min'),
+    3:  ('C', 'Pump diode current critical',
+         'Current set too close to current limit'),
+    4:  ('C', 'Pump diode temperature out of limit',
+         'Pump diode temperature is out of limit'),
+    5:  ('S', 'Door switch open',
+         'Close utility connector pin 2 permanently to pin 9 (GND)'),
+    7:  ('C', 'Pump diode NTC error',
+         'Invalid temperature measured or NTC broken'),
+    8:  ('C', 'Laser diode power stage over temperature',
+         'Temp <10\u00b0C or >65\u00b0C at controller; check controller cooling'),
+    9:  ('C', 'Internal hardware failure',
+         'Check environment for strong EMI; contact Coherent'),
+    10: ('C', 'Internal hardware failure',
+         'Check environment for strong EMI; contact Coherent'),
+    11: ('C', 'Internal hardware failure',
+         'Check environment for strong EMI; contact Coherent'),
+    15: ('S', 'Range-Error (not critical)',
+         'Input value out of range'),
+}
+
+CCE_FLAGS = {
+    0:  ('C', 'Resonator/SHG under/over temperature',
+         'Limit exceeded (<10\u00b0C or >60\u00b0C); temperature controller deactivated'),
+    1:  ('C', 'Resonator/SHG NTC failure',
+         'Temperature sensor broken or disconnected'),
+    2:  ('C', 'Resonator/SHG temperature out of range',
+         'Actual temp >2\u00b0C off setpoint for >1 min'),
+    3:  ('C', 'Q-switch ADC / temperature readout failure',
+         'Internal hardware error or no NTC connected'),
+    4:  ('C', 'Q-switch temperature out of range',
+         'Actual temp >2\u00b0C off setpoint for >1 min'),
+    5:  ('C', 'Q-switch under/over temperature',
+         'Limit exceeded (<10\u00b0C or >60\u00b0C); temperature controller deactivated'),
+    7:  ('C', 'Q-switch NTC failure',
+         'Internal hardware error or no NTC connected'),
+    8:  ('C', 'Internal hardware failure',
+         'Contact Coherent'),
+    15: ('S', 'Range-Error (not critical)',
+         'Input value out of range'),
+}
+
+SEVERITY_LABEL = {'C': '[CRIT]', 'S': '[STAT]', 'I': '[INPT]', '': '[INFO]'}
+
+
+def decode_register(flags_dict: dict, value: int) -> list:
+    """Return list of (bit, severity, description, comment) for each set bit."""
+    active = []
+    for bit, (sev, desc, comment) in flags_dict.items():
+        if value & (1 << bit):
+            active.append((bit, sev, desc, comment))
+    return active

helios_test_app.py

@@ -16,106 +16,9 @@ from PyQt6.QtCore import QThread, pyqtSignal, pyqtSlot, QObject
 from PyQt6.QtGui import QFont
 
 from hardware.helios_laser import HeliosLaser, PulseMode
-
+from hardware.helios_registers import (
-# ---------------------------------------------------------------------------
+    CCE_FLAGS, LCE_FLAGS, LER_FLAGS, SEVERITY_LABEL, decode_register,
-# Status register bit definitions (Tables 8-1, 8-2, 8-3 — Helios manual)
+)
-# Each entry: bit_number -> (severity, description, comment)
-#   severity: 'C' = critical error, 'S' = status, 'I' = input error, '' = none
-# ---------------------------------------------------------------------------
-_LER_FLAGS = {
-    0:  ('C', 'Controller temperature failure (resonator/SHG/q-switch)',
-         'Check CCE register for details'),
-    1:  ('S', 'Trigger input active',
-         'High when trigger signal applied or laser in continuous pulsing'),
-    2:  ('I', 'Command error',
-         'Unknown command sent to controller'),
-    3:  ('C', 'Laser disable pin open (utility connector)',
-         'Shuts down pump diodes; reset LER 0 required to restart'),
-    4:  ('C', 'Internal hardware failure',
-         'Contact Coherent'),
-    5:  ('C', 'Over voltage laser diode',
-         'Check for open circuit or voltage spikes'),
-    6:  ('C', 'Internal hardware failure',
-         'Contact Coherent'),
-    7:  ('C', 'Controller temperature failure at pump diodes',
-         'Check LCE register for details'),
-    8:  ('S', 'Laser start delay (60 s warmup)',
-         'Laser cannot be started yet; status error LED flashing'),
-    9:  ('C', 'Internal hardware failure',
-         'Check environment for strong EMI; contact Coherent'),
-    10: ('C', 'Internal hardware failure',
-         'Check environment for strong EMI; contact Coherent'),
-    11: ('C', 'Internal hardware failure',
-         'Check environment for strong EMI; contact Coherent'),
-    12: ('S', 'Slave controller error (remote input)',
-         'Valid only for master controller coupled with a slave'),
-    13: ('',  'Laserhead not found',
-         'Head not connected / not found; check EMI; ignore for double-electronic slave'),
-    14: ('',  'Laserhead I\u00b2C acknowledge error',
-         'Check environment for strong EMI; ignore for double-electronic slave'),
-    15: ('S', 'Range-Error (not critical)',
-         'Input value out of range'),
-}
-
-_LCE_FLAGS = {
-    0:  ('C', 'Pump diode over/under temperature',
-         'Limit exceeded (<10\u00b0C or >60\u00b0C); check head cooling'),
-    1:  ('C', 'Internal hardware failure',
-         'Contact Coherent'),
-    2:  ('C', 'Pump diode temperature out of range',
-         'Actual temp >2\u00b0C off setpoint for >1 min'),
-    3:  ('C', 'Pump diode current critical',
-         'Current set too close to current limit'),
-    4:  ('C', 'Pump diode temperature out of limit',
-         'Pump diode temperature is out of limit'),
-    5:  ('S', 'Door switch open',
-         'Close utility connector pin 2 permanently to pin 9 (GND)'),
-    7:  ('C', 'Pump diode NTC error',
-         'Invalid temperature measured or NTC broken'),
-    8:  ('C', 'Laser diode power stage over temperature',
-         'Temp <10\u00b0C or >65\u00b0C at controller; check controller cooling'),
-    9:  ('C', 'Internal hardware failure',
-         'Check environment for strong EMI; contact Coherent'),
-    10: ('C', 'Internal hardware failure',
-         'Check environment for strong EMI; contact Coherent'),
-    11: ('C', 'Internal hardware failure',
-         'Check environment for strong EMI; contact Coherent'),
-    15: ('S', 'Range-Error (not critical)',
-         'Input value out of range'),
-}
-
-_CCE_FLAGS = {
-    0:  ('C', 'Resonator/SHG under/over temperature',
-         'Limit exceeded (<10\u00b0C or >60\u00b0C); temperature controller deactivated'),
-    1:  ('C', 'Resonator/SHG NTC failure',
-         'Temperature sensor broken or disconnected'),
-    2:  ('C', 'Resonator/SHG temperature out of range',
-         'Actual temp >2\u00b0C off setpoint for >1 min'),
-    3:  ('C', 'Q-switch ADC / temperature readout failure',
-         'Internal hardware error or no NTC connected'),
-    4:  ('C', 'Q-switch temperature out of range',
-         'Actual temp >2\u00b0C off setpoint for >1 min'),
-    5:  ('C', 'Q-switch under/over temperature',
-         'Limit exceeded (<10\u00b0C or >60\u00b0C); temperature controller deactivated'),
-    7:  ('C', 'Q-switch NTC failure',
-         'Internal hardware error or no NTC connected'),
-    8:  ('C', 'Internal hardware failure',
-         'Contact Coherent'),
-    15: ('S', 'Range-Error (not critical)',
-         'Input value out of range'),
-}
-
-_SEVERITY_LABEL = {'C': '[CRIT]', 'S': '[STAT]', 'I': '[INPT]', '': '[INFO]'}
-
-
-def _decode_register(flags_dict: dict, value: int) -> list:
-    """Return list of (bit, severity, description, comment) for each set bit."""
-    active = []
-    for bit, (sev, desc, comment) in flags_dict.items():
-        if value & (1 << bit):
-            active.append((bit, sev, desc, comment))
-    return active
-
 
 # Configure logging
 logging.basicConfig(level=logging.INFO)
@@ -945,20 +848,20 @@ class HeliosTestApp(QMainWindow):
         # Decode and display individual flags
         lines = []
         for reg_name, value, flags_dict in (
-            ("LER", ler, _LER_FLAGS),
+            ("LER", ler, LER_FLAGS),
-            ("LCE", lce, _LCE_FLAGS),
+            ("LCE", lce, LCE_FLAGS),
-            ("CCE", cce, _CCE_FLAGS),
+            ("CCE", cce, CCE_FLAGS),
         ):
             if value is None:
                 lines.append(f"{reg_name}: <read error>")
                 continue
-            active = _decode_register(flags_dict, value)
+            active = decode_register(flags_dict, value)
             if not active:
                 lines.append(f"{reg_name} (raw={value}): OK — no flags set")
             else:
                 lines.append(f"{reg_name} (raw={value}):")
                 for bit, sev, desc, comment in active:
-                    label = _SEVERITY_LABEL.get(sev, '[    ]')
+                    label = SEVERITY_LABEL.get(sev, '[    ]')
                     lines.append(f"  {label} bit {bit:2d} ({1 << bit:>5}): {desc}")
                     lines.append(f"             → {comment}")
         self.text_register_decode.setPlainText("\n".join(lines))

sc3-aui-helios.ui

@@ -109,6 +109,29 @@
         </property>
        </widget>
       </item>
+      <item>
+       <widget class="QLabel" name="helios_current_readback_label">
+        <property name="toolTip">
+         <string>The diode current the controller reports. The spin box is the value that will be sent when Set is pressed.</string>
+        </property>
+        <property name="text">
+         <string>laser: --- mA</string>
+        </property>
+       </widget>
+      </item>
+      <item>
+       <spacer name="horizontalSpacer_current">
+        <property name="orientation">
+         <enum>Qt::Orientation::Horizontal</enum>
+        </property>
+        <property name="sizeHint" stdset="0">
+         <size>
+          <width>40</width>
+          <height>20</height>
+         </size>
+        </property>
+       </spacer>
+      </item>
      </layout>
     </widget>
    </item>

sc3_aui_app.py

@@ -337,12 +337,15 @@ class HeliosWorker(PollingQueueWorker):
     """
     enabled_updated          = pyqtSignal(bool)
     current_updated          = pyqtSignal(int)
+    current_unknown          = pyqtSignal()
     diode_temp_updated       = pyqtSignal(float)
     pstage_temp_updated      = pyqtSignal(float)
     qswitch_temp_updated     = pyqtSignal(float)
     status_registers_updated = pyqtSignal(object, object, object)  # LER, LCE, CCE (int|None)
 
-    POLL_INTERVAL_S = 1.0
+    # A full sweep is eight queries, ~360 ms of port time.  The interval is
+    # the gap between sweeps, so the panel refreshes about every 0.65 s.
+    POLL_INTERVAL_S = 0.3
 
     def __init__(self):
         super().__init__(poll_interval_s=self.POLL_INTERVAL_S)
@@ -399,28 +402,61 @@ class HeliosWorker(PollingQueueWorker):
     def _do_set_current(self, current_ma: int):
         if not self._laser:
             return
-        self._laser.set_current_ma(current_ma)
+        if self._laser.set_current_ma(current_ma):
             self.current_updated.emit(current_ma)
+            return
+        # The manual warns the controller drops set values unintentionally,
+        # so the driver verifies the write.  Say when it did not take: this
+        # used to emit the requested value regardless, and the next poll
+        # then quietly restored the old one.
+        actual = self._laser.get_current_ma()
+        if actual is not None:
+            self.current_updated.emit(actual)
+        self.error_occurred.emit(
+            f"Diode current: controller kept {actual} mA, "
+            f"did not accept {current_ma} mA")
 
     def _poll_once(self):
         """Read the full status set.  Individual reads are allowed to fail
-        (a timed-out register shouldn't suppress the rest of the panel)."""
+        (a timed-out register shouldn't suppress the rest of the panel).
+
+        Abandoned as soon as the operator queues something: the rest of the
+        sweep is worth less than a button that responds now, and the next
+        poll will pick it up.
+        """
         if not self._laser or not self._laser.is_connected:
             return
         try:
             self.status_registers_updated.emit(*self._laser.get_status_registers())
         except Exception:
             pass
+        if self._work_pending():
+            return
         try:
             self.enabled_updated.emit(self._laser.is_laser_enabled())
         except Exception:
             pass
+        if self._work_pending():
+            return
+        # A failed LDS read must not leave the last good value on screen
+        # looking live: that is indistinguishable from a setpoint that
+        # refuses to move, which is the failure this panel exists to show.
+        try:
+            current = self._laser.get_current_ma()
+        except Exception:
+            current = None
+        if current is None:
+            self.current_unknown.emit()
+        else:
+            self.current_updated.emit(current)
+
         for getter, signal in (
-            (self._laser.get_current_ma, self.current_updated),
             (self._laser.get_diode_temp_c, self.diode_temp_updated),
             (self._laser.get_power_stage_temp_c, self.pstage_temp_updated),
             (self._laser.get_qswitch_temp_c, self.qswitch_temp_updated),
         ):
+            if self._work_pending():
+                return
             try:
                 value = getter()
                 if value is not None:
@@ -982,6 +1018,7 @@ class HeliosWindow(QWidget):
         self._worker = worker
         # Polling is driven by the worker itself (self-rescheduling), so
         # there is no timer here to outpace the device.
+        self._seed_current_spin = False
         self._wire_signals()
         self._update_controls(False)
         self._set_emission_indicator(False)
@@ -1003,6 +1040,9 @@ class HeliosWindow(QWidget):
 
         self.helios_set_current_btn.clicked.connect(self._on_set_current)
         self._worker.current_updated.connect(self._on_current_updated)
+        self._worker.current_unknown.connect(
+            lambda: self.helios_current_readback_label.setText("laser: ? mA")
+        )
         self._worker.diode_temp_updated.connect(
             lambda t: self.helios_temp_diode_label.setText(f"{t:.1f} °C")
         )
@@ -1029,6 +1069,8 @@ class HeliosWindow(QWidget):
             self._worker.queue_disconnect()
 
     def _on_connected(self):
+        # Take the setpoint from the laser once, on connect.
+        self._seed_current_spin = True
         self.helios_connect_toggle.setEnabled(True)
         self.helios_connect_toggle.setText("Disconnect")
         self.helios_status_label.setText("Connected")
@@ -1077,6 +1119,17 @@ class HeliosWindow(QWidget):
         self._worker.queue_set_current(self.helios_current_spin.value())
 
     def _on_current_updated(self, current_ma: int):
+        """Show what the laser reports; leave the spin box to the operator.
+
+        The 1 Hz poll used to write its reading straight into the spin box,
+        so a number typed there was overwritten within a second — the
+        setpoint appeared to snap back to the controller's value before it
+        could be sent.  The spin box is now seeded once per connection and
+        is the operator's alone after that; the label is the read-back.
+        """
+        self.helios_current_readback_label.setText(f"laser: {current_ma} mA")
+        if self._seed_current_spin:
+            self._seed_current_spin = False
             self.helios_current_spin.blockSignals(True)
             self.helios_current_spin.setValue(current_ma)
             self.helios_current_spin.blockSignals(False)
@@ -1101,6 +1154,8 @@ class HeliosWindow(QWidget):
             self.helios_enable_btn.setText("Enable Laser")
             self.helios_enable_btn.blockSignals(False)
             self._set_emission_indicator(False)
+            self._seed_current_spin = False
+            self.helios_current_readback_label.setText("laser: --- mA")
             self._reset_temperatures()
             self.helios_ler_label.setText("---")
             self.helios_lce_label.setText("---")

tests/test_helios_laser.py

@@ -0,0 +1,327 @@
+"""HeliosLaser: replies that span more than one line.
+
+Regression: every status-register query answers with the value *and* a
+"Bit 15..0: ..." decode line.  The driver read one line per query and threw
+the rest away with reset_input_buffer(), which at 9600 baud cannot drop
+bytes that are still on the wire — so from the first LER read onward every
+reply was one line behind, and the panel showed a register as
+"Bit 15..0: 0000 0000 0000 0010" with the reads around it timing out.
+"""
+import pytest
+
+from hardware.helios_laser import HeliosLaser, PulseMode
+
+
+# What the controller actually sends back, transcribed from a session with
+# the laser (tools/helios_lds_probe.py): CRLF line ends, the value padded
+# out to a fixed width, and one or two blank lines closing every reply.
+#
+#     b'LDS =    100 mA\r\n\r\n'
+#     b'LCE =     32\r\nBit 15..0: 0000 0000 0010 0000\r\n\r\n\r\n'
+#
+# The blank lines matter: a reader that stops at CR leaves the LF of the
+# last one behind, and the next read waits out the port timeout for a CR
+# that only the next command will bring.
+_PAD = [""]
+_REGISTER_PAD = ["", ""]
+
+REPLIES = {
+    "LDO": ["LDO =      1 "] + _PAD,
+    "LDF": ["LDF =  20000 ns"] + _PAD,
+    "LDS": ["LDS =   1500 mA"] + _PAD,
+    "LDG": ["LDG =     14 "] + _PAD,
+    "LRE": ["LRE =      0 "] + _PAD,
+    "LTA": ["LTA =  25400 m°C"] + _PAD,
+    "LTT": ["LTT =  31200 m°C"] + _PAD,
+    "EOA": ["EOA =  40100 m°C"] + _PAD,
+    "CSR": ["CSR = 1234567"] + _PAD,
+    "HSR": ["HSR = 7654321"] + _PAD,
+    # The registers are the multi-line ones.
+    "LER": ["LER =      0", "Bit 15..0: 0000 0000 0000 0000"] + _REGISTER_PAD,
+    "LCE": ["LCE =      2", "Bit 15..0: 0000 0000 0000 0010"] + _REGISTER_PAD,
+    "CCE": ["CCE =      0", "Bit 15..0: 0000 0000 0000 0000"] + _REGISTER_PAD,
+}
+
+
+class FakePort:
+    """Serial stand-in that answers like the Helios controller.
+
+    ``reset_input_buffer`` is deliberately a no-op: the rest of a reply is
+    still in flight when the driver has read its first line, so a flush
+    cannot remove it.  The driver has to stay in step by reading what it
+    asked for, not by discarding what it happens to find.
+    """
+
+    def __init__(self, replies=None, timeout=1.0, discard_writes=False):
+        self.replies = dict(REPLIES) if replies is None else replies
+        self.timeout = timeout
+        self.is_open = True
+        # "Commands or set values can be discarded by the controller
+        # unintentionally" (manual, Section 6) — the case a verified write
+        # exists to catch.
+        self.discard_writes = discard_writes
+        self.written: list[str] = []
+        self._buf = bytearray()
+
+    def _store(self, mnemonic: str, value: str):
+        """Keep a written value, so a later query reads back what was set."""
+        if self.discard_writes:
+            return
+        previous = self.replies.get(mnemonic, [f"{mnemonic} = 0"])[0]
+        unit = previous.split()[3:]     # "LDS = 1500 mA" -> ["mA"]
+        self.replies[mnemonic] = [" ".join([f"{mnemonic} =", value, *unit])]
+
+    # ── the bits of pyserial.Serial the driver uses ──────────────────────────
+
+    def write(self, data: bytes) -> int:
+        text = data.decode("ascii").strip()
+        self.written.append(text)
+        fields = text.split()
+        if len(fields) > 1:
+            self._store(fields[0].upper(), fields[1])
+        for line in self.replies.get(fields[0].upper() if fields else "", []):
+            self._buf += line.encode("utf-8") + b"\r\n"
+        return len(data)
+
+    @property
+    def in_waiting(self) -> int:
+        return len(self._buf)
+
+    def read(self, size: int = 1) -> bytes:
+        chunk = bytes(self._buf[:size])
+        del self._buf[:size]
+        return chunk
+
+    def read_until(self, expected: bytes = b"\n", size=None) -> bytes:
+        # No terminator in the buffer models the read timing out: pyserial
+        # returns whatever it has, which is b"" when nothing is pending.
+        cut = self._buf.find(expected)
+        cut = len(self._buf) if cut < 0 else cut + len(expected)
+        chunk = bytes(self._buf[:cut])
+        del self._buf[:cut]
+        return chunk
+
+    def reset_input_buffer(self):
+        """No-op — see the class docstring."""
+
+    def close(self):
+        self.is_open = False
+
+
+@pytest.fixture
+def laser():
+    """A connected driver on a fake port, with the idle wait taken out.
+
+    TRAILING_QUIET_S covers the ~30 ms a decode line spends on the wire at
+    9600 baud; the fake answers instantly, so waiting for it only slows the
+    suite down.  Zeroing it also keeps the tests honest: they pass because
+    the driver reads the whole reply, not because it waited long enough.
+    """
+    drv = HeliosLaser(timeout=1.0)
+    drv.serial = FakePort()
+    drv.is_connected = True
+    drv.TRAILING_QUIET_S = 0.0
+    drv.SET_SETTLE_S = 0.0
+    return drv
+
+
+@pytest.fixture
+def stubborn_laser():
+    """A controller that answers every query but keeps its own set values."""
+    drv = HeliosLaser(timeout=1.0)
+    drv.serial = FakePort(discard_writes=True)
+    drv.is_connected = True
+    drv.TRAILING_QUIET_S = 0.0
+    drv.SET_SETTLE_S = 0.0
+    return drv
+
+
+def test_status_registers_are_read_in_step(laser):
+    """The regression: each register gets its own value, not the previous
+    register's decode line."""
+    assert laser.get_status_registers() == (0, 2, 0)
+
+
+def test_reads_after_a_register_are_not_a_line_behind(laser):
+    """A whole status poll, in the order HeliosWorker._poll_once issues it."""
+    assert laser.get_status_registers() == (0, 2, 0)
+    assert laser.is_laser_enabled() is True
+    assert laser.get_current_ma() == 1500
+    assert laser.get_diode_temp_c() == pytest.approx(25.4)
+    assert laser.get_power_stage_temp_c() == pytest.approx(31.2)
+    assert laser.get_qswitch_temp_c() == pytest.approx(40.1)
+
+
+def test_decode_line_is_consumed_not_left_behind(laser):
+    laser.get_status_registers()
+    assert laser.serial.in_waiting == 0
+
+
+def test_unit_suffix_is_stripped(laser):
+    assert laser.get_frequency_hz() == 50000       # "LDF = 20000 ns"
+    assert laser.get_current_ma() == 1500          # "LDS = 1500 mA"
+
+
+def test_a_reply_from_an_earlier_command_is_skipped(laser):
+    """An answer already in the buffer when the query goes out belongs to
+    whoever asked for it, and must not be returned as this query's value."""
+    laser.serial._buf += b"LER = 8\r\nBit 15..0: 0000 0000 0000 1000\r\n"
+    assert laser.get_current_ma() == 1500
+
+
+def test_reply_without_a_mnemonic_is_taken_as_the_value(laser):
+    """The serial numbers come back as a bare string on some firmware."""
+    laser.serial.replies = {"CSR": ["A1B2C3D4"]}
+    assert laser.get_controller_serial() == "A1B2C3D4"
+
+
+def test_silent_device_reports_a_timeout(laser):
+    laser.serial.replies = {}
+    assert laser.get_current_ma() is None
+    assert laser._query("LTA") is None
+
+
+def test_set_commands_clear_their_acknowledgement(laser):
+    """A setter that leaves the controller's echo in the buffer desynchronises
+    the next query just as a decode line does."""
+    assert laser.set_laser_enable(True) is True
+    assert laser.serial.in_waiting == 0
+    assert laser.set_pulse_mode(PulseMode.CONTINUOUS_PULSING) is True
+    assert laser.get_current_ma() == 1500
+
+
+def test_raw_command_returns_every_line(laser):
+    """The diagnostics console is where a multi-line reply should be visible."""
+    assert laser.send_raw_command("LCE") == (
+        "LCE =      2\nBit 15..0: 0000 0000 0000 0010"
+    )
+
+
+# ── Set values the controller may discard ────────────────────────────────────
+#
+# Section 6 of the manual: "Commands or set values can be discarded by the
+# controller unintentionally.  It is recommended to query the set value
+# after the command is entered to confirm the actual value."  A write that
+# reports success without reading back leaves the panel showing a setpoint
+# the laser never took, until the next poll replaces it with the old value.
+
+def test_a_set_current_is_read_back(laser):
+    assert laser.set_current_ma(900) is True
+    assert laser.get_current_ma() == 900
+    assert "LDS 900" in laser.serial.written
+
+
+def test_a_discarded_set_current_is_retried_then_reported(stubborn_laser):
+    assert stubborn_laser.set_current_ma(900) is False
+    # Retried, not given up on after one write.
+    assert stubborn_laser.serial.written.count("LDS 900") == \
+        stubborn_laser.SET_RETRIES
+    # And the controller's own value is what it still holds.
+    assert stubborn_laser.get_current_ma() == 1500
+
+
+def test_a_set_that_takes_on_a_retry_succeeds(laser):
+    """One dropped write, then the controller accepts — still a success."""
+    real_write = laser.serial.write
+    state = {"drops": 1}
+
+    def flaky(data: bytes) -> int:
+        if data.decode("ascii").strip().startswith("LDS ") and state["drops"]:
+            state["drops"] -= 1
+            laser.serial.written.append(data.decode("ascii").strip())
+            return len(data)        # swallowed: nothing stored, no reply
+        return real_write(data)
+
+    laser.serial.write = flaky
+    assert laser.set_current_ma(900) is True
+    assert laser.get_current_ma() == 900
+
+
+def test_a_set_frequency_is_read_back(laser):
+    assert laser.set_frequency_hz(25000) is True    # 40000 ns
+    assert laser.get_frequency_hz() == 25000
+    assert "LDF 40000" in laser.serial.written
+
+
+def test_an_out_of_range_current_is_not_sent(laser):
+    assert laser.set_current_ma(9000) is False
+    assert laser.serial.written == []
+
+
+def test_an_unlabelled_number_is_not_taken_as_a_register_value(laser):
+    """A bare number answers nothing in particular.
+
+    The controller's status registers read 32 when bit 5 is set (LER "Over
+    voltage laser diode", LCE "Door switch open"), and a diode current of
+    32 mA is a perfectly ordinary-looking value — so a stray "32" must not
+    be allowed to pass for the answer to LDS.
+    """
+    laser.serial.replies = {"LDS": ["32"]}
+    assert laser.get_current_ma() is None
+
+
+def test_a_serial_number_still_comes_back_bare(laser):
+    """The one reply that legitimately names nothing."""
+    laser.serial.replies = {"CSR": ["A1B2C3D4"], "HSR": ["7654321"]}
+    assert laser.get_controller_serial() == "A1B2C3D4"
+    assert laser.get_head_serial() == "7654321"
+
+
+class SplitReplyPort(FakePort):
+    """Answers LCE in two pieces, the tail arriving after the next command.
+
+    That is what the wire looks like when a query gives up early: at 9600
+    baud the rest of the reply is still coming, and reset_input_buffer()
+    cannot drop bytes that have not arrived.  The fragment left over is
+    "     32" — the value half of "LCE =     32", which is a plausible
+    diode current and was read as one.
+    """
+
+    def __init__(self):
+        super().__init__()
+        self._late = b""
+
+    def write(self, data: bytes) -> int:
+        text = data.decode("ascii").strip()
+        self.written.append(text)
+        mnemonic = text.split()[0].upper() if text.split() else ""
+        # Whatever is asked next, the last reply's tail lands in front of it.
+        self._buf += self._late
+        self._late = b""
+        if mnemonic == "LCE":
+            self._buf += b"LCE ="           # ...and no line ending yet
+            self._late = b"     32\r\n\r\n\r\n"
+            return len(data)
+        for line in self.replies.get(mnemonic, []):
+            self._buf += line.encode("utf-8") + b"\r\n"
+        return len(data)
+
+
+def test_a_late_fragment_is_not_the_next_query_s_value():
+    """The regression this branch exists for.
+
+    LCE's reply is cut in half, so the register read gives up.  The tail
+    arrives while the *next* query is being answered, and "32" is what the
+    panel showed as the pump diode current — LCE bit 5, "Door switch open",
+    read as milliamps.
+    """
+    drv = HeliosLaser(timeout=1.0)
+    drv.serial = SplitReplyPort()
+    drv.is_connected = True
+    drv.TRAILING_QUIET_S = 0.0
+
+    assert drv._query_int("LCE") is None         # cut off mid-reply
+    assert drv.get_current_ma() == 1500          # not 32
+
+
+def test_a_reply_is_read_without_waiting_out_the_port(laser):
+    """Nothing is left in either buffer once a reply has been read.
+
+    A leftover LF costs a whole port timeout on the next read, which is
+    what made a status poll take ~8.6 s against a 1 s interval.
+    """
+    laser.timeout = 0.01        # a wait would show up as a failure below
+    assert laser.get_status_registers() == (0, 2, 0)
+    assert laser.get_current_ma() == 1500
+    assert laser.serial.in_waiting == 0
+    assert laser._rx == bytearray()

tests/test_qt_workers.py

@@ -137,6 +137,55 @@ def test_polling_never_overlaps_or_backs_up(qapp):
     assert queued <= 1, f"{queued} stale polls queued up"
 
 
+class _YieldingPoller(PollingQueueWorker):
+    """A poll made of several reads that gives up as soon as work arrives."""
+
+    def __init__(self):
+        super().__init__(poll_interval_s=0.02)
+        self.reads = 0
+        self.handled = []
+        self.is_connected = True
+        self._handlers["click"] = self._click
+
+    def _click(self, value):
+        self.handled.append(value)
+
+    def _poll_once(self):
+        for _ in range(6):
+            if self._work_pending():
+                return
+            time.sleep(0.02)
+            self.reads += 1
+
+
+def test_a_queued_command_interrupts_a_poll(qapp):
+    """A button pressed mid-poll should not wait out the whole sweep.
+
+    The Helios sweep is eight serial queries; before this, a command queued
+    behind one waited for every last read to finish.
+    """
+    w = _YieldingPoller()
+    t = threading.Thread(target=w.run, daemon=True)
+    t.start()
+    w.start_polling()
+    time.sleep(0.03)                      # a poll is now in progress
+
+    pressed = time.monotonic()
+    w._enqueue("click", value="set current")
+    deadline = pressed + 2.0
+    while not w.handled and time.monotonic() < deadline:
+        time.sleep(0.002)
+    waited = time.monotonic() - pressed
+
+    w.stop_polling()
+    w.stop_worker()
+    t.join(timeout=5)
+
+    assert w.handled == ["set current"]
+    # A full sweep is 6 x 20 ms; the command must not have waited for it.
+    assert waited < 0.08, f"command waited {waited * 1000:.0f} ms for the poll"
+
+
 def test_stop_polling_halts_the_cycle(qapp):
     w = _Poller()
     t = threading.Thread(target=w.run, daemon=True)

tools/helios_lds_probe.py

@@ -0,0 +1,232 @@
+#!/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())