diff --git a/gui/qt_workers.py b/gui/qt_workers.py index f7719ca..3f6100c 100644 --- a/gui/qt_workers.py +++ b/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() diff --git a/hardware/helios_laser.py b/hardware/helios_laser.py index d54d41c..7a80473 100755 --- a/hardware/helios_laser.py +++ b/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}") @@ -105,23 +107,69 @@ class HeliosLaser: # 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. - TRAILING_QUIET_S = 0.05 # the line counts as idle after this long + # 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 CR-terminated line without its framing; None if nothing came.""" - raw = self.serial.read_until(b'\r') - if not raw: - return None - return raw.decode('ascii', errors='replace').strip() + """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: - if self.serial.in_waiting: + # 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 @@ -130,15 +178,21 @@ class HeliosLaser: return lines time.sleep(0.005) - @staticmethod - def _value_in(line: str, mnemonic: str) -> Optional[str]: + # 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 is taken as the value — that is how the serial - numbers come back. + 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: @@ -152,7 +206,9 @@ class HeliosLaser: fields = line.split() if fields and fields[0].upper() == mnemonic: return fields[1] if len(fields) > 1 else None - return line + 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. @@ -167,7 +223,7 @@ class HeliosLaser: mnemonic = fields[0].upper() if fields else "" try: # Anything volunteered while the port was idle answers no command. - self.serial.reset_input_buffer() + self._discard_input() if not self._send_command(command): return None @@ -221,7 +277,7 @@ class HeliosLaser: # 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.05 # let the controller store it before reading + 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. @@ -387,7 +443,7 @@ class HeliosLaser: 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 first = self._read_line() diff --git a/hardware/helios_registers.py b/hardware/helios_registers.py new file mode 100644 index 0000000..1d1ee9a --- /dev/null +++ b/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 diff --git a/helios_test_app.py b/helios_test_app.py index 0d4d307..5d4fa76 100755 --- a/helios_test_app.py +++ b/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}: ") 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)) diff --git a/sc3_aui_app.py b/sc3_aui_app.py index a510adc..ced8060 100755 --- a/sc3_aui_app.py +++ b/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) @@ -415,23 +418,45 @@ class HeliosWorker(PollingQueueWorker): 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: @@ -1015,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") ) diff --git a/tests/test_helios_laser.py b/tests/test_helios_laser.py index b471636..3eb798f 100644 --- a/tests/test_helios_laser.py +++ b/tests/test_helios_laser.py @@ -12,23 +12,34 @@ import pytest from hardware.helios_laser import HeliosLaser, PulseMode -# What the controller actually sends back, per Section 6 of the operator's -# manual and the LER/LCE/CCE tables in Section 8. +# 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"], - "LDF": ["LDF = 20000 ns"], - "LDS": ["LDS = 1500 mA"], - "LDG": ["LDG = 14"], - "LRE": ["LRE = 0"], - "LTA": ["LTA = 25400 m°C"], - "LTT": ["LTT = 31200 m°C"], - "EOA": ["EOA = 40100 m°C"], - "CSR": ["CSR = 1234567"], - "HSR": ["HSR = 7654321"], + "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"], - "LCE": ["LCE = 2", "Bit 15..0: 0000 0000 0000 0010"], - "CCE": ["CCE = 0", "Bit 15..0: 0000 0000 0000 0000"], + "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, } @@ -182,7 +193,7 @@ def test_set_commands_clear_their_acknowledgement(laser): 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" + "LCE = 2\nBit 15..0: 0000 0000 0000 0010" ) @@ -235,3 +246,82 @@ def test_a_set_frequency_is_read_back(laser): 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() diff --git a/tests/test_qt_workers.py b/tests/test_qt_workers.py index cec6c17..5254bd3 100644 --- a/tests/test_qt_workers.py +++ b/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) diff --git a/tools/helios_lds_probe.py b/tools/helios_lds_probe.py new file mode 100644 index 0000000..d0a9f34 --- /dev/null +++ b/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())