#!/usr/bin/env python3 """ Scanengine 3 Main Application """ import sys import json from pathlib import Path from PyQt6 import QtWidgets, QtCore from typing import Optional import serial.tools.list_ports from hardware.coherent_hops_laser import CoherentHOPSLaser, DummyLaser from hardware.helios_laser import HeliosLaser, PulseMode from hardware.uc480_camera import UC480Camera, CameraStreamThread from hardware.pybbd202 import ThorlabsServoDriver, AXIS_X, AXIS_Y, TriggerBitsServo from motion_worker import MotionWorker from scanning.stage_scan_plan_generator import StageScanPlanGenerator from genesis_worker import GenesisWorker, GenesisCommand from ui_mainwindow import Ui_MainWindow # Page indices in stackedWidget PAGE_START = 0 PAGE_OPTIONS = 1 PAGE_NEWSCAN = 2 PAGE_CONTINUESCAN = 3 PAGE_SCAN_PROGRESS = 4 CONFIG_PATH = Path(__file__).parent / "config.json" DEFAULT_CONFIG = { "stage": { "serial_port": "", "trigger": "Disabled", "scan_velocity_mm_s": 200.0, "scan_acceleration_mm_s2": 500.0, "optical_axis_x_mm": 0.0, "optical_axis_y_mm": 0.0, }, "fpga": { "serial_port": "", "pulse_divider": 1, "rowpack_enabled": False, }, "t3r": { "serial_port": "", "t_axis_current_ma": 0.0, "gr_axis_current_ma": 0.0, "t_axis_microstepping": "Full Step", "gr_axis_microstepping": "Full Step", }, "oscilloscope": { "ip_address": "", }, "generation_laser": { "serial_port": "", "pulse_frequency_hz": 125000, "diode_pump_current_ma": 0.0, }, "detection_laser": { "power_mw": 0.0, }, "genesis_laser": { "com_port": "/dev/ttyUSB0", }, } # Fixed option lists for combo boxes TRIGGER_OPTIONS = [ "Disabled", "Trigger Out: In Motion", "Trigger Out: Motion Complete", "Trigger Out: Max Velocity", "Trigger Out: High at Max Velocity", ] MICROSTEPPING_OPTIONS = [ "Full Step", "Half Step", "1/4 Step", "1/8 Step", "1/16 Step", "1/32 Step", ] class ScanWorker(QtCore.QObject): """Worker object for handling scanning in a separate thread.""" scan_started = QtCore.pyqtSignal() scan_completed = QtCore.pyqtSignal() scan_failed = QtCore.pyqtSignal(str) angle_started = QtCore.pyqtSignal(int, int) line_started = QtCore.pyqtSignal(int, int, float) current_progress = QtCore.pyqtSignal(int) overall_progress = QtCore.pyqtSignal(int) status_message = QtCore.pyqtSignal(str) def __init__(self, scan_params, motion_worker): super().__init__() self.scan_params = scan_params self.motion_worker = motion_worker self.should_stop = False @QtCore.pyqtSlot() def run_scan(self): """Execute the full scanning process.""" if self.motion_worker: self.motion_worker.scanning_active = True try: self.scan_started.emit() # TODO: implement scan execution logic self.scan_completed.emit() except Exception as e: self.scan_failed.emit(str(e)) finally: if self.motion_worker: self.motion_worker.scanning_active = False def stop(self): self.should_stop = True class MainWindow(QtWidgets.QMainWindow): def __init__(self): super().__init__() self.ui = Ui_MainWindow() self.ui.setupUi(self) self.config = self._load_config() # Hardware objects self.motion_worker: Optional[MotionWorker] = None self.motion_thread: Optional[QtCore.QThread] = None self.genesis_worker: Optional[GenesisWorker] = None self.genesis_thread: Optional[QtCore.QThread] = None self.camera: Optional[UC480Camera] = None self.camera_stream: Optional[CameraStreamThread] = None self.vis_laser: Optional[CoherentHOPSLaser] = None self.ir_laser: Optional[HeliosLaser] = None self.scan_worker: Optional[ScanWorker] = None self.scan_thread: Optional[QtCore.QThread] = None self._connect_signals() self._init_genesis_worker() self.ui.stackedWidget.setCurrentIndex(PAGE_START) # ------------------------------------------------------------------ # Config # ------------------------------------------------------------------ def _load_config(self) -> dict: if CONFIG_PATH.exists(): try: with open(CONFIG_PATH) as f: cfg = json.load(f) for section, values in DEFAULT_CONFIG.items(): cfg.setdefault(section, {}) for key, val in values.items(): cfg[section].setdefault(key, val) return cfg except Exception: pass return {k: dict(v) for k, v in DEFAULT_CONFIG.items()} def _save_config(self): with open(CONFIG_PATH, "w") as f: json.dump(self.config, f, indent=2) # ------------------------------------------------------------------ # Signal wiring # ------------------------------------------------------------------ def _connect_signals(self): # Start page self.ui.start_new_scan_btn.clicked.connect(self._go_to_newscan) self.ui.resume_scan_btn.clicked.connect(self._go_to_continuescan) self.ui.edit_options_btn.clicked.connect(self._go_to_options) # Options page self.ui.options_save_settings_btn.clicked.connect(self._on_options_save) self.ui.options_cancel_btn.clicked.connect(self._go_to_start) self.ui.stage_test_connection_btn.clicked.connect(self._on_test_stage_connection) self.ui.fpga_connect_button.clicked.connect(self._on_fpga_connect) self.ui.fpga_refresh_ports_btn.clicked.connect(self._on_fpga_refresh_ports) self.ui.refresh_serial_ports_btn.clicked.connect(self._on_refresh_serial_ports) self.ui.scope_connect_btn.clicked.connect(self._on_scope_connect) self.ui.generation_connect_button.clicked.connect(self._on_generation_connect) self.ui.detection_test_btn.clicked.connect(self._on_detection_test) self.ui.t3r_connect_btn.clicked.connect(self._on_t3r_connect) self.ui.t3r_refresh_ports_btn.clicked.connect(self._on_t3r_refresh_ports) # New scan page self.ui.newscan_browse_folders_btn.clicked.connect(self._on_newscan_browse) self.ui.newscan_set_current_as_start_btn.clicked.connect(self._on_newscan_set_start) self.ui.newscan_get_delta_from_current_btn.clicked.connect(self._on_newscan_get_delta) self.ui.newscan_toggle_vis_laser_btn.clicked.connect(self._on_newscan_toggle_vis_laser) self.ui.newscan_continue_to_next_btn.clicked.connect(self._on_newscan_start_scan) self.ui.newscan_jog_x_pos_btn.pressed.connect(self._on_jog_x_pos_pressed) self.ui.newscan_jog_x_pos_btn.released.connect(self._on_jog_stop) self.ui.newscan_jog_x_neg_btn.pressed.connect(self._on_jog_x_neg_pressed) self.ui.newscan_jog_x_neg_btn.released.connect(self._on_jog_stop) self.ui.newscan_jog_y_pos_btn.pressed.connect(self._on_jog_y_pos_pressed) self.ui.newscan_jog_y_pos_btn.released.connect(self._on_jog_stop) self.ui.newscan_jog_y_neg_btn.pressed.connect(self._on_jog_y_neg_pressed) self.ui.newscan_jog_y_neg_btn.released.connect(self._on_jog_stop) # Continue scan page self.ui.continuescan_resume_scans.clicked.connect(self._on_resume_scan) # Scan progress page self.ui.abort_scan_button.clicked.connect(self._on_abort_scan) # ------------------------------------------------------------------ # Navigation # ------------------------------------------------------------------ def _go_to_start(self): self.ui.stackedWidget.setCurrentIndex(PAGE_START) def _go_to_options(self): self._populate_options_page() self.ui.stackedWidget.setCurrentIndex(PAGE_OPTIONS) def _go_to_newscan(self): self._populate_newscan_page() self.ui.stackedWidget.setCurrentIndex(PAGE_NEWSCAN) def _go_to_continuescan(self): self._populate_continuescan_page() self.ui.stackedWidget.setCurrentIndex(PAGE_CONTINUESCAN) def _go_to_scan_progress(self): self.ui.stackedWidget.setCurrentIndex(PAGE_SCAN_PROGRESS) # ------------------------------------------------------------------ # Options page # ------------------------------------------------------------------ def _get_serial_ports(self) -> list[str]: return sorted(p.device for p in serial.tools.list_ports.comports()) def _populate_combo(self, combo: QtWidgets.QComboBox, items: list[str], current: str): """Refill a combo box, re-selecting `current` if present.""" combo.blockSignals(True) combo.clear() combo.addItems(items) idx = combo.findText(current) if idx >= 0: combo.setCurrentIndex(idx) elif current: combo.insertItem(0, current) combo.setCurrentIndex(0) combo.blockSignals(False) def _populate_options_page(self): cfg = self.config ports = self._get_serial_ports() # ---- Kinematics tab ---- self.ui.scan_velocity_edit.setText(str(cfg["stage"]["scan_velocity_mm_s"])) self.ui.scan_accel_edit.setText(str(cfg["stage"]["scan_acceleration_mm_s2"])) self.ui.optical_axis_x_edit.setText(str(cfg["stage"]["optical_axis_x_mm"])) self.ui.optical_axis_y_edit.setText(str(cfg["stage"]["optical_axis_y_mm"])) self.ui.stage_serial_edit.setText(cfg["stage"]["serial_port"]) self._populate_combo(self.ui.stage_trigger_combo, TRIGGER_OPTIONS, cfg["stage"]["trigger"]) # ---- Detection / VIS tab ---- self.ui.detection_power_edit.setText(str(cfg["detection_laser"]["power_mw"])) # ---- Generation / IR tab ---- self._populate_combo(self.ui.comboBox, ports, cfg["generation_laser"]["serial_port"]) self.ui.generation_pulse_freq_edit.setText(str(cfg["generation_laser"]["pulse_frequency_hz"])) self.ui.diode_pump_current_edit.setText(str(cfg["generation_laser"]["diode_pump_current_ma"])) # ---- PulseDecimator tab ---- self._populate_combo(self.ui.fpga_serial_port, ports, cfg["fpga"]["serial_port"]) self.ui.fpga_divider_value_edit.setText(str(cfg["fpga"]["pulse_divider"])) self.ui.checkBox.setChecked(cfg["fpga"]["rowpack_enabled"]) # ---- T3R-SL tab ---- self._populate_combo(self.ui.t3r_serial_port_edit, ports, cfg["t3r"]["serial_port"]) self.ui.lineEdit.setText(str(cfg["t3r"]["t_axis_current_ma"])) self.ui.lineEdit_2.setText(str(cfg["t3r"]["gr_axis_current_ma"])) self._populate_combo(self.ui.comboBox_2, MICROSTEPPING_OPTIONS, cfg["t3r"]["t_axis_microstepping"]) self._populate_combo(self.ui.comboBox_3, MICROSTEPPING_OPTIONS, cfg["t3r"]["gr_axis_microstepping"]) # ---- Oscilloscope tab ---- self.ui.scope_ip_address_edit.setText(cfg["oscilloscope"]["ip_address"]) def _on_options_save(self): try: # Kinematics self.config["stage"]["scan_velocity_mm_s"] = float(self.ui.scan_velocity_edit.text()) self.config["stage"]["scan_acceleration_mm_s2"] = float(self.ui.scan_accel_edit.text()) self.config["stage"]["optical_axis_x_mm"] = float(self.ui.optical_axis_x_edit.text()) self.config["stage"]["optical_axis_y_mm"] = float(self.ui.optical_axis_y_edit.text()) self.config["stage"]["serial_port"] = self.ui.stage_serial_edit.text().strip() self.config["stage"]["trigger"] = self.ui.stage_trigger_combo.currentText() # Detection / VIS self.config["detection_laser"]["power_mw"] = float(self.ui.detection_power_edit.text()) # Generation / IR self.config["generation_laser"]["serial_port"] = self.ui.comboBox.currentText() self.config["generation_laser"]["pulse_frequency_hz"] = int(self.ui.generation_pulse_freq_edit.text()) self.config["generation_laser"]["diode_pump_current_ma"] = float(self.ui.diode_pump_current_edit.text()) # PulseDecimator self.config["fpga"]["serial_port"] = self.ui.fpga_serial_port.currentText() self.config["fpga"]["pulse_divider"] = int(self.ui.fpga_divider_value_edit.text()) self.config["fpga"]["rowpack_enabled"] = self.ui.checkBox.isChecked() # T3R-SL self.config["t3r"]["serial_port"] = self.ui.t3r_serial_port_edit.currentText() self.config["t3r"]["t_axis_current_ma"] = float(self.ui.lineEdit.text()) self.config["t3r"]["gr_axis_current_ma"] = float(self.ui.lineEdit_2.text()) self.config["t3r"]["t_axis_microstepping"] = self.ui.comboBox_2.currentText() self.config["t3r"]["gr_axis_microstepping"] = self.ui.comboBox_3.currentText() # Oscilloscope self.config["oscilloscope"]["ip_address"] = self.ui.scope_ip_address_edit.text().strip() except ValueError as e: QtWidgets.QMessageBox.warning(self, "Invalid input", str(e)) return self._save_config() self._go_to_start() def _refresh_serial_ports_for_combos(self, *combos: QtWidgets.QComboBox): """Re-populate serial port combos, preserving current selections.""" ports = self._get_serial_ports() for combo in combos: self._populate_combo(combo, ports, combo.currentText()) def _on_refresh_serial_ports(self): self._refresh_serial_ports_for_combos(self.ui.comboBox) def _on_fpga_refresh_ports(self): self._refresh_serial_ports_for_combos(self.ui.fpga_serial_port) def _on_t3r_refresh_ports(self): self._refresh_serial_ports_for_combos(self.ui.t3r_serial_port_edit) def _on_test_stage_connection(self): pass # TODO def _on_fpga_connect(self): pass # TODO def _on_scope_connect(self): pass # TODO def _on_generation_connect(self): pass # TODO def _on_detection_test(self): pass # TODO def _on_t3r_connect(self): pass # TODO # ------------------------------------------------------------------ # New scan page # ------------------------------------------------------------------ def _populate_newscan_page(self): self.ui.newscan_save_directory_edit.setText(str(Path.home() / "scans")) def _on_newscan_browse(self): directory = QtWidgets.QFileDialog.getExistingDirectory(self, "Select save directory") if directory: self.ui.newscan_save_directory_edit.setText(directory) def _on_newscan_set_start(self): pass # TODO: capture current stage position as scan start def _on_newscan_get_delta(self): pass # TODO: capture current stage position as scan end (compute delta) def _on_newscan_toggle_vis_laser(self): pass # TODO: toggle vis laser on/off def _on_newscan_start_scan(self): scan_params = self._build_scan_params() if scan_params is None: return self._start_scan(scan_params) def _build_scan_params(self) -> Optional[dict]: """Read newscan page widgets and return scan parameter dict, or None on error.""" try: x_start = float(self.ui.newscan_start_x_coord_edit.text()) y_start = float(self.ui.newscan_start_y_coord_edit.text()) x_delta = float(self.ui.newscan_delta_x_coord_edit.text()) y_delta = float(self.ui.newscan_delta_y_coord_edit.text()) except ValueError: QtWidgets.QMessageBox.warning(self, "Invalid input", "Scan coordinates must be numbers.") return None pixel_size_map = { self.ui.newscan_50_micron_radio: 0.05, self.ui.newscan_100_micron_radio: 0.10, self.ui.newscan_250_micron_radio: 0.25, } row_spacing = next( (v for btn, v in pixel_size_map.items() if btn.isChecked()), 0.10 ) return { "x_start_mm": x_start, "y_start_mm": y_start, "x_delta_mm": x_delta, "y_delta_mm": y_delta, "row_spacing_mm": row_spacing, "num_angles": int(self.ui.newscan_num_angles_combo.currentText()), "friendly_name": self.ui.newscan_friendly_name_edit.text(), "file_prefix": self.ui.newcsan_file_prefix_edit.text(), "save_directory": self.ui.newscan_save_directory_edit.text(), "scan_velocity_mm_s": self.config["stage"]["scan_velocity_mm_s"], "scan_acceleration_mm_s2": self.config["stage"]["scan_acceleration_mm_s2"], } # ------------------------------------------------------------------ # Jog controls # ------------------------------------------------------------------ def _on_jog_x_pos_pressed(self): pass # TODO def _on_jog_x_neg_pressed(self): pass # TODO def _on_jog_y_pos_pressed(self): pass # TODO def _on_jog_y_neg_pressed(self): pass # TODO def _on_jog_stop(self): pass # TODO # ------------------------------------------------------------------ # Continue scan page # ------------------------------------------------------------------ def _populate_continuescan_page(self): pass # TODO: populate list of interrupted scans def _on_resume_scan(self): pass # TODO: resume selected scan # ------------------------------------------------------------------ # Scan execution # ------------------------------------------------------------------ def _start_scan(self, scan_params: dict): self.scan_thread = QtCore.QThread() self.scan_worker = ScanWorker(scan_params, self.motion_worker) self.scan_worker.moveToThread(self.scan_thread) self.scan_thread.started.connect(self.scan_worker.run_scan) self.scan_worker.scan_started.connect(self._on_scan_started) self.scan_worker.scan_completed.connect(self._on_scan_completed) self.scan_worker.scan_failed.connect(self._on_scan_failed) self.scan_worker.current_progress.connect(self.ui.scanning_scan_progbar.setValue) self.scan_worker.overall_progress.connect(self.ui.scanning_overall_progbar.setValue) self.scan_worker.status_message.connect(self.ui.scanning_stage_state_label.setText) self._go_to_scan_progress() self.scan_thread.start() @QtCore.pyqtSlot() def _on_scan_started(self): self.ui.abort_scan_button.setEnabled(True) @QtCore.pyqtSlot() def _on_scan_completed(self): self._cleanup_scan_thread() QtWidgets.QMessageBox.information(self, "Scan complete", "Scan finished successfully.") self._go_to_start() @QtCore.pyqtSlot(str) def _on_scan_failed(self, error: str): self._cleanup_scan_thread() QtWidgets.QMessageBox.critical(self, "Scan failed", error) self._go_to_start() def _on_abort_scan(self): if self.scan_worker: self.scan_worker.stop() def _cleanup_scan_thread(self): if self.scan_thread: self.scan_thread.quit() self.scan_thread.wait() self.scan_thread = None self.scan_worker = None # ------------------------------------------------------------------ # Genesis laser worker # ------------------------------------------------------------------ def _init_genesis_worker(self): com_port = self.config.get("genesis_laser", {}).get("com_port", "/dev/ttyUSB0") self.genesis_worker = GenesisWorker(com_port) self.genesis_thread = QtCore.QThread() self.genesis_worker.moveToThread(self.genesis_thread) self.genesis_thread.started.connect(self.genesis_worker.run) self.genesis_thread.start() def _cleanup_genesis_worker(self): if self.genesis_worker: self.genesis_worker.stop() if self.genesis_thread: self.genesis_thread.quit() self.genesis_thread.wait() self.genesis_worker = None self.genesis_thread = None # ------------------------------------------------------------------ # Lifecycle # ------------------------------------------------------------------ def closeEvent(self, event): self._cleanup_genesis_worker() self._cleanup_scan_thread() if self.motion_thread: self.motion_thread.quit() self.motion_thread.wait() super().closeEvent(event) def main(): app = QtWidgets.QApplication(sys.argv) qss_path = Path(__file__).parent / "app_style.qss" if qss_path.exists(): app.setStyleSheet(qss_path.read_text()) window = MainWindow() window.show() sys.exit(app.exec()) if __name__ == "__main__": main()