Thomas Ales [M S E]
546 lines
21 KiB
Python
546 lines
21 KiB
Python
#!/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()
|