[app] move PyRelacs main window to ui subpackage

pyrelacs/app.py

@@ -1,29 +1,12 @@
-import pathlib
-from PyQt6.QtGui import QAction
 import sys
+import pathlib
 
-from PyQt6.QtCore import QSize, QThreadPool, QSettings
+from PyQt6.QtCore import QSettings
-from PyQt6.QtWidgets import (
+from PyQt6.QtWidgets import QApplication
-    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
 
+from . import info
+from .ui.mainwindow import PyRelacs
+from .util.logging import config_logging
 log = config_logging()
 
 

pyrelacs/ui/mainwindow.py

@@ -0,0 +1,184 @@
+from PyQt6.QtGui import QAction
+from PyQt6.QtCore import QSize, QThreadPool
+from PyQt6.QtWidgets import (
+    QApplication,
+    QGridLayout,
+    QPushButton,
+    QToolBar,
+    QWidget,
+    QMainWindow,
+    QPlainTextEdit,
+)
+import uldaq
+import pathlib
+import numpy as np
+import nixio as nix
+import pyqtgraph as pg
+
+from scipy.signal import welch, find_peaks
+
+from ..worker import Worker
+from ..repros.repros import Repro
+from ..util.logging import config_logging
+log = config_logging()
+
+from IPython import embed
+
+class PyRelacs(QMainWindow):
+    def __init__(self):
+        super().__init__()
+        self.setWindowTitle("PyRelacs")
+        self.beat_plot = pg.PlotWidget()
+        self.power_plot = 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.beat_plot, 2, 0, 1, 2)
+        layout.addWidget(self.power_plot, 3, 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)
+
+        filename = pathlib.Path.joinpath(pathlib.Path.cwd(), "data.nix")
+        self.nix_file = nix.File.open(
+            str(filename), nix.FileMode.Overwrite
+        )
+
+    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]
+        colors = ["red", "green", "blue", "black", "yellow"]
+        for i, (stim, fish) in enumerate(
+            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(colors[i])
+            self.beat_plot.plot(
+                np.arange(0, len(beat)) / 40_000.0,
+                beat_squared,
+                pen=pen,
+                # name=stim.name,
+            )
+            self.power_plot.plot(f_sq, powerspec_sq, pen=pen)
+            self.power_plot.plot(f[peaks], powerspec_sq[peaks], pen=None, symbol="x")
+
+    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)