jgrewe/minipyrelacs
2
0
Fork 0
forked from awendt/pyrelacs
Code Pull Requests Activity

adding QProcess for executing scripts in the repro folder

Browse Source
This commit is contained in:
wendtalexander 2024-09-19 11:24:58 +02:00
parent bb88053aaa
commit dc96359f1d
1 changed files with 63 additions and 69 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

132
pyrelacs/app.py
View File

@ -3,7 +3,7 @@ import sys
import pathlib import pathlib
import ctypes import ctypes
from PyQt6.QtCore import QSize, QThreadPool, Qt from PyQt6.QtCore import QProcess, QSize, QThreadPool, Qt
from PyQt6.QtWidgets import ( from PyQt6.QtWidgets import (
QApplication, QApplication,
QGridLayout, QGridLayout,
@ -11,6 +11,7 @@ from PyQt6.QtWidgets import (
QToolBar, QToolBar,
QWidget, QWidget,
QMainWindow, QMainWindow,
QPlainTextEdit,
) )
import tomli import tomli
import uldaq import uldaq
@ -27,7 +28,11 @@ class PyRelacs(QMainWindow):
def __init__(self): def __init__(self):
super().__init__() super().__init__()
self.setWindowTitle("PyRelacs") self.setWindowTitle("PyRelacs")
self.setMinimumSize(1000, 1000)
self.threadpool = QThreadPool() self.threadpool = QThreadPool()
# for starting a Qprocess
self.p = None
self.daq_connect_button = QPushButton("Connect Daq") self.daq_connect_button = QPushButton("Connect Daq")
self.daq_connect_button.setCheckable(True) self.daq_connect_button.setCheckable(True)
@ -37,17 +42,17 @@ class PyRelacs(QMainWindow):
self.daq_disconnect_button.setCheckable(True) self.daq_disconnect_button.setCheckable(True)
self.daq_disconnect_button.clicked.connect(self.disconnect_dac) self.daq_disconnect_button.clicked.connect(self.disconnect_dac)
self.repro_button = QPushButton("Load Repros") self.text = QPlainTextEdit()
self.repro_button.setCheckable(True) self.text.setReadOnly(True)
self.repro_button.clicked.connect(self.repro)
layout = QGridLayout() layout = QGridLayout()
layout.addWidget(self.daq_connect_button, 0, 0) layout.addWidget(self.daq_connect_button, 0, 0)
layout.addWidget(self.daq_disconnect_button, 0, 1) layout.addWidget(self.daq_disconnect_button, 0, 1)
layout.addWidget(self.repro_button, 1, 0) layout.addWidget(self.text, 2, 0, 1, 2)
self.toolbar = QToolBar("Repros") self.toolbar = QToolBar("Repros")
self.addToolBar(self.toolbar) self.addToolBar(self.toolbar)
self.repro()
self.setFixedSize(QSize(400, 300)) self.setFixedSize(QSize(400, 300))
widget = QWidget() widget = QWidget()
@ -78,70 +83,59 @@ class PyRelacs(QMainWindow):
log.debug("DAQ was not connected") log.debug("DAQ was not connected")
def repro(self): def repro(self):
config_path = pathlib.Path(__file__).parent.resolve() / "repro.toml" repos_path = pathlib.Path(__file__).parent / "repros"
if not config_path.is_file: repos_names = list(repos_path.glob("*.py"))
log.error("repro.toml not found") # exclude the repos.py file
QApplication.quit() repos_names = [
repors = [] f.with_suffix("").name for f in repos_names if not f.name == "repos.py"
]
with open(config_path, mode="rb") as fp: for rep in repos_names:
self.config = tomli.load(fp) individual_repro_button = QAction(rep, self)
for r in self.config: individual_repro_button.setStatusTip("Button")
repors.append(r) individual_repro_button.triggered.connect(
lambda checked, n=rep: self.run_repro(n)
individual_repro_button = QAction(f"{repors[0]}", self) )
individual_repro_button.setStatusTip("Button") self.toolbar.addAction(individual_repro_button)
individual_repro_button.triggered.connect(self.run_repro)
self.toolbar.addAction(individual_repro_button) def message(self, s):
self.text.appendPlainText(s)
def run_repro(self):
daq_thread = Worker(self.read_daq) def run_repro(self, name_of_repo):
self.threadpool.start(daq_thread) if self.p is None:
self.message(f"Executing process {name_of_repo}")
def on_scan_finished(self): self.p = QProcess()
log.debug("Scan finished") self.p.setWorkingDirectory(str(pathlib.Path(__file__).parent / "repros/"))
# log.debug(pathlib.Path(__file__).parent / "repos")
def read_daq(self, progress_callback): self.p.readyReadStandardOutput.connect(self.handle_stdout)
log.debug("running repro") self.p.readyReadStandardError.connect(self.handle_stderr)
time = np.arange(0, 10, 1 / 30_000.0) self.p.stateChanged.connect(self.handle_state)
data = 2 * np.sin(2 * np.pi * 10 * time) self.p.finished.connect(self.process_finished)
self.p.start("python3", [f"{name_of_repo}" + ".py"])
buffer = ctypes.c_double * len(time)
buffer_ai = uldaq.create_float_buffer(1, len(time)) def handle_stderr(self):
data_c = buffer(*data) if self.p is not None:
log.debug(f"Created C_double data {data_c}") data = self.p.readAllStandardError()
stderr = bytes(data).decode("utf8")
ao_device = self.daq_device.get_ao_device() self.message(stderr)
ai_device = self.daq_device.get_ai_device()
def handle_stdout(self):
er = ai_device.a_in_scan( if self.p is not None:
1, data = self.p.readAllStandardOutput()
1, stdout = bytes(data).decode("utf8")
uldaq.AiInputMode.SINGLE_ENDED, self.message(stdout)
uldaq.Range.BIP10VOLTS,
int(len(data)), def handle_state(self, state):
30_000.0, states = {
uldaq.ScanOption.EXTTRIGGER, QProcess.ProcessState.NotRunning: "Not running",
uldaq.AInScanFlag.DEFAULT, QProcess.ProcessState.Starting: "Starting",
data=buffer_ai, QProcess.ProcessState.Running: "Running",
) }
# ai_device.scan_wait(uldaq.WaitType.WAIT_UNTIL_DONE, timeout=-1) state_name = states[state]
self.message(f"State changed: {state_name}")
err = ao_device.a_out_scan(
0, def process_finished(self):
0, self.text.appendPlainText("Process finished")
uldaq.Range.BIP10VOLTS, self.p = None
int(len(data)),
30_000.0,
uldaq.ScanOption.DEFAULTIO,
uldaq.AOutScanFlag.DEFAULT,
data_c,
)
# INFO: Need to wait for the acquistion
self.daq_disconnect_button.setDisabled(True)
self.daq_connect_button.setEnabled(True)
embed()
exit()
if __name__ == "__main__": if __name__ == "__main__":