Source code for prsctrl_gui.ui.main_window

"""
Implements QMainWindow
"""
import os.path

from PyQt6.QtCore import Qt, QTimer, QFileSystemWatcher
from PyQt6.QtWidgets import QMainWindow, QScrollArea, QWidget, QHBoxLayout, QLabel, QStatusBar, QFileDialog, \
    QVBoxLayout
from PyQt6.QtWidgets import QTabWidget
from PyQt6.QtGui import QIcon, QAction, QKeySequence, QDragEnterEvent
from PyQt6.QtWidgets import QDialog, QDialogButtonBox, QMessageBox

from devctrl import gui as devgui
from devctrl import devices as d
from devctrl.gui.widgets import ToolBar
from devctrl.gui.widgets import KeyValueInput
from devctrl.utility.event import Event

from .. import devices

# load custom tasks
# from .task import MeasurementTaskWidget
from ..task.measurement import MeasurementTask

from .widgets.help import HelpMenu
from ..resources import get_resource_path
from .widgets.menubar import MenuBar
from .widgets.settings import AppSettings
from .widgets.plot import Plot
from .widgets.device_select import ListChoice
from .widgets.about import About
from .widgets.help import HelpMenu
# from .widgets.treeview import TreeView

import time
import pickle
import logging
log = logging.getLogger(__name__)

import time

import multiprocessing as mp
import threading as mt

from ..utility.config import AppConfig

import devctrl.devices.lock_in as mod_lock_in
import devctrl.devices.led_controller as mod_led
import devctrl.devices.switch as mod_switch
import devctrl.devices.voltmeter as mod_voltmeter
from devctrl.gui.widgets import PythonConsole
from devctrl.gui.widgets import LogSignal, LogViewer, SettingsWidget

from prsctrl.data import PrsData
from prsctrl.measurement.measurement import MeasurementHandler
from prsctrl.measurement import settings as prs_settings

from devctrl.devices import *
from devctrl.utility.config_file import ConfigFile

from ..app_state import device_manager

devmg = device_manager.DeviceManager(prompt_choice_func=devgui.devices.select_device_dialog)
device_manager.init(devmg)
# TODO: move this somewhere else during initialization, provide proper loading screens while devices connect etc.
devmg.add_device(key="shutter",                name="Shutter",                global_varname="shutter",          device_type=Shutter)
# devmg = DeviceManager(config_file=config_file, prompt_choic_func=select_device_interactive)
devmg.add_device(key="power_switch-PRS",       name="Power Switch (PRS)",     global_varname="power_switch_PRS", device_type=Switch, connect_priority=100)
# devmg.add_device(key="power_switch-TAS",       name="Power Switch (TAS)",     global_varname="power_switch_TAS", device_type=Switch, connect_priority=99)
# devmg.add_device(key="shutter",                name="Shutter",                global_varname="shutter",          device_type=Shutter)
# devmg.add_device(key="lamp-xenon",             name="Xenon lamp",             global_varname="xenon",            device_type=Lamp)
# devmg.add_device(key="lamp-halogen",           name="Halogen lamp",           global_varname="halogen",          device_type=Lamp)
# devmg.add_device(key="monochromator",          name="Monochromator",          global_varname="mcm",              device_type=Monochromator)
# devmg.add_device(key="monochromator-detector", name="Monochromator-Detector", global_varname="kymera",           device_type=Monochromator)
devmg.add_device(key="ref_lock-in",            name="Lock-In (REF)",          global_varname="ref_lockin",       device_type=LockInAmp)
# devmg.add_device(key="tra_lock-in",            name="Lock-In (TRA)",          global_varname="tra_lockin",       device_type=LockInAmp)
# devmg.add_device(key="laser",                  name="Laser",                  global_varname="laser",            device_type=Laser)
# devmg.add_device(key="ref_amplifier",          name="Amplifier (REF)",        global_varname="ref_amplifier",    device_type=Amplifier)
# devmg.add_device(key="tra_amplifier",          name="Amplifier (TRA)",        global_varname="tra_amplifier",    device_type=Amplifier)
# devmg.add_device(key="led_controller",         name="LED Controller",         global_varname="led",              device_type=LedController)
# devmg.add_device(key="temperature-sensor",     name="Temperature Sensor",     global_varname="temp",             device_type=TemperatureSensor)
# devmg.add_device(key="wheel-sample",           name="Sample Wheel",           global_varname="wheel",            device_type=Wheel)
# devmg.add_device(key="lamp-mirror",            name="Lamp Mirror (SAM)",      global_varname="sam",              device_type=Switch)

from devctrl.devices.shutter import DummyShutter
DummyShutter.add_device("Test Dummy #2")

[docs] class MainWindow(QMainWindow): """ The main window of the app. Contains most logic as well as measurement functionality with live updates. """ def __init__(self) -> None: """ Initialize the Main-Window. """ super().__init__() # Initialize log capture for the logviewer window log.debug(f"Adding log viewer log handler") self._log_signal = LogSignal() logging.getLogger().addHandler(self._log_signal) self._log_signal.setFormatter(logging.Formatter('%(asctime)s %(levelname)s %(module)s %(funcName)s %(message)s')) # Initialize log viewer. The viewer must be initialized here to capture signals before it is opened self._logviewer = LogViewer(self._log_signal) self._logviewer.setWindowTitle("Log Viewer") self._logviewer.resize(600, 400) # Window-Settings self.setWindowTitle(AppConfig.APP_NAME) x = AppConfig.MAIN_CFG.get_or("geometry_x", 100) y = AppConfig.MAIN_CFG.get_or("geometry_y", 100) width = AppConfig.MAIN_CFG.get_or("geometry_width", 1000) height = AppConfig.MAIN_CFG.get_or("geometry_height", 700) self.setGeometry(x, y, width, height) central_widget = QWidget(self) self.setCentralWidget(central_widget) # Actions self.a_open_help = QAction(text="Help", parent=self) self.a_open_help.setIcon(QIcon.fromTheme(QIcon.ThemeIcon.HelpFaq)) self.a_open_help.setShortcut(QKeySequence("F1")) self.a_open_help.triggered.connect(lambda: self.app_open_help()) self.a_open_about = QAction(text="About", parent=self) self.a_open_about.setIcon(QIcon.fromTheme(get_resource_path("icons/icon.svg"))) self.a_open_about.triggered.connect(lambda: self.app_open_about()) self.a_open_log_viewer = QAction(text="Log Viewer") self.a_open_log_viewer.setShortcut(QKeySequence("F8")) self.a_open_log_viewer.setIcon(QIcon.fromTheme(QIcon.ThemeIcon.MailRead)) self.a_open_log_viewer.triggered.connect(lambda: self.app_open_log_viewer()) self.a_open_console = QAction(text="Console") self.a_open_console.setShortcut(QKeySequence("F7")) self.a_open_console.setIcon(QIcon.fromTheme(u"Terminal")) self.a_open_console.triggered.connect(lambda: self.app_open_console()) self.create_toolbars() self.setMenuBar(MenuBar(self)) # must come after toolbars self.setStatusBar(QStatusBar(self)) layout = QHBoxLayout(central_widget) central_widget.setLayout(layout) # Left: Toolbox self.w_lefttab = QTabWidget(self) self.w_lefttab.setMinimumWidth(300) # self.w_lefttab.setMaximumWidth(400) layout.addWidget(self.w_lefttab) # Devices # Keep a reference to the global devmg to make it available in the console via app.devmg self.devmg = devmg for k in devmg: devmg.connect_dummy(k) self.w_devmg = devgui.devices.device_manager.DeviceManagerWidget(devmg) w = QWidget() w.setLayout(QVBoxLayout()) self.w_devmg_settings = devgui.devices.device_manager.DeviceManagerSettingsWidget(devmg) w.layout().addWidget(self.w_devmg_settings) w.layout().addWidget(self.w_devmg) self.w_devmg_scroll = QScrollArea() self.w_devmg_scroll.setWidgetResizable(True) self.w_devmg_scroll.setWidget(w) self.w_lefttab.addTab(self.w_devmg_scroll, "Devices") metadata_init_dict = AppConfig.MEAS_CFG.get_or("metadata", {}) # Measurement metadata self.w_metadata = KeyValueInput(metadata_init_dict) self.w_metadata.dataChanged.connect(self._metadata_updated) self.w_lefttab.addTab(self.w_metadata, "Measurement Metadata") self.w_measurement_settings_2 = SettingsWidget(prs_settings.PrsSettings()) self.w_lefttab.addTab(self.w_measurement_settings_2, "Measurement Settings 2") from devctrl.gui.task import TaskListWidget, TaskRunWidget, TaskManagerWidget from devctrl.gui.task.run import DebugTaskRunWidget, WaitTaskRunWidget, UserConfirmTaskRunWidget from devctrl.task.tasks import Task, DebugTask, WaitTask, UserConfirmTask from ..task.measurement import MeasurementTask from ..task.runner import MeasurementRunWidget TaskManagerWidget.set_runner(DebugTask, DebugTaskRunWidget) TaskManagerWidget.set_runner(WaitTask, WaitTaskRunWidget) TaskManagerWidget.set_runner(MeasurementTask, MeasurementRunWidget) TaskManagerWidget.set_runner(UserConfirmTask, UserConfirmTaskRunWidget) self.w_taskmanager = TaskManagerWidget() self.w_taskmanager.w_tasklist.load_tasklist(AppConfig.MAIN_CFG.get_path_or("tasklist_path", None)) self.w_lefttab.addTab(self.w_taskmanager, "Tasks") # TODO: the call_f solution isnt pretty # TODO: only accept single file? # TODO: das drag-drop zeug muss später im lefttab implementiert werden # - if a folder, an pickle file or an hdf5 file is being dragged in -> open dataset in new tab # - if a yaml file is dragged in, read contents and determine whether its a task config or sth else and pass on to tasklist class RightTab(QTabWidget): def __init__(self, call_f, parent=None): super().__init__(parent) self.setAcceptDrops(True) self.call_f = call_f def dragEnterEvent(self, event): if not event.mimeData().hasUrls(): event.ignore() if len(event.mimeData().urls()) != 1: event.ignore() event.setDropAction(Qt.DropAction.LinkAction) event.accept() def dropEvent(self, event): for url in event.mimeData().urls(): file_path = url.path()[1:] self.call_f(file_path) # # Right: Tabs: Script, Plot # self.w_right_tab = RightTab(self.measurement_load) # layout.addWidget(self.w_right_tab) max_data_points = AppConfig.MAIN_CFG.get_or("plot_max_data_points", 40000) self.w_plot = Plot(max_data_points=max_data_points) # self.w_right_tab.addTab(self.w_plot, "Plot") self.w_lefttab.addTab(self.w_plot, "Live Plot") self.verbose = True # devices # TODO exe = AppConfig.MAIN_CFG.get_or("power_switch_exe", "") # Measurement self.set_status("Ready") self.menuBar().m_file.addAction(self.a_open_about) self.menuBar().m_file.addAction(self.a_open_help) self.idle_start() def set_status(self, msg): self.statusBar().showMessage(msg)
[docs] def create_toolbars(self) -> None: """ Creates and adds the top and right toolbars to the main window. """ # Top Toolbar [PyQt6.QtWidgets.QToolBar] self.topbar = ToolBar(self, orientation=Qt.Orientation.Horizontal, style=Qt.ToolButtonStyle.ToolButtonTextUnderIcon, icon_size=(24, 24)) # Top Toolbar Buttons self.topbar.add_button("connect_vmdev", "CPD Devices", get_resource_path("icons/volt_device.svg"), self.todo) self.topbar.add_button("connect_leddev", "LED Devices", get_resource_path("icons/led_device.svg"), self.todo) self.topbar.add_button("meas_start", "Start", QIcon.fromTheme(QIcon.ThemeIcon.MediaPlaybackStart), self.measure_start) self.topbar.add_button("meas_stop", "Stop", QIcon.fromTheme(QIcon.ThemeIcon.MediaPlaybackStop), self.measure_stop) self.topbar.add_button("meas_save", "Save", QIcon.fromTheme(QIcon.ThemeIcon.DocumentSaveAs), self.measurement_save) self.topbar.add_button("meas_load", "Load", QIcon.fromTheme(QIcon.ThemeIcon.DocumentOpen), self.measurement_load_dialog) self.topbar.add_separator() self.topbar.add_button("power_on", "Power On", get_resource_path("icons/power_on.svg"), self.power_on) self.topbar.add_button("power_off", "Power Off", get_resource_path("icons/power_off.svg"), self.power_off) self.topbar.add_separator() self.topbar.add_button("app_settings", "Settings", QIcon.fromTheme(QIcon.ThemeIcon.DocumentProperties), self.app_open_settings) self.topbar.addAction(self.a_open_help) self.topbar.addAction(self.a_open_about) self.topbar.add_button("app_exit", "Exit", QIcon.fromTheme(QIcon.ThemeIcon.ApplicationExit), self.app_exit) self.addToolBar(Qt.ToolBarArea.TopToolBarArea, self.topbar) # disable the Stop and Save buttons self.topbar.disable_button("meas_stop") self.topbar.disable_button("meas_save") self.idle_timer = None
def todo(self): log.error(f"TODO") # IDLE - NOT IN MEASUREMENT # check if devices stayed connected def idle_start(self): if self.idle_timer is None: self.idle_timer = QTimer(self) self.idle_timer.timeout.connect(self.idle_update) self.idle_timer.start(AppConfig.MAIN_CFG.get_or("idle_update_interval_s", 10)*1000) else: log.debug("Not starting idle timer - it is already running") self.timer_handle_events = QTimer(self) self.timer_handle_events.timeout.connect(lambda: Event.handle_events()) self.timer_handle_events.start(500) def idle_update(self): pass def idle_stop(self): if self.idle_timer is not None: self.idle_timer.stop() self.idle_timer = None else: log.debug("Not stopping idle timer - it is not running") if self.timer_handle_events is not None: self.timer_handle_events.stop() # MEASUREMENT def measure_start(self): log.info(f"Starting measurement with:\n\tinterval = {interval}\n\tflush_after = {flush_after}\n\tuse_buffer = {use_buffer}\n\tmax_measurements = {max_measurements}\n\tstop_on_script_end = {stop_on_script_end}") self.topbar.disable_button("meas_start") self.topbar.disable_button("connect_vmdev") self.topbar.disable_button("connect_leddev") self.topbar.disable_button("meas_save") self.topbar.disable_button("meas_load") self.topbar.enable_button("meas_stop") if not AppConfig.MAIN_CFG.get_or("metadata_auto_update", True): self.w_metadata.setEnabled(False) self.w_plot.clear_data() raise NotImplementedError(f"Measurement not implemented yet") def measure_stop(self): log.info("Stopping measurement") if not self.measurement_is_running(): raise RuntimeError("measure_stop: Measurement is not running") self.measurement_timer.stop() self.set_status("Stopping measurement") self.topbar.enable_button("meas_start") self.topbar.enable_button("connect_vmdev") self.topbar.enable_button("connect_leddev") self.topbar.enable_button("meas_save") self.topbar.enable_button("meas_load") self.topbar.disable_button("meas_stop") self.w_metadata.setEnabled(True) self.set_status("Ready") self.idle_start() def measure_update(self): self.w_led_script_viewer.update_time(time.time()) if self.proc_measure.is_alive(): while not self.data_queue.empty(): # print(data_queue.qsize(), "\n\n") current_data = self.data_queue.get(block=False) i, tval, vval, led_val = current_data # print(f"Data {i:03}: {tval}s, {vval}V, {led_val}%") self.set_status(f"Data {i:03}: {tval:.3f}s, {vval:.10f}V, {led_val:03}%") # update the plot self.w_plot.update_plot(tval, vval, led_val) else: # measurement might have stopped after max N or script end self.measure_stop() # this should only be run when the measurement is stopped automatically # and is therefore not part of measure_stop() if AppConfig.MEAS_CFG.get_or("power_off_script_end", False): self.power_off() QMessageBox.information(self, "Powered Off", "The measurement has stopped, power switch has been turned off.") def measurement_is_running(self): return self.proc_measure is not None
[docs] def measurement_save(self) -> None: """ Save the last measurement data to a file. """ if self.data_collector is None: raise RuntimeError("Can not save, not data collector initialized") # if not loaded already, this will load the data into memory, which might take a while # this should be done before the dialog self.data_collector.get_data() # open file dialog with previous in directory with previous file extension last_dir = AppConfig.MAIN_CFG.get_or("tmp_last_measurement_save_dir", "") ext = AppConfig.MAIN_CFG.get_or("tmp_last_measurement_save_ext", "csv") if ext not in ["csv", "pkl"]: ext = "csv" # data_collector.dirname gets the name, not path init_path = os.path.join(last_dir, self.data_collector.dirname + "." + ext) file_path, _ = QFileDialog.getSaveFileName(self, "Save File", init_path, "CSV Files (*.csv);;Pickle Files (*.pkl)") if file_path: AppConfig.MAIN_CFG.set("tmp_last_measurement_load_dir", os.path.dirname(file_path)) if file_path.endswith(".csv"): AppConfig.MAIN_CFG.set("tmp_last_measurement_save_ext", "csv") csv = self.data_collector.to_csv() with open(file_path, "w") as f: f.write(csv) elif file_path.endswith(".pkl"): AppConfig.MAIN_CFG.set("tmp_last_measurement_save_ext", "pkl") data = self.data_collector.get_data() with open(file_path, "wb") as f: pickle.dump(data, f) else: raise ValueError(f"Unknown file extension in path: '{file_path}'") self.set_status(f"Saved data to {file_path}") else: self.set_status(f"Aborted saving data, no file selected")
[docs] def measurement_load(self, filepath): """ Load measurement data from the filepath. Only updates the plot. Parameters ---------- filepath The path to the file or directory to load the data from. File may be csv or pickle containing (data, metadata) """ log.info(f"Loading measurement data from '{filepath}'") if self.measurement_is_running(): QMessageBox.critical(self, "Measurement running", "Can not load data while measurement is running") return if os.path.isfile(filepath): if filepath.endswith(".pkl"): with open(filepath, "rb") as file: data, mdata = pickle.load(file) elif filepath.endswith(".csv"): data, mdata = PrsData.load_data_from_csv(filepath) else: raise NotImplementedError(f"Unknown file extension in path: '{filepath}'.\nOnly .pkl and .csv can be loaded") elif os.path.isdir(filepath): data, mdata = PrsData.load_data_from_dir(filepath) else: raise FileNotFoundError(f"No such file or directory: '{filepath}'") self.w_plot.set_data(data[:,1], data[:,2], data[:,3])
[docs] def measurement_load_dialog(self): """ Open a file dialog to load measurement data. This only updates the plot """ if self.measurement_is_running(): QMessageBox.critical(self, "Measurement running", "Can not load data while measurement is running") return last_dir = AppConfig.MAIN_CFG.get_or("tmp_last_measurement_load_dir", "") file_path, _ = QFileDialog.getOpenFileName(self, "Load File", last_dir, "CSV Files (*.csv)") if file_path: dir_name = os.path.dirname(file_path) AppConfig.MAIN_CFG.set("tmp_last_measurement_load_dir", dir_name) self.measurement_load(file_path) else: self.set_status(f"Aborted loading data, no file selected") self.measurement_load(file_path)
def _metadata_updated(self): """ Send the new metadata to the measurement thread. """ if self.measurement_is_running() and AppConfig.MAIN_CFG.get_or("metadata_update_live", True): metadata = self.w_metadata.get_dict() self.command_queue.put(("metadata", metadata))
[docs] def power_on(self): """ Send power on command to the power switch. If configured, wait for a while and then try to reconnect to unconnected devices. """ if self.power_switch is None: raise RuntimeError("No power switch configured") log.info("Powering on") self.power_switch.on() # TODO: change to zero timeout = AppConfig.MAIN_CFG.get_or("power_switch_autoconnect_devices_timeout_s", 20) if not timeout > 0: return if self.vmdev and self.leddev: return dialog = QDialog() dialog.setWindowTitle("Autoconnect devices") buttons = QDialogButtonBox(QDialogButtonBox.StandardButton.Cancel) buttons.accepted.connect(dialog.accept) buttons.rejected.connect(dialog.reject) dialog.setLayout(QVBoxLayout()) dialog.l_time = QLabel(f"{timeout} s") dialog.time_left = timeout dialog.layout().addWidget(QLabel(f"Trying to reconnect to devices in")) dialog.layout().addWidget(dialog.l_time) dialog.layout().addWidget(buttons) dialog.time_left = timeout def update_time_left(dialog): dialog.time_left -= 1 dialog.l_time.setText(f"{dialog.time_left} s") if dialog.time_left <= 0: dialog.accept() dialog.update_time_left = update_time_left dialog.timer = QTimer() dialog.timer.timeout.connect(lambda: dialog.update_time_left(dialog)) dialog.timer.start(1000) ret = dialog.exec() dialog.timer.stop() if ret == QDialog.DialogCode.Accepted: log.info("Trying to autoconnect last devices") if self.vmdev is None: self.vmdev_autoconnect() if self.leddev is None: self.leddev_autoconnect() else: pass
def power_off(self): if self.power_switch is None: raise RuntimeError("No power switch configured") log.info("Powering off") self.idle_stop() # dont show device disconnected errors self.power_switch.off() self.idle_start()
[docs] def app_exit(self) -> None: """ Closes the application. """ self.w_taskmanager.w_tasklist.save_tasklist(AppConfig.MAIN_CFG.get_path_or("tasklist_path", None)) self.app_close_console() self.app_close_log_viewer() self.close()
[docs] def app_open_console(self) -> None: """ Open and focus the log viewer window. The console is created if not existing already """ # Only create the console if it doesn't exist yet if not hasattr(self, '_console') or self._console is None: local_variables = { "app": self, } # | locals() banner = "Python console - use only if you know what you are doing!" banner += "\nThe following variables are available: " + ", ".join(local_variables.keys()) + "\n" self._console = PythonConsole(locals=local_variables, use_globals=True, banner=banner, history_file=AppConfig.MAIN_CFG.get_path_or("console_history_path", None)) self._console.setWindowTitle("Python Console") self._console.resize(600, 400) self._console.show() self._console.activateWindow() self._console.raise_()
def app_close_console(self) -> None: if hasattr(self, '_console') and self._console is not None: self._console.close()
[docs] def app_open_log_viewer(self) -> None: """ Open and focus the log viewer window. """ self._logviewer.show() self._logviewer.activateWindow() self._logviewer.raise_()
def app_close_log_viewer(self) -> None: self._logviewer.close() def app_open_about(self) -> None: dialog = QDialog() dialog.setWindowTitle("About prsctrl-gui") buttons = QDialogButtonBox(QDialogButtonBox.StandardButton.Ok) buttons.accepted.connect(dialog.accept) dialog.setLayout(QVBoxLayout()) dialog.layout().addWidget(About()) dialog.layout().addWidget(buttons) dialog.exec() def app_open_help(self) -> None: dialog = QDialog() dialog.setWindowTitle("Help") buttons = QDialogButtonBox(QDialogButtonBox.StandardButton.Ok) buttons.accepted.connect(dialog.accept) dialog.setLayout(QVBoxLayout()) dialog.layout().addWidget(HelpMenu()) dialog.layout().addWidget(buttons) # set larger window size dialog.resize(800, 600) dialog.exec() def app_open_settings(self) -> None: dialog = QDialog() dialog.setWindowTitle("Settings") layout = QVBoxLayout() w_app_settings = AppSettings() scroll = QScrollArea() scroll.setWidgetResizable(True) scroll.setWidget(w_app_settings) layout.addWidget(scroll) buttons = QDialogButtonBox(QDialogButtonBox.StandardButton.Ok) buttons.accepted.connect(dialog.accept) layout.addWidget(buttons) dialog.setLayout(layout) dialog.exec()
[docs] def closeEvent(self, event): if self.measurement_is_running(): self.measure_stop() # save the metadata metadata = self.w_metadata.get_dict() AppConfig.MEAS_CFG.set("metadata", metadata) # store the geometry as text values in config geo = self.geometry() AppConfig.MAIN_CFG.set("geometry_x", geo.x()) AppConfig.MAIN_CFG.set("geometry_y", geo.y()) AppConfig.MAIN_CFG.set("geometry_width", geo.width()) AppConfig.MAIN_CFG.set("geometry_height", geo.height())