Merge fix-scope-channel-coupling: the scope could not be connected to

set_channel_coupling() looped over CHANNEL_COUPLING.items(), so it called
.upper() on the list half of each pair and raised "'list' object has no
attribute 'upper'" before looking at its argument.  configure_channels()
sets the coupling on every channel as part of connecting, so no IP would
work.  Fixed to .values(), with a test that drives the real instrument
class rather than the FakeScope that stubbed the setters out.

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
-other reads are three-letter mnemonics like `LDO`, `LDS`, `LTA`). If one
-exists, add `get_power_mw()` to `hardware/helios_laser.py` using
-`_query_int`, then re-enable the button. If it doesn't, delete the Power
-Monitoring group from the test app and fix the README.
+**Answered (2026-09-08):** the mnemonic is `HMP` (Table 6-3, "Laser Power
+Monitor", read-only, 0-5000 mW) — but the table adds "not available on all
+models", and the operator confirms this rig's laser has no power meter.  So
+the button stays disabled; what remains is to delete the Power Monitoring
+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:
-        the device usually answers in a few ms, so the old unconditional
-        0.05 + 0.2 s cost ~250 ms per query and made an 8-query status poll
-        take ~2 s — longer than the 1 s interval that scheduled it.
+        Reads until this command's reply arrives rather than sleeping a fixed
+        interval: the device usually answers in a few ms, so the old
+        unconditional 0.05 + 0.2 s cost ~250 ms per query and made an 8-query
+        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
-            # mistaken for this command's response.
-            self.serial.reset_input_buffer()
+            # Anything volunteered while the port was idle answers no command.
+            self._discard_input()
 
             if not self._send_command(command):
                 return None
 
-            response = self.serial.read_until(b'\r').decode('ascii', errors='replace').strip()
-            if not response:
-                logger.warning(f"Query '{command}' timed out after {self.timeout}s")
-                return None
-            logger.debug(f"Query '{command}' response: {response}")
+            deadline = time.monotonic() + self.timeout
+            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
 
-            # 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
+            logger.warning(f"Query '{command}' timed out after {self.timeout}s")
+            self._read_pending_lines()
+            return None
 
         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._send_command(command)
+        return self._write_verified("LDF", period_ns)
 
     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._send_command(command)
+        return self._write_verified("LDS", current)
 
     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')
-            if not raw:
-                raw = self.serial.read(self.serial.in_waiting)
-            return raw.decode('ascii', errors='replace').strip()
+            first = self._read_line()
+            lines = [first] if first else []
+            lines += self._read_pending_lines()
+            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

hardware/tektronix_base.py

@@ -292,7 +292,7 @@ class TektronixOscilloscopeBase:
 
         # Validate coupling
         valid = False
-        for variants in self.CHANNEL_COUPLING.items():
+        for variants in self.CHANNEL_COUPLING.values():
             if coupling_upper in [v.upper() for v in variants]:
                 valid = True
                 break

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
-
-# ---------------------------------------------------------------------------
-# 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
-
+from hardware.helios_registers import (
+    CCE_FLAGS, LCE_FLAGS, LER_FLAGS, SEVERITY_LABEL, decode_register,
+)
 
 # 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),
-            ("LCE", lce, _LCE_FLAGS),
-            ("CCE", cce, _CCE_FLAGS),
+            ("LER", ler, LER_FLAGS),
+            ("LCE", lce, LCE_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)
-        self.current_updated.emit(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,9 +1119,20 @@ class HeliosWindow(QWidget):
         self._worker.queue_set_current(self.helios_current_spin.value())
 
     def _on_current_updated(self, current_ma: int):
-        self.helios_current_spin.blockSignals(True)
-        self.helios_current_spin.setValue(current_ma)
-        self.helios_current_spin.blockSignals(False)
+        """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)
 
     def _reset_temperatures(self):
         self.helios_temp_diode_label.setText("--- °C")
@@ -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)

tests/test_tektronix_base.py

@@ -0,0 +1,71 @@
+"""TektronixOscilloscopeBase: the SCPI setters that validate their input.
+
+These run against the real class with only the socket replaced.  The scan
+tests use FakeScope, which stubs the setters out entirely — so a setter
+could raise on every call and nothing in the suite would notice, which is
+what happened: set_channel_coupling() iterated CHANNEL_COUPLING.items()
+instead of .values(), called .upper() on the list half of each pair, and
+raised AttributeError for any coupling at all.  configure_channels() runs
+on connect, so the scope could not be connected to.
+"""
+import pytest
+
+from core.scope_sras import SRAS_CHANNELS, configure_channels
+from hardware.tektronix_base import TektronixOscilloscopeBase
+
+
+class RecordingScope(TektronixOscilloscopeBase):
+    """The real instrument class with the wire replaced by a list."""
+
+    def __init__(self):
+        super().__init__(resource_name="192.0.2.1")
+        self._connected = True
+        self.written: list[str] = []
+
+    def write(self, command):
+        self.written.append(command)
+
+
+@pytest.fixture
+def scope():
+    return RecordingScope()
+
+
+def test_connect_time_channel_setup_reaches_the_wire(scope):
+    """The regression: every command configure_channels() sends must go out."""
+    configure_channels(scope)
+
+    for ch, profile in SRAS_CHANNELS.items():
+        assert f"SELect:CH{ch} ON" in scope.written
+        assert f"CH{ch}:COUPling {profile.coupling}" in scope.written
+        assert f"CH{ch}:TERmination {profile.termination_ohm}" in scope.written
+        assert f"CH{ch}:SCAle {profile.scale_v_div}" in scope.written
+
+
+@pytest.mark.parametrize("coupling", ["DC", "AC", "dc", "ac"])
+def test_channel_coupling_accepts_both_modes_in_any_case(scope, coupling):
+    scope.set_channel_coupling(1, coupling)
+    assert scope.written == [f"CH1:COUPling {coupling}"]
+
+
+def test_channel_coupling_rejects_an_unknown_mode(scope):
+    """Rejection has to be the documented ValueError, not an AttributeError
+    raised while building the list of valid options."""
+    with pytest.raises(ValueError, match="Invalid coupling mode"):
+        scope.set_channel_coupling(1, "GND")
+    assert scope.written == []
+
+
+# The other setters share this shape; a table keeps them honest together.
+@pytest.mark.parametrize("setter,good,bad,sent", [
+    ("set_acquire_mode", "AVERAGE", "SMOOTH", "ACQuire:MODe AVERAGE"),
+    ("set_acquire_mode", "SAM", "SMOOTH", "ACQuire:MODe SAM"),
+    ("set_trigger_slope", "RISE", "SIDEWAYS", "TRIGger:A:EDGE:SLOpe RISE"),
+    ("set_trigger_mode", "NORMAL", "SOMETIMES", "TRIGger:A:MODe NORMAL"),
+])
+def test_validated_setters_take_valid_values_and_reject_the_rest(
+        scope, setter, good, bad, sent):
+    getattr(scope, setter)(good)
+    assert scope.written == [sent]
+    with pytest.raises(ValueError):
+        getattr(scope, setter)(bad)

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())