""" 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 @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) 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. 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 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.serial.reset_input_buffer() 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.05 # 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.serial.reset_input_buffer() 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.