Thomas Ales
b473aac6aa
Three things were making the panel feel slow, all of them waiting rather than working. The trailing quiet window is waited out once per query, so it set the pace of the whole sweep: at 50 ms that was 400 ms of a 590 ms poll spent listening to silence. A reply streams at the baud rate — ~1 ms between bytes, no measurable gap between its lines — and the deadline restarts on every line read, so 20 ms outlasts the gap it exists for twenty times over. A tail that still arrives late is caught by _discard_input(), which is what actually protects the next query. Replayed against the rig transcript, a sweep goes from 590 ms to 356 ms. The poll interval was 1 s on top of that, so a value could be 1.6 s stale. At 0.3 s the panel comes round about every 0.65 s. And a poll held the port for its whole sweep, so a button pressed during one waited for all eight queries. _work_pending() on the worker base lets a poll drop what is left as soon as the operator queues something: a click now waits ~135 ms for the register read in progress instead of the full sweep, and the rest is picked up next time round. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
466 lines
18 KiB
Python
Executable File
466 lines
18 KiB
Python
Executable File
"""
|
||
Helios Laser System Driver
|
||
Basic implementation for controlling the Helios pulsed laser.
|
||
"""
|
||
|
||
import time
|
||
import logging
|
||
from typing import Optional, List
|
||
from enum import Enum
|
||
|
||
from hardware.serial_util import open_8n1
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
class PulseMode(Enum):
|
||
"""Helios pulse mode enumeration"""
|
||
SINGLE_PULSE = 0
|
||
CONTINUOUS_GATING = 1
|
||
CONTINUOUS_PULSING = 2
|
||
|
||
|
||
class HeliosLaser:
|
||
"""
|
||
Driver for Helios pulsed laser system.
|
||
|
||
Communication: RS-232, 9600 baud, 8N1
|
||
Commands are ASCII strings terminated with CR
|
||
"""
|
||
|
||
def __init__(self, port: str = None, timeout: float = 1.0):
|
||
"""Initialize Helios laser driver."""
|
||
self.port = port
|
||
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]:
|
||
"""List available serial ports, likeliest devices first."""
|
||
from hardware.serial_util import list_port_devices
|
||
return list_port_devices()
|
||
|
||
def connect(self, port: str = None) -> bool:
|
||
"""Connect to the Helios laser."""
|
||
if port:
|
||
self.port = port
|
||
|
||
if not self.port:
|
||
logger.error("No port specified")
|
||
return False
|
||
|
||
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}")
|
||
return True
|
||
|
||
except Exception as e:
|
||
logger.error(f"Failed to connect to Helios laser: {e}")
|
||
self.is_connected = False
|
||
return False
|
||
|
||
def disconnect(self):
|
||
"""Disconnect from the laser"""
|
||
if self.serial and self.serial.is_open:
|
||
try:
|
||
# Disable laser before disconnecting
|
||
self.set_laser_enable(False)
|
||
self.serial.close()
|
||
logger.info("Disconnected from Helios laser")
|
||
except Exception as e:
|
||
logger.error(f"Error during disconnect: {e}")
|
||
|
||
self.is_connected = False
|
||
self.serial = None
|
||
|
||
def _send_command(self, command: str) -> bool:
|
||
"""Send a command to the laser."""
|
||
if not self.is_connected or not self.serial:
|
||
logger.error("Not connected to laser")
|
||
return False
|
||
|
||
try:
|
||
cmd_bytes = (command + '\r').encode('ascii')
|
||
self.serial.write(cmd_bytes)
|
||
logger.debug(f"Sent command: {command}")
|
||
return True
|
||
|
||
except Exception as e:
|
||
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 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:
|
||
# Anything volunteered while the port was idle answers no command.
|
||
self._discard_input()
|
||
|
||
if not self._send_command(command):
|
||
return None
|
||
|
||
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
|
||
|
||
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)
|
||
if raw is None:
|
||
return None
|
||
try:
|
||
return int(raw)
|
||
except ValueError:
|
||
logger.error(f"Query '{command}' returned non-integer {raw!r}")
|
||
return None
|
||
|
||
def set_frequency_hz(self, frequency: int) -> bool:
|
||
"""Set laser pulse frequency in Hz."""
|
||
if not (16700 <= frequency <= 125000):
|
||
logger.error(f"Frequency {frequency} Hz out of range (16700-125000)")
|
||
return False
|
||
|
||
# Convert frequency to period in nanoseconds
|
||
period_ns = int(1e9 / frequency)
|
||
|
||
# Clamp to valid range (8000-60000 ns)
|
||
if not (8000 <= period_ns <= 60000):
|
||
logger.error(f"Period {period_ns} ns out of range (8000-60000)")
|
||
return False
|
||
|
||
return self._write_verified("LDF", period_ns)
|
||
|
||
def set_current_ma(self, current: int) -> bool:
|
||
"""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
|
||
|
||
return self._write_verified("LDS", current)
|
||
|
||
def set_pulse_mode(self, mode: PulseMode) -> bool:
|
||
"""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._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._write_command(command)
|
||
|
||
if success:
|
||
state = "enabled" if enable else "disabled"
|
||
logger.info(f"Laser {state}")
|
||
|
||
return success
|
||
|
||
def is_laser_enabled(self) -> bool:
|
||
"""True if laser emission is currently enabled."""
|
||
return self._query_int("LDO") == 1
|
||
|
||
def get_frequency_hz(self) -> Optional[int]:
|
||
"""Current laser frequency in Hz, or None on error."""
|
||
period_ns = self._query_int("LDF")
|
||
if not period_ns:
|
||
return None
|
||
return int(1e9 / period_ns)
|
||
|
||
def get_current_ma(self) -> Optional[int]:
|
||
"""Current pump diode current in mA, or None on error."""
|
||
return self._query_int("LDS")
|
||
|
||
def _query_millicelsius(self, command: str) -> Optional[float]:
|
||
"""Query a temperature register (milli-°C) and convert to °C."""
|
||
value = self._query_int(command)
|
||
return None if value is None else value / 1000.0
|
||
|
||
def get_diode_temp_c(self) -> Optional[float]:
|
||
"""Diode temperature in °C (LTA, 5000–50000 milli-°C)."""
|
||
return self._query_millicelsius("LTA")
|
||
|
||
def get_power_stage_temp_c(self) -> Optional[float]:
|
||
"""Power stage temperature in °C (LTT, 5000–65355 milli-°C)."""
|
||
return self._query_millicelsius("LTT")
|
||
|
||
def get_qswitch_temp_c(self) -> Optional[float]:
|
||
"""Q-switch temperature in °C (EOA, 5000–50000 milli-°C)."""
|
||
return self._query_millicelsius("EOA")
|
||
|
||
def get_controller_serial(self) -> Optional[str]:
|
||
"""Controller serial number, or None on error."""
|
||
return self._query("CSR")
|
||
|
||
def get_head_serial(self) -> Optional[str]:
|
||
"""Laser head serial number, or None on error."""
|
||
return self._query("HSR")
|
||
|
||
def get_status_registers(self) -> tuple:
|
||
"""Query the LER, LCE and CCE status registers.
|
||
|
||
Each is a bitmask (sum of flags); non-zero means active faults,
|
||
cleared with reset_faults(). Returns (ler, lce, cce), any of which
|
||
is None if that register could not be read.
|
||
"""
|
||
return (self._query_int("LER"), self._query_int("LCE"),
|
||
self._query_int("CCE"))
|
||
|
||
def reset_faults(self) -> bool:
|
||
"""
|
||
Execute the controller reset sequence to clear status registers.
|
||
|
||
Protocol-specified sequence: CCE 0 -> LCE 0 -> LER 0
|
||
|
||
Returns:
|
||
True if all three commands sent successfully
|
||
"""
|
||
ok = True
|
||
ok = self._write_command("CCE 0") and ok
|
||
time.sleep(0.1)
|
||
ok = self._write_command("LCE 0") and ok
|
||
time.sleep(0.1)
|
||
ok = self._write_command("LER 0") and ok
|
||
if ok:
|
||
logger.info("Fault reset sequence sent")
|
||
return ok
|
||
|
||
def get_remote_enable(self) -> Optional[bool]:
|
||
"""Remote enable state (LRE — utility connector pin 8); None on error."""
|
||
value = self._query_int("LRE")
|
||
return None if value is None else value == 1
|
||
|
||
def send_raw_command(self, command: str) -> Optional[str]:
|
||
"""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._discard_input()
|
||
if not self._send_command(command):
|
||
return None
|
||
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
|
||
|
||
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._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
|
||
# unpredictable time (including interpreter shutdown, when the port may
|
||
# already be torn down). Callers close the driver explicitly.
|