Phase 1a: delete dead parallel app stack (~4,600 lines)
app.py was an abandoned skeleton (15 'pass # TODO' handlers, loads a deleted .qss); everything else was reachable only from it: ui_mainwindow.py (pyuic6 artifact), sc3-new.ui, motion_worker.py, genesis_worker.py, coherent_hops_laser.py (stubs), scanning/ (dead C# port + unused plan generator), config.json, plus helios_diagnostic.py (sends wrong protocol commands) and helios_terminal.py (worse duplicate of helios_test_app's Terminal tab). hardware/__init__.py no longer wildcard-imports every driver, so the stage driver imports without the uEye camera SDK installed. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
Thomas Ales
@@ -1,615 +0,0 @@
|
||||
#!/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 hardware.t3r_driver import T3RDriver
|
||||
from motion_worker import MotionWorker
|
||||
from scanning.stage_scan_plan_generator import StageScanPlanGenerator
|
||||
from genesis_worker import GenesisWorker, GenesisCommand
|
||||
from t3r_control_panel import T3RControlPanel
|
||||
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, t3r_driver=None):
|
||||
super().__init__()
|
||||
self.scan_params = scan_params
|
||||
self.motion_worker = motion_worker
|
||||
self.t3r_driver = t3r_driver
|
||||
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()
|
||||
num_angles = self.scan_params.get("num_angles", 1)
|
||||
angle_step = 360.0 / num_angles if num_angles > 1 else 0.0
|
||||
gr_microsteps = self.scan_params.get("gr_axis_microsteps", 16)
|
||||
|
||||
for angle_idx in range(num_angles):
|
||||
if self.should_stop:
|
||||
break
|
||||
self.angle_started.emit(angle_idx, num_angles)
|
||||
self.status_message.emit(
|
||||
f"Scanning angle {angle_idx + 1}/{num_angles}")
|
||||
# TODO: execute scan lines for this angle via motion_worker
|
||||
|
||||
if angle_idx < num_angles - 1 and angle_step and self.t3r_driver:
|
||||
if self.t3r_driver.is_open:
|
||||
self.status_message.emit(
|
||||
f"Rotating stage {angle_step:.3f}° for next angle…")
|
||||
self.t3r_driver.rotate_stage(
|
||||
angle_step,
|
||||
gr_microsteps,
|
||||
self.scan_params.get("rotation_velocity", 8000),
|
||||
self.scan_params.get("rotation_accel", 4000),
|
||||
)
|
||||
# TODO: wait for MOTION_DONE event before proceeding
|
||||
|
||||
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
|
||||
|
||||
# T3R focusing / rotation driver (lives in main thread; reader runs internally)
|
||||
self.t3r_driver = T3RDriver(self)
|
||||
self.t3r_panel: Optional[T3RControlPanel] = None
|
||||
|
||||
self._connect_signals()
|
||||
self._init_genesis_worker()
|
||||
self._init_t3r_menu()
|
||||
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):
|
||||
port = self.ui.t3r_serial_port_edit.currentText().split(" ")[0]
|
||||
self._show_t3r_panel()
|
||||
if port and not self.t3r_driver.is_open:
|
||||
try:
|
||||
self.t3r_driver.connect(port)
|
||||
except Exception as exc:
|
||||
QtWidgets.QMessageBox.warning(self, "T3R Connect", str(exc))
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# 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"],
|
||||
# T3R rotation between angles (GR-axis, ch1)
|
||||
"gr_axis_microsteps": 16,
|
||||
"rotation_velocity": 8000,
|
||||
"rotation_accel": 4000,
|
||||
}
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# 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.t3r_driver)
|
||||
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
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# T3R focusing / rotation panel
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
def _init_t3r_menu(self):
|
||||
"""Add a Hardware menu with a T3R panel toggle action."""
|
||||
hw_menu = self.menuBar().addMenu("Hardware")
|
||||
self._t3r_action = hw_menu.addAction("T3R Focusing && Rotation…")
|
||||
self._t3r_action.setCheckable(True)
|
||||
self._t3r_action.setShortcut("Ctrl+T")
|
||||
self._t3r_action.triggered.connect(self._on_t3r_action_toggled)
|
||||
|
||||
def _show_t3r_panel(self):
|
||||
if self.t3r_panel is None:
|
||||
self.t3r_panel = T3RControlPanel(self.t3r_driver, self)
|
||||
self.t3r_panel.finished.connect(
|
||||
lambda: self._t3r_action.setChecked(False))
|
||||
self.t3r_panel.show()
|
||||
self.t3r_panel.raise_()
|
||||
self._t3r_action.setChecked(True)
|
||||
|
||||
def _on_t3r_action_toggled(self, checked: bool):
|
||||
if checked:
|
||||
self._show_t3r_panel()
|
||||
elif self.t3r_panel is not None:
|
||||
self.t3r_panel.hide()
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# 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):
|
||||
if self.t3r_driver.is_open:
|
||||
self.t3r_driver.disconnect()
|
||||
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()
|
||||
-31
@@ -1,31 +0,0 @@
|
||||
{
|
||||
"genesis_laser": {
|
||||
"com_port": "/dev/ttyUSB0"
|
||||
},
|
||||
"detection_laser": {
|
||||
"scan_power_mw": "125"
|
||||
},
|
||||
"generation_laser": {
|
||||
"com_port": "/dev/ttyACM0",
|
||||
"frequency_hz": "20000",
|
||||
"pump_diode_current_ma": "750",
|
||||
"focusing_frequency_hz": "20000",
|
||||
"focusing_pump_current_ma": "300"
|
||||
},
|
||||
"scanning_stage": {
|
||||
"scan_velocity_mm_s": "200",
|
||||
"scan_acceleration_mm_s2": "1500",
|
||||
"x_trigger_mode": 6,
|
||||
"y_trigger_mode": 0,
|
||||
"optical_axis_x_mm": "55",
|
||||
"optical_axis_y_mm": "37.5"
|
||||
},
|
||||
"t3r": {
|
||||
"com_port": "/dev/ttyUSB0"
|
||||
},
|
||||
"oscilloscope": {
|
||||
"socket_address": "192.168.0.1",
|
||||
"scratch_directory": "/opt/",
|
||||
"save_location": "pc"
|
||||
}
|
||||
}
|
||||
@@ -1,193 +0,0 @@
|
||||
"""
|
||||
Genesis Laser Worker Thread
|
||||
|
||||
Manages Genesis laser connection in a separate thread to keep the UI responsive.
|
||||
Provides async querying and status monitoring via Qt signals.
|
||||
"""
|
||||
|
||||
from PyQt6 import QtCore
|
||||
from hardware.genesis_core import SerialComm, I2CProtocol, I2CDevices, LaserControl
|
||||
import queue
|
||||
import time
|
||||
from typing import Optional
|
||||
|
||||
|
||||
class GenesisCommand:
|
||||
"""Represents a genesis laser command"""
|
||||
def __init__(self, cmd_type: str, **kwargs):
|
||||
self.cmd_type = cmd_type
|
||||
self.params = kwargs
|
||||
|
||||
|
||||
class GenesisWorker(QtCore.QObject):
|
||||
"""
|
||||
Worker object for handling Genesis laser control in a separate thread.
|
||||
|
||||
Signals:
|
||||
connected: Emitted when laser connects successfully
|
||||
disconnected: Emitted when laser disconnects
|
||||
connection_failed: Emitted when connection fails (error_msg: str)
|
||||
laser_info_updated: Emitted with laser status
|
||||
query_completed: Emitted when a query operation completes (result: dict)
|
||||
error_occurred: Emitted when an error occurs (error_msg: str)
|
||||
"""
|
||||
|
||||
# Signals
|
||||
connected = QtCore.pyqtSignal()
|
||||
disconnected = QtCore.pyqtSignal()
|
||||
connection_failed = QtCore.pyqtSignal(str)
|
||||
laser_info_updated = QtCore.pyqtSignal(dict) # Status information
|
||||
query_completed = QtCore.pyqtSignal(dict) # Query result
|
||||
error_occurred = QtCore.pyqtSignal(str) # Error message
|
||||
|
||||
def __init__(self, port: str = "/dev/ttyUSB0", baudrate: int = 9600):
|
||||
super().__init__()
|
||||
self.serial_comm = SerialComm()
|
||||
self.i2c_protocol = I2CProtocol(self.serial_comm)
|
||||
self.i2c_devices = I2CDevices(self.i2c_protocol)
|
||||
self.laser_control = LaserControl(self.i2c_devices)
|
||||
|
||||
self.port = port
|
||||
self.baudrate = baudrate
|
||||
self.is_connected = False
|
||||
self.command_queue = queue.Queue()
|
||||
self.running = True
|
||||
|
||||
# Last known laser state
|
||||
self.last_laser_state = {}
|
||||
|
||||
# Update interval for status polling
|
||||
self.last_status_update_time = 0
|
||||
self.status_update_interval = 1.0 # seconds
|
||||
|
||||
@QtCore.pyqtSlot()
|
||||
def run(self):
|
||||
"""Main worker loop - processes commands from queue"""
|
||||
print(f"Genesis laser worker thread started - connecting to {self.port}")
|
||||
|
||||
# Try to connect on startup
|
||||
if self.connect():
|
||||
self.connected.emit()
|
||||
else:
|
||||
error_msg = f"Failed to connect to Genesis laser on {self.port}"
|
||||
print(error_msg)
|
||||
self.connection_failed.emit(error_msg)
|
||||
|
||||
while self.running:
|
||||
try:
|
||||
# Check for commands with timeout to allow periodic status updates
|
||||
try:
|
||||
cmd = self.command_queue.get(timeout=0.05) # 50ms timeout
|
||||
self.process_command(cmd)
|
||||
except queue.Empty:
|
||||
pass
|
||||
|
||||
# Periodically update status if connected
|
||||
if self.is_connected:
|
||||
current_time = time.time()
|
||||
if current_time - self.last_status_update_time >= self.status_update_interval:
|
||||
self.update_laser_status()
|
||||
self.last_status_update_time = current_time
|
||||
|
||||
except Exception as e:
|
||||
print(f"Error in genesis worker loop: {e}")
|
||||
self.error_occurred.emit(str(e))
|
||||
|
||||
# Cleanup on exit
|
||||
self.disconnect()
|
||||
print("Genesis laser worker thread stopped")
|
||||
|
||||
def connect(self) -> bool:
|
||||
"""Establish connection to the laser"""
|
||||
try:
|
||||
if self.serial_comm.connect(self.port, self.baudrate):
|
||||
self.is_connected = True
|
||||
print(f"Connected to Genesis laser on {self.port}")
|
||||
return True
|
||||
else:
|
||||
print(f"Failed to open serial port {self.port}")
|
||||
return False
|
||||
except Exception as e:
|
||||
print(f"Connection error: {e}")
|
||||
return False
|
||||
|
||||
def disconnect(self):
|
||||
"""Disconnect from the laser"""
|
||||
if self.is_connected:
|
||||
self.serial_comm.disconnect()
|
||||
self.is_connected = False
|
||||
self.disconnected.emit()
|
||||
print("Disconnected from Genesis laser")
|
||||
|
||||
def process_command(self, cmd: GenesisCommand):
|
||||
"""Process a command from the queue"""
|
||||
if not self.is_connected:
|
||||
self.error_occurred.emit("Laser not connected")
|
||||
return
|
||||
|
||||
try:
|
||||
if cmd.cmd_type == "query_all":
|
||||
result = self.query_all_status()
|
||||
self.query_completed.emit(result)
|
||||
elif cmd.cmd_type == "query_current":
|
||||
result = {"current": self.laser_control.get_current_actual()}
|
||||
self.query_completed.emit(result)
|
||||
elif cmd.cmd_type == "query_interlock":
|
||||
result = {"interlock": self.laser_control.get_interlock_status()}
|
||||
self.query_completed.emit(result)
|
||||
elif cmd.cmd_type == "set_current":
|
||||
value = cmd.params.get("value", 0)
|
||||
success = self.laser_control.set_current(int(value))
|
||||
self.query_completed.emit({"success": success})
|
||||
elif cmd.cmd_type == "set_shutter":
|
||||
state = cmd.params.get("state", False)
|
||||
success = self.laser_control.set_shutter(state)
|
||||
self.query_completed.emit({"success": success})
|
||||
else:
|
||||
self.error_occurred.emit(f"Unknown command: {cmd.cmd_type}")
|
||||
except Exception as e:
|
||||
self.error_occurred.emit(f"Command execution error: {e}")
|
||||
|
||||
def update_laser_status(self):
|
||||
"""Query and emit current laser status"""
|
||||
if not self.is_connected:
|
||||
return
|
||||
|
||||
try:
|
||||
status = {
|
||||
"connected": True,
|
||||
"current_actual": self.laser_control.get_current_actual(),
|
||||
"interlock_status": self.laser_control.get_interlock_status(),
|
||||
"ldd_enable_status": self.laser_control.get_ldd_enable_status(),
|
||||
"psglue_in_status": self.laser_control.get_psglue_in_status(),
|
||||
"psglue_out_status": self.laser_control.get_psglue_out_status(),
|
||||
"head_dio_status": self.laser_control.get_head_dio_status(),
|
||||
}
|
||||
|
||||
# Only emit if something changed
|
||||
if status != self.last_laser_state:
|
||||
self.last_laser_state = status
|
||||
self.laser_info_updated.emit(status)
|
||||
|
||||
except Exception as e:
|
||||
print(f"Error updating laser status: {e}")
|
||||
|
||||
def query_all_status(self) -> dict:
|
||||
"""Query all laser status information"""
|
||||
return {
|
||||
"connected": True,
|
||||
"current_actual": self.laser_control.get_current_actual(),
|
||||
"interlock_status": self.laser_control.get_interlock_status(),
|
||||
"ldd_enable_status": self.laser_control.get_ldd_enable_status(),
|
||||
"psglue_in_status": self.laser_control.get_psglue_in_status(),
|
||||
"psglue_out_status": self.laser_control.get_psglue_out_status(),
|
||||
"head_dio_status": self.laser_control.get_head_dio_status(),
|
||||
}
|
||||
|
||||
def queue_command(self, cmd: GenesisCommand):
|
||||
"""Queue a command for execution"""
|
||||
self.command_queue.put(cmd)
|
||||
|
||||
def stop(self):
|
||||
"""Stop the worker thread"""
|
||||
self.running = False
|
||||
@@ -1,7 +1,6 @@
|
||||
"""Hardware driver modules for ScanEngine-3"""
|
||||
from .pybbd202 import ThorlabsServoDriver, TriggerBitsServo, AXIS_X, AXIS_Y, CONTROLLER
|
||||
from .t3r_driver import T3RDriver
|
||||
from .uc480_camera import *
|
||||
from .tektronix_base import *
|
||||
from .coherent_hops_laser import *
|
||||
from .genesis_core import *
|
||||
"""Hardware driver modules for ScanEngine-3.
|
||||
|
||||
Import drivers by module (e.g. ``from hardware.t3r_driver import T3RDriver``);
|
||||
nothing is re-exported here so that importing one driver never drags in
|
||||
another driver's SDK (the uEye camera stack in particular).
|
||||
"""
|
||||
|
||||
@@ -1,45 +0,0 @@
|
||||
"""
|
||||
Coherent HOPS Laser Driver - Stub Module
|
||||
This is a temporary stub to allow testing camera integration.
|
||||
"""
|
||||
|
||||
|
||||
class CoherentHOPSLaser:
|
||||
"""Stub class for Coherent HOPS Laser"""
|
||||
pass
|
||||
|
||||
|
||||
class DummyLaser:
|
||||
"""Dummy laser for testing without hardware"""
|
||||
|
||||
def connect(self):
|
||||
"""Simulate connection"""
|
||||
pass
|
||||
|
||||
def disconnect(self):
|
||||
"""Simulate disconnection"""
|
||||
pass
|
||||
|
||||
def get_hardware_id(self):
|
||||
"""Return simulated hardware ID"""
|
||||
return "SIM-12345"
|
||||
|
||||
def get_laser_model(self):
|
||||
"""Return simulated model"""
|
||||
return "Genesis Simulator"
|
||||
|
||||
def get_interlock_status(self):
|
||||
"""Return simulated interlock status"""
|
||||
return "OK"
|
||||
|
||||
def get_key_switch_status(self):
|
||||
"""Return simulated key switch status"""
|
||||
return "ON"
|
||||
|
||||
def get_temperature_main(self):
|
||||
"""Return simulated main temperature"""
|
||||
return 25.5
|
||||
|
||||
def get_temperature_eta(self):
|
||||
"""Return simulated ETA temperature"""
|
||||
return 26.3
|
||||
@@ -1,217 +0,0 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Helios Laser Serial Communication Diagnostic Tool
|
||||
Helps troubleshoot communication issues with the Helios laser.
|
||||
"""
|
||||
|
||||
import serial
|
||||
import time
|
||||
import sys
|
||||
|
||||
def test_port(port, baudrate=9600):
|
||||
"""Test basic communication on a serial port."""
|
||||
print(f"\n{'='*60}")
|
||||
print(f"Testing {port} at {baudrate} baud")
|
||||
print(f"{'='*60}")
|
||||
|
||||
try:
|
||||
ser = serial.Serial(
|
||||
port=port,
|
||||
baudrate=baudrate,
|
||||
bytesize=serial.EIGHTBITS,
|
||||
parity=serial.PARITY_NONE,
|
||||
stopbits=serial.STOPBITS_ONE,
|
||||
timeout=1.0
|
||||
)
|
||||
print(f"✓ Port opened successfully")
|
||||
time.sleep(0.1)
|
||||
|
||||
# Try to query the controller serial number
|
||||
print("\nSending: 'SN?'")
|
||||
ser.write(b'SN?\r')
|
||||
time.sleep(0.5)
|
||||
|
||||
response = ser.readline().decode('ascii', errors='replace').strip()
|
||||
print(f"Response: '{response}'")
|
||||
|
||||
if response:
|
||||
print(f"✓ Got response: {response}")
|
||||
return True, response
|
||||
else:
|
||||
print(f"✗ No response received")
|
||||
|
||||
# Try head serial number
|
||||
print("\nSending: 'HSN?'")
|
||||
ser.write(b'HSN?\r')
|
||||
time.sleep(0.5)
|
||||
|
||||
response = ser.readline().decode('ascii', errors='replace').strip()
|
||||
print(f"Response: '{response}'")
|
||||
|
||||
if response:
|
||||
print(f"✓ Got response: {response}")
|
||||
ser.close()
|
||||
return True, response
|
||||
else:
|
||||
print(f"✗ No response received")
|
||||
|
||||
# Try laser enable status
|
||||
print("\nSending: 'LE?'")
|
||||
ser.write(b'LE?\r')
|
||||
time.sleep(0.5)
|
||||
|
||||
response = ser.readline().decode('ascii', errors='replace').strip()
|
||||
print(f"Response: '{response}'")
|
||||
|
||||
if response:
|
||||
print(f"✓ Got response: {response}")
|
||||
ser.close()
|
||||
return True, response
|
||||
else:
|
||||
print(f"✗ No response received")
|
||||
|
||||
ser.close()
|
||||
return False, "No response to any query"
|
||||
|
||||
except Exception as e:
|
||||
print(f"✗ Error: {e}")
|
||||
return False, str(e)
|
||||
|
||||
|
||||
def test_raw_communication(port, baudrate=9600):
|
||||
"""Test raw serial communication and display hex."""
|
||||
print(f"\n{'='*60}")
|
||||
print(f"Raw Communication Test: {port} at {baudrate} baud")
|
||||
print(f"{'='*60}")
|
||||
|
||||
try:
|
||||
ser = serial.Serial(
|
||||
port=port,
|
||||
baudrate=baudrate,
|
||||
bytesize=serial.EIGHTBITS,
|
||||
parity=serial.PARITY_NONE,
|
||||
stopbits=serial.STOPBITS_ONE,
|
||||
timeout=2.0
|
||||
)
|
||||
print(f"✓ Port opened successfully")
|
||||
time.sleep(0.2)
|
||||
|
||||
# Send a simple query
|
||||
command = b'SN?\r'
|
||||
print(f"\nSending command (hex): {command.hex()}")
|
||||
print(f"Sending command (ascii): {command}")
|
||||
|
||||
ser.write(command)
|
||||
time.sleep(0.5)
|
||||
|
||||
# Read response byte by byte
|
||||
response = b''
|
||||
while True:
|
||||
byte = ser.read(1)
|
||||
if not byte:
|
||||
break
|
||||
response += byte
|
||||
if byte == b'\n' or byte == b'\r':
|
||||
break
|
||||
|
||||
print(f"\nRaw response (hex): {response.hex()}")
|
||||
print(f"Raw response (ascii): {response}")
|
||||
print(f"Response length: {len(response)} bytes")
|
||||
|
||||
# Check for common issues
|
||||
if not response:
|
||||
print("✗ No response - device may not be responding or wrong baud rate")
|
||||
elif response == b'\r' or response == b'\n':
|
||||
print("⚠ Only got line terminator - device may be echoing but not responding to command")
|
||||
else:
|
||||
print("✓ Got a response!")
|
||||
|
||||
ser.close()
|
||||
return True
|
||||
|
||||
except Exception as e:
|
||||
print(f"✗ Error: {e}")
|
||||
return False
|
||||
|
||||
|
||||
def test_echo(port, baudrate=9600):
|
||||
"""Test if the device echoes commands back."""
|
||||
print(f"\n{'='*60}")
|
||||
print(f"Echo Test: {port} at {baudrate} baud")
|
||||
print(f"{'='*60}")
|
||||
|
||||
try:
|
||||
ser = serial.Serial(
|
||||
port=port,
|
||||
baudrate=baudrate,
|
||||
bytesize=serial.EIGHTBITS,
|
||||
parity=serial.PARITY_NONE,
|
||||
stopbits=serial.STOPBITS_ONE,
|
||||
timeout=1.0
|
||||
)
|
||||
|
||||
# Send a test character
|
||||
test_char = b'T'
|
||||
print(f"Sending test character: {test_char}")
|
||||
ser.write(test_char)
|
||||
time.sleep(0.1)
|
||||
|
||||
echo = ser.read(1)
|
||||
if echo == test_char:
|
||||
print(f"✓ Device echoes input")
|
||||
elif echo:
|
||||
print(f"⚠ Device sent something but not the same: {echo}")
|
||||
else:
|
||||
print(f"✗ No echo")
|
||||
|
||||
ser.close()
|
||||
return True
|
||||
|
||||
except Exception as e:
|
||||
print(f"✗ Error: {e}")
|
||||
return False
|
||||
|
||||
|
||||
def main():
|
||||
"""Run diagnostic tests."""
|
||||
port = "/dev/ttyUSB2"
|
||||
|
||||
if len(sys.argv) > 1:
|
||||
port = sys.argv[1]
|
||||
|
||||
print(f"\n{'#'*60}")
|
||||
print(f"# Helios Laser Serial Diagnostic Tool")
|
||||
print(f"# Testing port: {port}")
|
||||
print(f"{'#'*60}")
|
||||
|
||||
# Test standard baud rate
|
||||
success, response = test_port(port, 9600)
|
||||
|
||||
if not success:
|
||||
print("\n" + "="*60)
|
||||
print("Standard baud rate (9600) failed. Trying alternatives...")
|
||||
print("="*60)
|
||||
|
||||
# Try other common baud rates
|
||||
for baudrate in [115200, 19200, 4800, 2400]:
|
||||
success, response = test_port(port, baudrate)
|
||||
if success:
|
||||
print(f"\n✓ SUCCESS! Device responds at {baudrate} baud")
|
||||
break
|
||||
else:
|
||||
print(f"\n✓ SUCCESS! Device responds at 9600 baud")
|
||||
|
||||
# Run additional diagnostics
|
||||
print("\n")
|
||||
test_raw_communication(port, 9600)
|
||||
|
||||
print("\n")
|
||||
test_echo(port, 9600)
|
||||
|
||||
print(f"\n{'#'*60}")
|
||||
print("# Diagnostic Tests Complete")
|
||||
print(f"{'#'*60}\n")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -1,71 +0,0 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Simple serial terminal for manual Helios laser testing.
|
||||
Allows sending raw commands and viewing responses.
|
||||
"""
|
||||
|
||||
import serial
|
||||
import sys
|
||||
from threading import Thread
|
||||
import time
|
||||
|
||||
def read_from_port(ser):
|
||||
"""Read data from serial port and display it."""
|
||||
while True:
|
||||
try:
|
||||
if ser.in_waiting:
|
||||
data = ser.read(ser.in_waiting)
|
||||
print(f"\n[RX] {data.decode('ascii', errors='replace')}", end='')
|
||||
sys.stdout.flush()
|
||||
except:
|
||||
break
|
||||
time.sleep(0.01)
|
||||
|
||||
def main():
|
||||
"""Run interactive serial terminal."""
|
||||
port = "/dev/ttyUSB2"
|
||||
|
||||
if len(sys.argv) > 1:
|
||||
port = sys.argv[1]
|
||||
|
||||
try:
|
||||
ser = serial.Serial(
|
||||
port=port,
|
||||
baudrate=9600,
|
||||
bytesize=serial.EIGHTBITS,
|
||||
parity=serial.PARITY_NONE,
|
||||
stopbits=serial.STOPBITS_ONE,
|
||||
timeout=0.1
|
||||
)
|
||||
print(f"Connected to {port} at 9600 baud")
|
||||
print("Type commands and press Enter. Type 'quit' to exit.\n")
|
||||
|
||||
# Start reader thread
|
||||
reader_thread = Thread(target=read_from_port, args=(ser,), daemon=True)
|
||||
reader_thread.start()
|
||||
|
||||
while True:
|
||||
try:
|
||||
user_input = input("[TX] ")
|
||||
if user_input.lower() == 'quit':
|
||||
break
|
||||
|
||||
# Send command with carriage return
|
||||
command = user_input + '\r'
|
||||
ser.write(command.encode('ascii'))
|
||||
time.sleep(0.1)
|
||||
|
||||
except KeyboardInterrupt:
|
||||
break
|
||||
except Exception as e:
|
||||
print(f"Error: {e}")
|
||||
|
||||
ser.close()
|
||||
print("\nDisconnected")
|
||||
|
||||
except Exception as e:
|
||||
print(f"Failed to open {port}: {e}")
|
||||
sys.exit(1)
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -1,395 +0,0 @@
|
||||
"""
|
||||
Motion Controller Worker Thread
|
||||
|
||||
Handles all motion control operations in a separate thread to keep the UI responsive.
|
||||
Provides async command queueing and position updates via Qt signals.
|
||||
"""
|
||||
|
||||
from PyQt6 import QtCore
|
||||
from hardware.pybbd202 import ThorlabsServoDriver, AXIS_X, AXIS_Y
|
||||
import queue
|
||||
import time
|
||||
from typing import Optional, Dict, Any
|
||||
|
||||
|
||||
class MotionCommand:
|
||||
"""Represents a motion command"""
|
||||
def __init__(self, cmd_type: str, **kwargs):
|
||||
self.cmd_type = cmd_type
|
||||
self.params = kwargs
|
||||
|
||||
|
||||
class MotionWorker(QtCore.QObject):
|
||||
"""
|
||||
Worker object for handling motion control in a separate thread.
|
||||
|
||||
Signals:
|
||||
connected: Emitted when controller connects successfully
|
||||
disconnected: Emitted when controller disconnects
|
||||
connection_failed: Emitted when connection fails (error_msg: str)
|
||||
position_updated: Emitted when position changes (x: float, y: float)
|
||||
homed_status: Emitted with home status (x_homed: bool, y_homed: bool)
|
||||
move_completed: Emitted when a move completes (axis: str)
|
||||
error_occurred: Emitted when an error occurs (error_msg: str)
|
||||
"""
|
||||
|
||||
# Signals
|
||||
connected = QtCore.pyqtSignal()
|
||||
disconnected = QtCore.pyqtSignal()
|
||||
connection_failed = QtCore.pyqtSignal(str)
|
||||
position_updated = QtCore.pyqtSignal(float, float) # x, y in mm
|
||||
homed_status = QtCore.pyqtSignal(bool, bool) # x_homed, y_homed
|
||||
motion_status = QtCore.pyqtSignal(bool, bool) # x_moving, y_moving
|
||||
move_completed = QtCore.pyqtSignal(str) # axis name
|
||||
error_occurred = QtCore.pyqtSignal(str) # error message
|
||||
|
||||
def __init__(self):
|
||||
super().__init__()
|
||||
self.controller: Optional[ThorlabsServoDriver] = None
|
||||
self.is_connected = False
|
||||
self.command_queue = queue.Queue()
|
||||
self.running = True
|
||||
|
||||
# Default parameters
|
||||
self.jog_speed = 20.0 # mm/s
|
||||
self.acceleration = 50.0 # mm/s^2
|
||||
self.step_size = 1.0 # mm
|
||||
|
||||
# Position tracking
|
||||
self.last_x = None
|
||||
self.last_y = None
|
||||
|
||||
# Status tracking
|
||||
self.last_x_homed = None
|
||||
self.last_y_homed = None
|
||||
self.last_x_moving = None
|
||||
self.last_y_moving = None
|
||||
|
||||
# Position update throttling
|
||||
self.last_position_update_time = 0
|
||||
self.position_update_interval = 0.2 # seconds between position reads
|
||||
|
||||
# Flag to pause polling during scanning (scan worker handles its own position queries)
|
||||
self.scanning_active = False
|
||||
|
||||
@QtCore.pyqtSlot()
|
||||
def run(self):
|
||||
"""Main worker loop - processes commands from queue"""
|
||||
print("Motion worker thread started")
|
||||
|
||||
while self.running:
|
||||
try:
|
||||
# Check for commands with timeout to allow periodic position updates
|
||||
try:
|
||||
cmd = self.command_queue.get(timeout=0.05) # 50ms timeout
|
||||
self.process_command(cmd)
|
||||
except queue.Empty:
|
||||
pass
|
||||
|
||||
# Periodically update position and status if connected
|
||||
# Skip updates during scanning - scan worker handles its own position queries
|
||||
if self.is_connected and self.controller and not self.scanning_active:
|
||||
self.update_position()
|
||||
self.update_home_status()
|
||||
self.update_motion_status()
|
||||
|
||||
except Exception as e:
|
||||
print(f"Error in motion worker loop: {e}")
|
||||
self.error_occurred.emit(str(e))
|
||||
|
||||
# Cleanup on exit
|
||||
if self.controller:
|
||||
try:
|
||||
self.controller.disconnect()
|
||||
except:
|
||||
pass
|
||||
|
||||
print("Motion worker thread stopped")
|
||||
|
||||
def process_command(self, cmd: MotionCommand):
|
||||
"""Process a motion command"""
|
||||
try:
|
||||
if cmd.cmd_type == 'connect':
|
||||
self.do_connect()
|
||||
elif cmd.cmd_type == 'disconnect':
|
||||
self.do_disconnect()
|
||||
elif cmd.cmd_type == 'jog':
|
||||
self.do_jog(cmd.params['axis'], cmd.params['direction'])
|
||||
elif cmd.cmd_type == 'home':
|
||||
self.do_home(cmd.params['axis'])
|
||||
elif cmd.cmd_type == 'set_velocity':
|
||||
self.do_set_velocity(cmd.params['speed'], cmd.params['accel'])
|
||||
elif cmd.cmd_type == 'set_step_size':
|
||||
self.step_size = cmd.params['step_size']
|
||||
elif cmd.cmd_type == 'set_axis_enable':
|
||||
self.do_set_axis_enable(cmd.params['axis'], cmd.params['enabled'])
|
||||
elif cmd.cmd_type == 'stop':
|
||||
self.running = False
|
||||
|
||||
except Exception as e:
|
||||
print(f"Error processing command {cmd.cmd_type}: {e}")
|
||||
self.error_occurred.emit(f"Command '{cmd.cmd_type}' failed: {str(e)}")
|
||||
|
||||
def do_connect(self):
|
||||
"""Connect to the motion controller"""
|
||||
try:
|
||||
self.controller = ThorlabsServoDriver()
|
||||
self.controller.connect()
|
||||
|
||||
# Enable channels
|
||||
self.controller.enable_axis(AXIS_X)
|
||||
self.controller.enable_axis(AXIS_Y)
|
||||
|
||||
# Start polling to populate cached state (positions, homed, moving, errors)
|
||||
self.controller.start_polling(interval=0.2)
|
||||
|
||||
# Wait for first polling cycle to populate status
|
||||
time.sleep(0.3)
|
||||
|
||||
# Set initial velocity parameters
|
||||
for dest in [AXIS_X, AXIS_Y]:
|
||||
self.controller.set_velocity_params(
|
||||
dest,
|
||||
max_velocity=self.jog_speed,
|
||||
acceleration=self.acceleration
|
||||
)
|
||||
|
||||
self.is_connected = True
|
||||
# Force initial updates (they will be emitted because last values are None)
|
||||
self.update_position()
|
||||
self.update_home_status()
|
||||
self.update_motion_status()
|
||||
self.connected.emit()
|
||||
|
||||
print("Motion controller connected successfully")
|
||||
|
||||
except Exception as e:
|
||||
print(f"Failed to connect to motion controller: {e}")
|
||||
self.connection_failed.emit(str(e))
|
||||
|
||||
def do_disconnect(self):
|
||||
"""Disconnect from the motion controller"""
|
||||
if self.controller:
|
||||
try:
|
||||
self.controller.disconnect()
|
||||
print("Motion controller disconnected")
|
||||
except Exception as e:
|
||||
print(f"Error during disconnect: {e}")
|
||||
|
||||
self.controller = None
|
||||
self.is_connected = False
|
||||
self.disconnected.emit()
|
||||
|
||||
def do_jog(self, axis: str, direction: int):
|
||||
"""Execute a jog move"""
|
||||
if not self.is_connected or not self.controller:
|
||||
return
|
||||
|
||||
try:
|
||||
dest = AXIS_X if axis == 'x' else AXIS_Y
|
||||
|
||||
# Calculate relative distance
|
||||
distance = self.step_size * direction
|
||||
|
||||
# Execute the move (blocking, with short timeout for continuous jogging)
|
||||
self.controller.move_axis_relative(dest, distance, timeout=0.5)
|
||||
|
||||
# Update position
|
||||
self.update_position()
|
||||
|
||||
self.move_completed.emit(axis)
|
||||
|
||||
except TimeoutError:
|
||||
# Timeout is expected during continuous jog - don't report as error
|
||||
pass
|
||||
except Exception as e:
|
||||
print(f"Jog error: {e}")
|
||||
self.error_occurred.emit(f"Jog failed: {str(e)}")
|
||||
|
||||
def do_home(self, axis: str):
|
||||
"""Home an axis"""
|
||||
if not self.is_connected or not self.controller:
|
||||
return
|
||||
|
||||
try:
|
||||
dest = AXIS_X if axis == 'x' else AXIS_Y
|
||||
|
||||
print(f"Homing {axis.upper()} axis...")
|
||||
self.controller.home_axis(dest, timeout=60.0)
|
||||
|
||||
# Update position and status after homing
|
||||
self.update_position()
|
||||
self.update_home_status()
|
||||
|
||||
print(f"{axis.upper()} axis homed successfully")
|
||||
|
||||
except TimeoutError:
|
||||
print(f"Home timeout: {axis.upper()} axis")
|
||||
self.error_occurred.emit(f"Homing {axis.upper()} timed out")
|
||||
except Exception as e:
|
||||
print(f"Home error: {e}")
|
||||
self.error_occurred.emit(f"Homing {axis.upper()} failed: {str(e)}")
|
||||
|
||||
def do_set_velocity(self, speed: float, accel: float):
|
||||
"""Set velocity parameters"""
|
||||
if not self.is_connected or not self.controller:
|
||||
self.jog_speed = speed
|
||||
self.acceleration = accel
|
||||
return
|
||||
|
||||
try:
|
||||
self.jog_speed = speed
|
||||
self.acceleration = accel
|
||||
|
||||
for dest in [AXIS_X, AXIS_Y]:
|
||||
self.controller.set_velocity_params(
|
||||
dest,
|
||||
max_velocity=self.jog_speed,
|
||||
acceleration=self.acceleration
|
||||
)
|
||||
|
||||
except Exception as e:
|
||||
print(f"Set velocity error: {e}")
|
||||
|
||||
def do_set_axis_enable(self, axis: str, enabled: bool):
|
||||
"""Enable or disable an axis for manual movement"""
|
||||
if not self.is_connected or not self.controller:
|
||||
return
|
||||
|
||||
try:
|
||||
dest = AXIS_X if axis == 'x' else AXIS_Y
|
||||
if enabled:
|
||||
self.controller.enable_axis(dest)
|
||||
else:
|
||||
self.controller.disable_axis(dest)
|
||||
state_str = "enabled" if enabled else "disabled"
|
||||
print(f"{axis.upper()} axis {state_str}")
|
||||
|
||||
except Exception as e:
|
||||
print(f"Set axis enable error: {e}")
|
||||
self.error_occurred.emit(f"Failed to {'enable' if enabled else 'disable'} {axis.upper()} axis: {str(e)}")
|
||||
|
||||
def update_position(self):
|
||||
"""Update current position and emit signal if changed"""
|
||||
if not self.is_connected or not self.controller:
|
||||
return
|
||||
|
||||
# Throttle position reads to avoid excessive signal emission
|
||||
current_time = time.time()
|
||||
if current_time - self.last_position_update_time < self.position_update_interval:
|
||||
return
|
||||
self.last_position_update_time = current_time
|
||||
|
||||
try:
|
||||
# Read cached positions (populated by polling worker)
|
||||
x_pos = self.controller.positions[0]
|
||||
y_pos = self.controller.positions[1]
|
||||
|
||||
# Always emit on first update, or if position changed significantly (> 0.001mm)
|
||||
if (self.last_x is None or self.last_y is None or
|
||||
abs(x_pos - self.last_x) > 0.001 or abs(y_pos - self.last_y) > 0.001):
|
||||
self.last_x = x_pos
|
||||
self.last_y = y_pos
|
||||
print(f"Position update: X={x_pos:.3f}mm, Y={y_pos:.3f}mm")
|
||||
self.position_updated.emit(x_pos, y_pos)
|
||||
|
||||
except Exception as e:
|
||||
print(f"Error updating position: {e}")
|
||||
import traceback
|
||||
traceback.print_exc()
|
||||
|
||||
def update_home_status(self):
|
||||
"""Update home status and emit signal if changed"""
|
||||
if not self.is_connected or not self.controller:
|
||||
return
|
||||
|
||||
try:
|
||||
x_homed = self.controller.am_homed[0]
|
||||
y_homed = self.controller.am_homed[1]
|
||||
|
||||
# Only emit if status changed
|
||||
if x_homed != self.last_x_homed or y_homed != self.last_y_homed:
|
||||
self.last_x_homed = x_homed
|
||||
self.last_y_homed = y_homed
|
||||
self.homed_status.emit(x_homed, y_homed)
|
||||
|
||||
except Exception as e:
|
||||
print(f"Error updating home status: {e}")
|
||||
import traceback
|
||||
traceback.print_exc()
|
||||
|
||||
def update_motion_status(self):
|
||||
"""Update motion status and emit signal if changed.
|
||||
|
||||
The new driver's polling worker keeps am_moving[], am_error[]
|
||||
up to date automatically via status update messages.
|
||||
"""
|
||||
if not self.is_connected or not self.controller:
|
||||
return
|
||||
|
||||
try:
|
||||
# Check for any error conditions
|
||||
if self.controller.am_error[0]:
|
||||
self.error_occurred.emit("X-axis error detected")
|
||||
if self.controller.am_error[1]:
|
||||
self.error_occurred.emit("Y-axis error detected")
|
||||
|
||||
# Read cached motion status (updated by polling worker)
|
||||
x_moving = self.controller.am_moving[0]
|
||||
y_moving = self.controller.am_moving[1]
|
||||
|
||||
# Only emit if status changed
|
||||
if x_moving != self.last_x_moving or y_moving != self.last_y_moving:
|
||||
self.last_x_moving = x_moving
|
||||
self.last_y_moving = y_moving
|
||||
self.motion_status.emit(x_moving, y_moving)
|
||||
|
||||
except Exception as e:
|
||||
print(f"Error updating motion status: {e}")
|
||||
import traceback
|
||||
traceback.print_exc()
|
||||
|
||||
# Slot methods for queuing commands
|
||||
@QtCore.pyqtSlot()
|
||||
def queue_connect(self):
|
||||
"""Queue a connect command"""
|
||||
self.command_queue.put(MotionCommand('connect'))
|
||||
|
||||
@QtCore.pyqtSlot()
|
||||
def queue_disconnect(self):
|
||||
"""Queue a disconnect command"""
|
||||
self.command_queue.put(MotionCommand('disconnect'))
|
||||
|
||||
@QtCore.pyqtSlot(str, int)
|
||||
def queue_jog(self, axis: str, direction: int):
|
||||
"""Queue a jog command"""
|
||||
self.command_queue.put(MotionCommand('jog', axis=axis, direction=direction))
|
||||
|
||||
@QtCore.pyqtSlot(str)
|
||||
def queue_home(self, axis: str):
|
||||
"""Queue a home command"""
|
||||
self.command_queue.put(MotionCommand('home', axis=axis))
|
||||
|
||||
@QtCore.pyqtSlot(float, float)
|
||||
def queue_set_velocity(self, speed: float, accel: float):
|
||||
"""Queue a set velocity command"""
|
||||
self.command_queue.put(MotionCommand('set_velocity', speed=speed, accel=accel))
|
||||
|
||||
@QtCore.pyqtSlot(float)
|
||||
def queue_set_step_size(self, step_size: float):
|
||||
"""Queue a set step size command"""
|
||||
self.command_queue.put(MotionCommand('set_step_size', step_size=step_size))
|
||||
|
||||
@QtCore.pyqtSlot(str, bool)
|
||||
def queue_set_axis_enable(self, axis: str, enabled: bool):
|
||||
"""Queue a command to enable or disable an axis"""
|
||||
self.command_queue.put(MotionCommand('set_axis_enable', axis=axis, enabled=enabled))
|
||||
|
||||
@QtCore.pyqtSlot()
|
||||
def stop(self):
|
||||
"""Stop the worker thread"""
|
||||
# Set running to False immediately so the main loop can exit
|
||||
# even if it's blocked waiting for a response from the controller
|
||||
self.running = False
|
||||
# Also queue a stop command to ensure the command_queue.get() returns
|
||||
self.command_queue.put(MotionCommand('stop'))
|
||||
-2635
File diff suppressed because it is too large
Load Diff
@@ -1,3 +0,0 @@
|
||||
"""Scan planning and modeling modules"""
|
||||
from .sc3_scan_model import SC3ScanModel
|
||||
from .stage_scan_plan_generator import *
|
||||
@@ -1,638 +0,0 @@
|
||||
"""
|
||||
Scan Model Class.
|
||||
Holds the scan configuration, and computes the
|
||||
required start and end points for various specified angles.
|
||||
Python implementation of SC3ScanModel.cs
|
||||
"""
|
||||
|
||||
import math
|
||||
from decimal import Decimal, InvalidOperation
|
||||
from typing import List, Optional
|
||||
import csv
|
||||
|
||||
|
||||
class SC3ScanModel:
|
||||
"""
|
||||
Scan Model for generating scan paths and rotated scans.
|
||||
Manages scan configuration and computes scan coordinates at various angles.
|
||||
"""
|
||||
|
||||
def __init__(self):
|
||||
# Private variables
|
||||
self._x_origin: Decimal = Decimal('0.0')
|
||||
self._y_origin: Decimal = Decimal('0.0')
|
||||
self._x_delta: Decimal = Decimal('0.0')
|
||||
self._y_delta: Decimal = Decimal('0.0')
|
||||
self._row_spacing: Decimal = Decimal('0.0')
|
||||
self._laser_frequency: Decimal = Decimal('2000.0') # Default: 2000 Hz
|
||||
self._scan_velocity: Decimal = Decimal('100.0') # Default: 100 mm/s
|
||||
self._scan_acceleration: Decimal = Decimal('0.0')
|
||||
self._scan_angles: int = 0
|
||||
self._points_required: int = 0
|
||||
self._rows_required: int = 0
|
||||
self._points_per_line: int = 0
|
||||
|
||||
# Optical axis centerline in stage coordinates
|
||||
# Stage: MLS203-1
|
||||
self._optical_x_origin: Decimal = Decimal('55.0')
|
||||
self._optical_y_origin: Decimal = Decimal('37.5')
|
||||
|
||||
# Data storage
|
||||
self._scan_coordinates: List[List[Decimal]] = []
|
||||
self._scan_velocities: List[List[Decimal]] = []
|
||||
self._scan_accelerations: List[List[Decimal]] = []
|
||||
self._rotated_coordinates: List[List[List[Decimal]]] = []
|
||||
|
||||
# Constants
|
||||
self._deg2rad: float = math.pi / 180.0
|
||||
|
||||
# Properties
|
||||
@property
|
||||
def x_origin(self) -> Decimal:
|
||||
"""X coordinate of scan origin (mm)"""
|
||||
return self._x_origin
|
||||
|
||||
@x_origin.setter
|
||||
def x_origin(self, value: Decimal):
|
||||
try:
|
||||
self._x_origin = Decimal(str(value))
|
||||
except (ValueError, InvalidOperation) as e:
|
||||
raise ValueError(f"Invalid x_origin value: {value}") from e
|
||||
|
||||
@property
|
||||
def y_origin(self) -> Decimal:
|
||||
"""Y coordinate of scan origin (mm)"""
|
||||
return self._y_origin
|
||||
|
||||
@y_origin.setter
|
||||
def y_origin(self, value: Decimal):
|
||||
try:
|
||||
self._y_origin = Decimal(str(value))
|
||||
except (ValueError, InvalidOperation) as e:
|
||||
raise ValueError(f"Invalid y_origin value: {value}") from e
|
||||
|
||||
@property
|
||||
def x_delta(self) -> Decimal:
|
||||
"""Total X distance to scan (mm)"""
|
||||
return self._x_delta
|
||||
|
||||
@x_delta.setter
|
||||
def x_delta(self, value: Decimal):
|
||||
try:
|
||||
val = Decimal(str(value))
|
||||
if val < 0:
|
||||
raise ValueError("x_delta must be non-negative")
|
||||
self._x_delta = val
|
||||
self.calculate_points_per_line()
|
||||
self.calculate_points_required()
|
||||
except (ValueError, InvalidOperation) as e:
|
||||
raise ValueError(f"Invalid x_delta value: {value}") from e
|
||||
|
||||
@property
|
||||
def y_delta(self) -> Decimal:
|
||||
"""Total Y distance to scan (mm)"""
|
||||
return self._y_delta
|
||||
|
||||
@y_delta.setter
|
||||
def y_delta(self, value: Decimal):
|
||||
try:
|
||||
val = Decimal(str(value))
|
||||
if val < 0:
|
||||
raise ValueError("y_delta must be non-negative")
|
||||
self._y_delta = val
|
||||
self.calculate_rows_required()
|
||||
self.calculate_points_required()
|
||||
except (ValueError, InvalidOperation) as e:
|
||||
raise ValueError(f"Invalid y_delta value: {value}") from e
|
||||
|
||||
@property
|
||||
def row_spacing(self) -> Decimal:
|
||||
"""Spacing between scan rows (mm)"""
|
||||
return self._row_spacing
|
||||
|
||||
@row_spacing.setter
|
||||
def row_spacing(self, value: Decimal):
|
||||
try:
|
||||
val = Decimal(str(value))
|
||||
if val < 0:
|
||||
raise ValueError("row_spacing must be non-negative")
|
||||
self._row_spacing = val
|
||||
self.calculate_rows_required()
|
||||
self.calculate_points_required()
|
||||
except (ValueError, InvalidOperation) as e:
|
||||
raise ValueError(f"Invalid row_spacing value: {value}") from e
|
||||
|
||||
@property
|
||||
def laser_frequency(self) -> Decimal:
|
||||
"""Laser pulse frequency (Hz)"""
|
||||
return self._laser_frequency
|
||||
|
||||
@laser_frequency.setter
|
||||
def laser_frequency(self, value: Decimal):
|
||||
try:
|
||||
val = Decimal(str(value))
|
||||
if val <= 0:
|
||||
raise ValueError("laser_frequency must be positive")
|
||||
self._laser_frequency = val
|
||||
self.calculate_points_per_line()
|
||||
self.calculate_points_required()
|
||||
except (ValueError, InvalidOperation) as e:
|
||||
raise ValueError(f"Invalid laser_frequency value: {value}") from e
|
||||
|
||||
@property
|
||||
def scan_velocity(self) -> Decimal:
|
||||
"""Scan velocity (mm/s)"""
|
||||
return self._scan_velocity
|
||||
|
||||
@scan_velocity.setter
|
||||
def scan_velocity(self, value: Decimal):
|
||||
try:
|
||||
val = Decimal(str(value))
|
||||
if val <= 0:
|
||||
raise ValueError("scan_velocity must be positive")
|
||||
self._scan_velocity = val
|
||||
self.calculate_points_per_line()
|
||||
self.calculate_points_required()
|
||||
except (ValueError, InvalidOperation) as e:
|
||||
raise ValueError(f"Invalid scan_velocity value: {value}") from e
|
||||
|
||||
@property
|
||||
def scan_acceleration(self) -> Decimal:
|
||||
"""Scan acceleration (mm/s²)"""
|
||||
return self._scan_acceleration
|
||||
|
||||
@scan_acceleration.setter
|
||||
def scan_acceleration(self, value: Decimal):
|
||||
try:
|
||||
val = Decimal(str(value))
|
||||
if val < 0:
|
||||
raise ValueError("scan_acceleration must be non-negative")
|
||||
self._scan_acceleration = val
|
||||
except (ValueError, InvalidOperation) as e:
|
||||
raise ValueError(f"Invalid scan_acceleration value: {value}") from e
|
||||
|
||||
@property
|
||||
def scan_angles(self) -> int:
|
||||
"""Number of scan angles to compute"""
|
||||
return self._scan_angles
|
||||
|
||||
@scan_angles.setter
|
||||
def scan_angles(self, value: int):
|
||||
if value < 0:
|
||||
raise ValueError("scan_angles must be non-negative")
|
||||
self._scan_angles = value
|
||||
# Only compute rotated scans if we have base scan coordinates
|
||||
if self._scan_coordinates:
|
||||
self.compute_rotated_scans()
|
||||
|
||||
@property
|
||||
def points_required(self) -> int:
|
||||
"""Total number of points required for a single angle scan (computed)"""
|
||||
return self._points_required
|
||||
|
||||
@property
|
||||
def rows_required(self) -> int:
|
||||
"""Number of rows required for the scan (computed)"""
|
||||
return self._rows_required
|
||||
|
||||
@property
|
||||
def points_per_line(self) -> int:
|
||||
"""Number of points per scan line (computed)"""
|
||||
return self._points_per_line
|
||||
|
||||
@property
|
||||
def scan_coordinates(self) -> List[List[Decimal]]:
|
||||
"""List of scan coordinates [x_start, y_start, x_end, y_end] in mm"""
|
||||
return self._scan_coordinates
|
||||
|
||||
@property
|
||||
def scan_velocities(self) -> List[List[Decimal]]:
|
||||
"""List of velocity vectors [vx, vy] in mm/s for each angle"""
|
||||
return self._scan_velocities
|
||||
|
||||
@property
|
||||
def scan_accelerations(self) -> List[List[Decimal]]:
|
||||
"""List of acceleration vectors [ax, ay] in mm/s² for each angle"""
|
||||
return self._scan_accelerations
|
||||
|
||||
@property
|
||||
def rotated_coordinates(self) -> List[List[List[Decimal]]]:
|
||||
"""List of rotated scan coordinates for each angle in mm"""
|
||||
return self._rotated_coordinates
|
||||
|
||||
@property
|
||||
def optical_x_origin(self) -> Decimal:
|
||||
"""X coordinate of optical axis origin in mm (read-only, MLS203-1 stage)"""
|
||||
return self._optical_x_origin
|
||||
|
||||
@property
|
||||
def optical_y_origin(self) -> Decimal:
|
||||
"""Y coordinate of optical axis origin in mm (read-only, MLS203-1 stage)"""
|
||||
return self._optical_y_origin
|
||||
|
||||
# Calculation Methods
|
||||
def calculate_points_per_line(self):
|
||||
"""
|
||||
Calculates the number of data points per scan line based on
|
||||
x_delta, scan_velocity, and laser_frequency.
|
||||
|
||||
Formula: points = (distance / velocity) * frequency
|
||||
"""
|
||||
if self._scan_velocity != 0:
|
||||
self._points_per_line = int(
|
||||
(self._x_delta / self._scan_velocity) * self._laser_frequency
|
||||
)
|
||||
|
||||
def calculate_points_required(self):
|
||||
"""
|
||||
Calculates the total number of data points for a complete single-angle scan.
|
||||
Also triggers computation of the zero-angle scan coordinates.
|
||||
|
||||
Formula: total_points = points_per_line * rows_required
|
||||
"""
|
||||
if self._rows_required != 0:
|
||||
self._points_required = self._points_per_line * self._rows_required
|
||||
self.compute_zero_scan()
|
||||
|
||||
def calculate_rows_required(self):
|
||||
"""
|
||||
Calculates the number of scan rows needed based on y_delta and row_spacing.
|
||||
|
||||
Formula: rows = ceil(y_delta / row_spacing)
|
||||
"""
|
||||
if self._row_spacing == 0:
|
||||
self._rows_required = 0
|
||||
else:
|
||||
self._rows_required = int(math.ceil(self._y_delta / self._row_spacing))
|
||||
|
||||
def compute_zero_scan(self):
|
||||
"""
|
||||
Computes the zero-angle (reference) scan coordinates.
|
||||
Each coordinate is [x_start, y_start, x_end, y_end].
|
||||
"""
|
||||
# Ignore the zero-row case
|
||||
if self._rows_required == 0:
|
||||
return
|
||||
|
||||
y_offset = Decimal('0.0')
|
||||
self._scan_coordinates.clear()
|
||||
|
||||
for row in range(self._rows_required + 1):
|
||||
# Calculate x/y origin/delta for each needed row
|
||||
y_offset = Decimal(row) * self._row_spacing
|
||||
|
||||
coords = [
|
||||
self._x_origin, # x_start
|
||||
self._y_origin + y_offset, # y_start
|
||||
self._x_origin + self._x_delta, # x_end
|
||||
self._y_origin + y_offset # y_end (same as y_start for horizontal scan)
|
||||
]
|
||||
self._scan_coordinates.append(coords)
|
||||
|
||||
def _rotate_point(self, x: Decimal, y: Decimal, cosine: Decimal, sine: Decimal) -> tuple:
|
||||
"""
|
||||
Rotate a point around the optical axis origin.
|
||||
|
||||
Args:
|
||||
x, y: Point coordinates to rotate
|
||||
cosine, sine: Precomputed cos and sin of rotation angle
|
||||
|
||||
Returns:
|
||||
Tuple of (rotated_x, rotated_y)
|
||||
"""
|
||||
# Rotation transformation:
|
||||
# Xr = (X - Xo)*cos(a) + (Y - Yo)*sin(a) + Xo
|
||||
# Yr = -(X - Xo)*sin(a) + (Y - Yo)*cos(a) + Yo
|
||||
x_rot = ((x - self._optical_x_origin) * cosine +
|
||||
(y - self._optical_y_origin) * sine +
|
||||
self._optical_x_origin)
|
||||
y_rot = (-(x - self._optical_x_origin) * sine +
|
||||
(y - self._optical_y_origin) * cosine +
|
||||
self._optical_y_origin)
|
||||
return (x_rot, y_rot)
|
||||
|
||||
def _compute_rotated_aoi_bbox(self, angle_rad: float) -> tuple:
|
||||
"""
|
||||
Compute the bounding box of the rotated area of interest.
|
||||
|
||||
Rotates the four corners of the AoI rectangle and finds the
|
||||
min/max extents to create a bounding box.
|
||||
|
||||
Args:
|
||||
angle_rad: Rotation angle in radians
|
||||
|
||||
Returns:
|
||||
Tuple of (min_x, min_y, max_x, max_y) as Decimals
|
||||
"""
|
||||
cosine = Decimal(str(math.cos(angle_rad)))
|
||||
sine = Decimal(str(math.sin(angle_rad)))
|
||||
|
||||
# Define the four corners of the AoI rectangle
|
||||
corners = [
|
||||
(self._x_origin, self._y_origin),
|
||||
(self._x_origin + self._x_delta, self._y_origin),
|
||||
(self._x_origin + self._x_delta, self._y_origin + self._y_delta),
|
||||
(self._x_origin, self._y_origin + self._y_delta)
|
||||
]
|
||||
|
||||
# Rotate all corners
|
||||
rotated_corners = []
|
||||
for x, y in corners:
|
||||
x_rot, y_rot = self._rotate_point(x, y, cosine, sine)
|
||||
rotated_corners.append((x_rot, y_rot))
|
||||
|
||||
# Find bounding box extents
|
||||
x_coords = [corner[0] for corner in rotated_corners]
|
||||
y_coords = [corner[1] for corner in rotated_corners]
|
||||
|
||||
return (min(x_coords), min(y_coords), max(x_coords), max(y_coords))
|
||||
|
||||
def compute_rotated_scans(self):
|
||||
"""
|
||||
Computes rotated scan coordinates by rotating the area of interest (AoI)
|
||||
and generating horizontal (+x direction) scans through the bounding box
|
||||
of the rotated AoI.
|
||||
|
||||
The rotation covers 0 to 180 degrees with spacing determined by scan_angles.
|
||||
For each angle:
|
||||
1. Rotate the AoI rectangle around the optical axis origin
|
||||
2. Compute the bounding box of the rotated rectangle
|
||||
3. Generate horizontal scan lines through the bounding box
|
||||
"""
|
||||
if self._rows_required == 0:
|
||||
return
|
||||
|
||||
# Compute spacing between scans
|
||||
if self._scan_angles == 0:
|
||||
angle_spacing = 180
|
||||
else:
|
||||
angle_spacing = 180 / self._scan_angles
|
||||
|
||||
self._rotated_coordinates.clear()
|
||||
|
||||
# Iterate through each required angle from 0 to 180 degrees
|
||||
i = 0
|
||||
while i < 180:
|
||||
current_angle_radians = i * (math.pi / 180.0)
|
||||
|
||||
# Get bounding box of rotated AoI
|
||||
min_x, min_y, max_x, max_y = self._compute_rotated_aoi_bbox(current_angle_radians)
|
||||
|
||||
# Calculate the y extent of the bounding box
|
||||
y_extent = max_y - min_y
|
||||
|
||||
# Determine number of rows needed for this bounding box
|
||||
if self._row_spacing == 0:
|
||||
num_rows = 0
|
||||
else:
|
||||
num_rows = int(math.ceil(y_extent / self._row_spacing))
|
||||
|
||||
temp_list = []
|
||||
|
||||
# Generate horizontal scan lines through the bounding box
|
||||
for row in range(num_rows + 1):
|
||||
y_offset = Decimal(row) * self._row_spacing
|
||||
y_pos = min_y + y_offset
|
||||
|
||||
# Create horizontal scan line at this y position
|
||||
scan_line = [
|
||||
min_x, # x_start
|
||||
y_pos, # y_start
|
||||
max_x, # x_end
|
||||
y_pos # y_end (same as y_start for horizontal scan)
|
||||
]
|
||||
temp_list.append(scan_line)
|
||||
|
||||
self._rotated_coordinates.append(temp_list)
|
||||
i += int(round(angle_spacing))
|
||||
|
||||
def compute_kinematics(self, offset: int = 0):
|
||||
"""
|
||||
Computes velocity and acceleration component vectors for each scan angle.
|
||||
|
||||
For each angle, decomposes the scalar velocity and acceleration into
|
||||
X and Y components based on the scan direction angle.
|
||||
|
||||
Args:
|
||||
offset: Angle offset in degrees (default: 0)
|
||||
"""
|
||||
if self._scan_angles == 0:
|
||||
scan_increment = 180
|
||||
else:
|
||||
scan_increment = 180 // self._scan_angles
|
||||
|
||||
self._scan_velocities.clear()
|
||||
self._scan_accelerations.clear()
|
||||
|
||||
for i in range(self._scan_angles):
|
||||
deg_angle = scan_increment * i + offset
|
||||
angle_rad = deg_angle * self._deg2rad
|
||||
|
||||
# Compute velocity components: V = V_mag * [cos(θ), sin(θ)]
|
||||
velocities = [
|
||||
Decimal(str(math.cos(angle_rad))) * self._scan_velocity,
|
||||
Decimal(str(math.sin(angle_rad))) * self._scan_velocity
|
||||
]
|
||||
|
||||
# Compute acceleration components: A = A_mag * [cos(θ), sin(θ)]
|
||||
accels = [
|
||||
Decimal(str(math.cos(angle_rad))) * self._scan_acceleration,
|
||||
Decimal(str(math.sin(angle_rad))) * self._scan_acceleration
|
||||
]
|
||||
|
||||
self._scan_velocities.append(velocities)
|
||||
self._scan_accelerations.append(accels)
|
||||
|
||||
# Export Methods
|
||||
def export_zero_scan_csv(self, filename: Optional[str] = None) -> str:
|
||||
"""
|
||||
Export the zero-angle scan coordinates to a CSV file.
|
||||
|
||||
Args:
|
||||
filename: Output filename. If None, generates from row count.
|
||||
|
||||
Returns:
|
||||
The filename that was written
|
||||
|
||||
Raises:
|
||||
ValueError: If no scan coordinates have been computed
|
||||
IOError: If file cannot be written
|
||||
"""
|
||||
if not self._scan_coordinates:
|
||||
raise ValueError("No scan coordinates available. Configure scan parameters first.")
|
||||
|
||||
if filename is None:
|
||||
filename = f"scantest-{self._rows_required}rows.csv"
|
||||
|
||||
try:
|
||||
with open(filename, 'w', newline='') as f:
|
||||
writer = csv.writer(f)
|
||||
for coords in self._scan_coordinates:
|
||||
writer.writerow([str(c) for c in coords])
|
||||
except IOError as e:
|
||||
raise IOError(f"Failed to write file {filename}: {e}") from e
|
||||
|
||||
return filename
|
||||
|
||||
def export_rotated_scan_csv(self, angle_index: int, filename: Optional[str] = None) -> str:
|
||||
"""
|
||||
Export a specific rotated scan to CSV.
|
||||
|
||||
Args:
|
||||
angle_index: Index of the angle to export (0-based)
|
||||
filename: Output filename. If None, generates from angle and row count.
|
||||
|
||||
Returns:
|
||||
The filename that was written
|
||||
|
||||
Raises:
|
||||
ValueError: If angle_index is invalid or no rotated coordinates exist
|
||||
IOError: If file cannot be written
|
||||
"""
|
||||
if not self._rotated_coordinates:
|
||||
raise ValueError("No rotated coordinates available. Set scan_angles first.")
|
||||
|
||||
if angle_index < 0 or angle_index >= len(self._rotated_coordinates):
|
||||
raise ValueError(
|
||||
f"Invalid angle_index {angle_index}. Must be 0-{len(self._rotated_coordinates)-1}"
|
||||
)
|
||||
|
||||
if filename is None:
|
||||
angle_deg = angle_index * (180 // self._scan_angles if self._scan_angles > 0 else 180)
|
||||
filename = f"scantest-{angle_deg:03d}deg-{self._rows_required}rows.csv"
|
||||
|
||||
try:
|
||||
with open(filename, 'w', newline='') as f:
|
||||
writer = csv.writer(f)
|
||||
for coords in self._rotated_coordinates[angle_index]:
|
||||
writer.writerow([str(c) for c in coords])
|
||||
except IOError as e:
|
||||
raise IOError(f"Failed to write file {filename}: {e}") from e
|
||||
|
||||
return filename
|
||||
|
||||
def export_all_rotated_scans_csv(self, output_dir: str = ".") -> List[str]:
|
||||
"""
|
||||
Export all rotated scans to separate CSV files.
|
||||
|
||||
Args:
|
||||
output_dir: Directory to write files to (default: current directory)
|
||||
|
||||
Returns:
|
||||
List of filenames that were written
|
||||
|
||||
Raises:
|
||||
ValueError: If no rotated coordinates exist
|
||||
IOError: If files cannot be written
|
||||
"""
|
||||
if not self._rotated_coordinates:
|
||||
raise ValueError("No rotated coordinates available. Set scan_angles first.")
|
||||
|
||||
import os
|
||||
filenames = []
|
||||
|
||||
for angle_index in range(len(self._rotated_coordinates)):
|
||||
angle_deg = angle_index * (180 // self._scan_angles if self._scan_angles > 0 else 180)
|
||||
filename = f"scantest-{angle_deg:03d}deg-{self._rows_required}rows.csv"
|
||||
filepath = os.path.join(output_dir, filename)
|
||||
|
||||
try:
|
||||
with open(filepath, 'w', newline='') as f:
|
||||
writer = csv.writer(f)
|
||||
for coords in self._rotated_coordinates[angle_index]:
|
||||
writer.writerow([str(c) for c in coords])
|
||||
filenames.append(filepath)
|
||||
except IOError as e:
|
||||
raise IOError(f"Failed to write file {filepath}: {e}") from e
|
||||
|
||||
return filenames
|
||||
|
||||
def export_kinematics_csv(self, filename: Optional[str] = None) -> str:
|
||||
"""
|
||||
Export velocity and acceleration data to CSV.
|
||||
Format: vx, vy, ax, ay for each angle.
|
||||
|
||||
Args:
|
||||
filename: Output filename. If None, generates from row count.
|
||||
|
||||
Returns:
|
||||
The filename that was written
|
||||
|
||||
Raises:
|
||||
ValueError: If no kinematics data has been computed
|
||||
IOError: If file cannot be written
|
||||
"""
|
||||
if not self._scan_velocities or not self._scan_accelerations:
|
||||
raise ValueError("No kinematics data available. Call compute_kinematics() first.")
|
||||
|
||||
if filename is None:
|
||||
filename = f"kinematics-{self._rows_required}rows.csv"
|
||||
|
||||
try:
|
||||
with open(filename, 'w', newline='') as f:
|
||||
writer = csv.writer(f)
|
||||
# Optional: write header
|
||||
writer.writerow(['vx', 'vy', 'ax', 'ay'])
|
||||
for i in range(len(self._scan_velocities)):
|
||||
row = [
|
||||
str(self._scan_velocities[i][0]),
|
||||
str(self._scan_velocities[i][1]),
|
||||
str(self._scan_accelerations[i][0]),
|
||||
str(self._scan_accelerations[i][1])
|
||||
]
|
||||
writer.writerow(row)
|
||||
except IOError as e:
|
||||
raise IOError(f"Failed to write file {filename}: {e}") from e
|
||||
|
||||
return filename
|
||||
|
||||
# Utility Methods
|
||||
def get_angle_list(self) -> List[int]:
|
||||
"""
|
||||
Get the list of scan angles in degrees.
|
||||
|
||||
Returns:
|
||||
List of angles in degrees for the configured scan_angles
|
||||
"""
|
||||
if self._scan_angles == 0:
|
||||
return []
|
||||
|
||||
scan_increment = 180 // self._scan_angles
|
||||
return [scan_increment * i for i in range(self._scan_angles)]
|
||||
|
||||
def get_scan_info(self) -> dict:
|
||||
"""
|
||||
Get a dictionary with current scan configuration and computed values.
|
||||
|
||||
Returns:
|
||||
Dictionary containing scan parameters and computed values
|
||||
"""
|
||||
return {
|
||||
'x_origin': float(self._x_origin),
|
||||
'y_origin': float(self._y_origin),
|
||||
'x_delta': float(self._x_delta),
|
||||
'y_delta': float(self._y_delta),
|
||||
'row_spacing': float(self._row_spacing),
|
||||
'laser_frequency': float(self._laser_frequency),
|
||||
'scan_velocity': float(self._scan_velocity),
|
||||
'scan_acceleration': float(self._scan_acceleration),
|
||||
'scan_angles': self._scan_angles,
|
||||
'points_per_line': self._points_per_line,
|
||||
'rows_required': self._rows_required,
|
||||
'points_required': self._points_required,
|
||||
'optical_x_origin': float(self._optical_x_origin),
|
||||
'optical_y_origin': float(self._optical_y_origin),
|
||||
'num_scan_coordinates': len(self._scan_coordinates),
|
||||
'num_rotated_angles': len(self._rotated_coordinates),
|
||||
'angle_list': self.get_angle_list(),
|
||||
}
|
||||
|
||||
def __repr__(self) -> str:
|
||||
"""String representation of the scan model."""
|
||||
return (
|
||||
f"SC3ScanModel("
|
||||
f"origin=({self._x_origin},{self._y_origin}), "
|
||||
f"delta=({self._x_delta},{self._y_delta}), "
|
||||
f"rows={self._rows_required}, "
|
||||
f"angles={self._scan_angles})"
|
||||
)
|
||||
@@ -1,117 +0,0 @@
|
||||
"""
|
||||
This module contains a StageScanPlanGenerator class that generates scanning plans
|
||||
for microscope stages. The scans are generated in a single direction based on provided
|
||||
start and end coordinates, as well as spacing between scan lines.
|
||||
"""
|
||||
|
||||
import numpy as np
|
||||
from typing import List, Tuple, Optional
|
||||
|
||||
|
||||
class StageScanPlanGenerator:
|
||||
"""
|
||||
A class to generate scanning plans for microscope stages.
|
||||
|
||||
Attributes:
|
||||
start_coords (Tuple[float, float]): Starting X and Y coordinates.
|
||||
end_coords (Tuple[float, float]): Ending X and Y coordinates.
|
||||
spacing (float): Spacing between scan lines in the perpendicular direction.
|
||||
"""
|
||||
|
||||
def __init__(self, start_x: float, start_y: float,
|
||||
end_x: float, end_y: float, spacing: float):
|
||||
"""
|
||||
Initialize the StageScanPlanGenerator with scan parameters.
|
||||
|
||||
Args:
|
||||
start_x (float): Starting X coordinate.
|
||||
start_y (float): Starting Y coordinate.
|
||||
end_x (float): Ending X coordinate.
|
||||
end_y (float): Ending Y coordinate.
|
||||
spacing (float): Spacing between scan lines in the perpendicular direction.
|
||||
"""
|
||||
self.start_coords = (start_x, start_y)
|
||||
self.end_coords = (end_x, end_y)
|
||||
self.spacing = spacing
|
||||
|
||||
def _calculate_scan_direction(self) -> Tuple[float, float]:
|
||||
"""
|
||||
Calculate the direction vector of the scan.
|
||||
|
||||
Returns:
|
||||
Tuple[float, float]: Normalized direction vector (dx, dy).
|
||||
"""
|
||||
dx = self.end_coords[0] - self.start_coords[0]
|
||||
dy = self.end_coords[1] - self.start_coords[1]
|
||||
length = np.sqrt(dx**2 + dy**2)
|
||||
|
||||
if length == 0:
|
||||
raise ValueError("Start and end coordinates cannot be the same")
|
||||
|
||||
return dx / length, dy / length
|
||||
|
||||
def _calculate_perpendicular_direction(self) -> Tuple[float, float]:
|
||||
"""
|
||||
Calculate a perpendicular direction vector to the scan direction.
|
||||
|
||||
Returns:
|
||||
Tuple[float, float]: Perpendicular vector (px, py).
|
||||
"""
|
||||
dx, dy = self._calculate_scan_direction()
|
||||
# Rotate (dx, dy) by 90 degrees to get perpendicular vector
|
||||
px = -dy
|
||||
py = dx
|
||||
return px, py
|
||||
|
||||
def generate_scan_plan(self) -> List[Tuple[Tuple[float, float], Tuple[float, float]]]:
|
||||
"""
|
||||
Generate a scan plan with waypoints for the microscope stage.
|
||||
|
||||
Returns:
|
||||
List[Tuple[Tuple[float, float], Tuple[float, float]]]:
|
||||
A list of (start_point, end_point) tuples for each scan line.
|
||||
"""
|
||||
dx, dy = self._calculate_scan_direction()
|
||||
px, py = self._calculate_perpendicular_direction()
|
||||
|
||||
# Calculate the total length in the perpendicular direction
|
||||
start_x, start_y = self.start_coords
|
||||
end_x, end_y = self.end_coords
|
||||
min_coord_perp = min(start_x * px + start_y * py, end_x * px + end_y * py)
|
||||
max_coord_perp = max(start_x * px + start_y * py, end_x * px + end_y * py)
|
||||
|
||||
# Generate scan lines
|
||||
waypoints = []
|
||||
current_pos_perp = min_coord_perp
|
||||
|
||||
while current_pos_perp <= max_coord_perp:
|
||||
# Calculate start and end points for this scan line
|
||||
perp_offset = current_pos_perp - (start_x * px + start_y * py)
|
||||
line_start_x = start_x + perp_offset * dx
|
||||
line_start_y = start_y + perp_offset * dy
|
||||
|
||||
line_end_x = line_start_x + dx * abs(self.end_coords[0] - self.start_coords[0])
|
||||
line_end_y = line_start_y + dy * abs(self.end_coords[1] - self.start_coords[1])
|
||||
|
||||
waypoints.append(((line_start_x, line_start_y), (line_end_x, line_end_y)))
|
||||
current_pos_perp += self.spacing
|
||||
|
||||
return waypoints
|
||||
|
||||
|
||||
# Example usage:
|
||||
if __name__ == "__main__":
|
||||
# Create a scan plan generator
|
||||
generator = StageScanPlanGenerator(
|
||||
start_x=0.0, start_y=0.0,
|
||||
end_x=10.0, end_y=10.0,
|
||||
spacing=2.5
|
||||
)
|
||||
|
||||
# Generate the scan plan
|
||||
scan_plan = generator.generate_scan_plan()
|
||||
|
||||
# Print the scan plan
|
||||
print("Scan Plan:")
|
||||
for i, (start, end) in enumerate(scan_plan):
|
||||
print(f"Line {i+1}: Start at {start}, End at {end}")
|
||||
-1321
File diff suppressed because it is too large
Load Diff
Reference in New Issue
Block a user