From aeac5fe4d66c0e1c70327bcb5cd0ee7006756d95 Mon Sep 17 00:00:00 2001 From: Thomas Ales Date: Sat, 5 Sep 2026 20:29:53 -0500 Subject: [PATCH] commit --- hardware/helios_laser.py | 154 +++++++++++++++++++++++++++-------- tests/test_helios_laser.py | 160 +++++++++++++++++++++++++++++++++++++ 2 files changed, 279 insertions(+), 35 deletions(-) create mode 100644 tests/test_helios_laser.py diff --git a/hardware/helios_laser.py b/hardware/helios_laser.py index 7b85eb7..7dad247 100755 --- a/hardware/helios_laser.py +++ b/hardware/helios_laser.py @@ -92,44 +92,124 @@ 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. + TRAILING_QUIET_S = 0.05 # the line counts as idle after this long + MAX_REPLY_LINES = 8 + + 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() + + 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: + line = self._read_line() + if line: + lines.append(line) + deadline = time.monotonic() + self.TRAILING_QUIET_S + continue + if time.monotonic() >= deadline: + return lines + time.sleep(0.005) + + @staticmethod + def _value_in(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. + """ + 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 + return line + 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. + # Anything volunteered while the port was idle answers no command. self.serial.reset_input_buffer() if not self._send_command(command): return None - response = self.serial.read_until(b'\r').decode('ascii', errors='replace').strip() - 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 + 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) @@ -156,7 +236,7 @@ class HeliosLaser: return False command = f"LDF {period_ns}" - return self._send_command(command) + return self._write_command(command) def set_current_ma(self, current: int) -> bool: """Set pump diode current in mA.""" @@ -165,17 +245,17 @@ class HeliosLaser: return False command = f"LDS {current}" - return self._send_command(command) + return self._write_command(command) def set_pulse_mode(self, mode: PulseMode) -> bool: """Set pulse mode.""" 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 +323,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,7 +338,11 @@ 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 @@ -266,10 +350,10 @@ class HeliosLaser: self.serial.reset_input_buffer() 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 +361,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 diff --git a/tests/test_helios_laser.py b/tests/test_helios_laser.py new file mode 100644 index 0000000..1c036b9 --- /dev/null +++ b/tests/test_helios_laser.py @@ -0,0 +1,160 @@ +"""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, per Section 6 of the operator's +# manual and the LER/LCE/CCE tables in Section 8. +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"], + # 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"], +} + + +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): + self.replies = REPLIES if replies is None else replies + self.timeout = timeout + self.is_open = True + self.written: list[str] = [] + self._buf = bytearray() + + # ── 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() + 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 + 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" + )