forked from awendt/pyrelacs
152 lines
4.6 KiB
Python
152 lines
4.6 KiB
Python
from PyQt6.QtGui import QAction
|
|
import sys
|
|
import pathlib
|
|
import ctypes
|
|
|
|
from PyQt6.QtCore import QSize, QThreadPool, Qt
|
|
from PyQt6.QtWidgets import (
|
|
QApplication,
|
|
QGridLayout,
|
|
QPushButton,
|
|
QToolBar,
|
|
QWidget,
|
|
QMainWindow,
|
|
)
|
|
import tomli
|
|
import uldaq
|
|
from IPython import embed
|
|
import numpy as np
|
|
|
|
from pyrelacs.util.logging import config_logging
|
|
from pyrelacs.daq_thread import Worker
|
|
|
|
log = config_logging()
|
|
|
|
|
|
class PyRelacs(QMainWindow):
|
|
def __init__(self):
|
|
super().__init__()
|
|
self.setWindowTitle("PyRelacs")
|
|
self.threadpool = QThreadPool()
|
|
|
|
self.daq_connect_button = QPushButton("Connect Daq")
|
|
self.daq_connect_button.setCheckable(True)
|
|
self.daq_connect_button.clicked.connect(self.connect_dac)
|
|
|
|
self.daq_disconnect_button = QPushButton("Disconnect Daq")
|
|
self.daq_disconnect_button.setCheckable(True)
|
|
self.daq_disconnect_button.clicked.connect(self.disconnect_dac)
|
|
|
|
self.repro_button = QPushButton("Load Repros")
|
|
self.repro_button.setCheckable(True)
|
|
self.repro_button.clicked.connect(self.repro)
|
|
|
|
layout = QGridLayout()
|
|
layout.addWidget(self.daq_connect_button, 0, 0)
|
|
layout.addWidget(self.daq_disconnect_button, 0, 1)
|
|
layout.addWidget(self.repro_button, 1, 0)
|
|
|
|
self.toolbar = QToolBar("Repros")
|
|
self.addToolBar(self.toolbar)
|
|
|
|
self.setFixedSize(QSize(400, 300))
|
|
widget = QWidget()
|
|
widget.setLayout(layout)
|
|
self.setCentralWidget(widget)
|
|
|
|
def connect_dac(self):
|
|
devices = uldaq.get_daq_device_inventory(uldaq.InterfaceType.USB)
|
|
try:
|
|
self.daq_device = uldaq.DaqDevice(devices[0])
|
|
log.debug(f"Found daq devices {len(devices)}, connecting to the first one")
|
|
self.daq_device.connect()
|
|
log.debug("Connected")
|
|
except IndexError:
|
|
log.debug("DAQ is not connected, closing")
|
|
QApplication.quit()
|
|
self.daq_connect_button.setDisabled(True)
|
|
|
|
def disconnect_dac(self):
|
|
try:
|
|
log.debug(f"{self.daq_device}")
|
|
self.daq_device.disconnect()
|
|
self.daq_device.release()
|
|
log.debug(f"{self.daq_device}")
|
|
self.daq_disconnect_button.setDisabled(True)
|
|
self.daq_connect_button.setEnabled(True)
|
|
except AttributeError:
|
|
log.debug("DAQ was not connected")
|
|
|
|
def repro(self):
|
|
config_path = pathlib.Path(__file__).parent.resolve() / "repro.toml"
|
|
if not config_path.is_file:
|
|
log.error("repro.toml not found")
|
|
QApplication.quit()
|
|
repors = []
|
|
|
|
with open(config_path, mode="rb") as fp:
|
|
self.config = tomli.load(fp)
|
|
for r in self.config:
|
|
repors.append(r)
|
|
|
|
individual_repro_button = QAction(f"{repors[0]}", self)
|
|
individual_repro_button.setStatusTip("Button")
|
|
individual_repro_button.triggered.connect(self.run_repro)
|
|
self.toolbar.addAction(individual_repro_button)
|
|
|
|
def run_repro(self):
|
|
daq_thread = Worker(self.read_daq)
|
|
self.threadpool.start(daq_thread)
|
|
|
|
def on_scan_finished(self):
|
|
log.debug("Scan finished")
|
|
|
|
def read_daq(self, progress_callback):
|
|
log.debug("running repro")
|
|
time = np.arange(0, 10, 1 / 30_000.0)
|
|
data = 2 * np.sin(2 * np.pi * 10 * time)
|
|
|
|
buffer = ctypes.c_double * len(time)
|
|
buffer_ai = uldaq.create_float_buffer(1, len(time))
|
|
data_c = buffer(*data)
|
|
log.debug(f"Created C_double data {data_c}")
|
|
|
|
ao_device = self.daq_device.get_ao_device()
|
|
ai_device = self.daq_device.get_ai_device()
|
|
|
|
er = ai_device.a_in_scan(
|
|
1,
|
|
1,
|
|
uldaq.AiInputMode.SINGLE_ENDED,
|
|
uldaq.Range.BIP10VOLTS,
|
|
int(len(data)),
|
|
30_000.0,
|
|
uldaq.ScanOption.EXTTRIGGER,
|
|
uldaq.AInScanFlag.DEFAULT,
|
|
data=buffer_ai,
|
|
)
|
|
# ai_device.scan_wait(uldaq.WaitType.WAIT_UNTIL_DONE, timeout=-1)
|
|
|
|
err = ao_device.a_out_scan(
|
|
0,
|
|
0,
|
|
uldaq.Range.BIP10VOLTS,
|
|
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__":
|
|
app = QApplication(sys.argv)
|
|
window = PyRelacs()
|
|
window.show()
|
|
app.exec()
|