import pathlib from PyQt6.QtGui import QAction import sys from PyQt6.QtCore import QSize, QThreadPool, QSettings from PyQt6.QtWidgets import ( QApplication, QGridLayout, QPushButton, QToolBar, QWidget, QMainWindow, QPlainTextEdit, ) import pyqtgraph as pg import uldaq from IPython import embed from scipy.signal import welch, find_peaks import numpy as np import nixio as nix from pyrelacs.util.logging import config_logging import pyrelacs.info as info from pyrelacs.worker import Worker from pyrelacs.repros.repros import Repro log = config_logging() class PyRelacs(QMainWindow): def __init__(self): super().__init__() self.setWindowTitle("PyRelacs") self.setMinimumSize(1000, 1000) self.plot_graph = pg.PlotWidget() self.threadpool = QThreadPool() self.repros = Repro() 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.plot_calibration_button = QPushButton("Plot Calibration") self.plot_calibration_button.setCheckable(True) self.plot_calibration_button.clicked.connect(self.plot_calibration) self.text = QPlainTextEdit() self.text.setReadOnly(True) layout = QGridLayout() layout.addWidget(self.plot_calibration_button, 0, 0) layout.addWidget(self.daq_disconnect_button, 0, 1) layout.addWidget(self.text, 3, 0, 1, 2) layout.addWidget(self.plot_graph, 2, 0, 1, 2) self.toolbar = QToolBar("Repros") self.addToolBar(self.toolbar) self.repros_to_toolbar() self.setFixedSize(QSize(400, 300)) widget = QWidget() widget.setLayout(layout) self.setCentralWidget(widget) self.nix_file = nix.File.open( str(pathlib.Path(__file__).parent / "data"), nix.FileMode.ReadOnly ) def plot_calibration(self): def decibel(power, ref_power=1.0, min_power=1e-20): """Transform power to decibel relative to ref_power. \\[ decibel = 10 \\cdot \\log_{10}(power/ref\\_power) \\] Power values smaller than `min_power` are set to `-np.inf`. Parameters ---------- power: float or array Power values, for example from a power spectrum or spectrogram. ref_power: float or None or 'peak' Reference power for computing decibel. If set to `None` or 'peak', the maximum power is used. min_power: float Power values smaller than `min_power` are set to `-np.inf`. Returns ------- decibel_psd: array Power values in decibel relative to `ref_power`. """ if np.isscalar(power): tmp_power = np.array([power]) decibel_psd = np.array([power]) else: tmp_power = power decibel_psd = power.copy() if ref_power is None or ref_power == "peak": ref_power = np.max(decibel_psd) decibel_psd[tmp_power <= min_power] = float("-inf") decibel_psd[tmp_power > min_power] = 10.0 * np.log10( decibel_psd[tmp_power > min_power] / ref_power ) if np.isscalar(power): return decibel_psd[0] else: return decibel_psd block = self.nix_file.blocks[0] for stim, fish in zip( list(block.data_arrays)[::2], list(block.data_arrays)[1::2] ): beat = stim[:] + fish[:] beat_squared = beat**2 f, powerspec = welch(beat, fs=40_000.0) powerspec = decibel(powerspec) f_sq, powerspec_sq = welch(beat_squared, fs=40_000.0) powerspec_sq = decibel(powerspec_sq) peaks = find_peaks(powerspec_sq, prominence=20)[0] pen = pg.mkPen() self.plot_graph.plot( np.arange(0, len(beat)) / 40_000.0, beat_squared, pen=pen ) 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 repros_to_toolbar(self): repro_names, file_names = self.repros.names_of_repros() for rep, fn in zip(repro_names, file_names): individual_repro_button = QAction(rep, self) individual_repro_button.setStatusTip("Button") individual_repro_button.triggered.connect( lambda checked, n=rep, f=fn: self.run_repro(n, f) ) self.toolbar.addAction(individual_repro_button) def run_repro(self, n, fn): self.text.appendPlainText(f"started Repro {n}, {fn}") worker = Worker(self.repros.run_repro, self.nix_file, n, fn) worker.signals.result.connect(self.print_output) worker.signals.finished.connect(self.thread_complete) worker.signals.progress.connect(self.progress_fn) self.threadpool.start(worker) def print_output(self, s): print(s) def thread_complete(self): print("THREAD COMPLETE!") def progress_fn(self, n): print("%d%% done" % n) def main(): app = QApplication(sys.argv) app.setApplicationName(info.NAME) app.setApplicationVersion(str(info.VERSION)) app.setOrganizationDomain(info.ORGANIZATION) # read window settings settings = QSettings(info.ORGANIZATION, info.NAME) width = int(settings.value("app/width", 1024)) height = int(settings.value("app/height", 768)) x = int(settings.value("app/pos_x", 100)) y = int(settings.value("app/pos_y", 100)) window = PyRelacs() window.setMinimumWidth(200) window.setMinimumHeight(200) window.resize(width, height) window.move(x, y) window.show() exit_code = app.exec() # store window position and size pos = window.pos() settings.setValue("app/width", window.width()) settings.setValue("app/height", window.height()) settings.setValue("app/pos_x", pos.x()) settings.setValue("app/pos_y", pos.y()) sys.exit(exit_code) if __name__ == "__main__": main()