example of reading input

pyrelacs/daq_thread.py

@@ -1,78 +0,0 @@
-import traceback
-import sys
-from PyQt6.QtCore import QRunnable, pyqtSignal, pyqtSlot, QObject
-
-from pyrelacs.util.logging import config_logging
-
-log = config_logging()
-
-
-class WorkerSignals(QObject):
-    """
-    Defines the signals available from a running worker thread.
-
-    Supported signals are:
-
-    finished
-        No data
-
-    error
-        tuple (exctype, value, traceback.format_exc() )
-
-    result
-        object data returned from processing, anything
-
-    progress
-        int indicating % progress
-
-    """
-
-    finished = pyqtSignal()
-    error = pyqtSignal(tuple)
-    result = pyqtSignal(object)
-    progress = pyqtSignal(int)
-
-
-class Worker(QRunnable):
-    """
-    Worker thread
-
-    Inherits from QRunnable to handler worker thread setup, signals and wrap-up.
-
-    :param callback: The function callback to run on this worker thread. Supplied args and
-                     kwargs will be passed through to the runner.
-    :type callback: function
-    :param args: Arguments to pass to the callback function
-    :param kwargs: Keywords to pass to the callback function
-
-    """
-
-    def __init__(self, fn, *args, **kwargs):
-        super(Worker, self).__init__()
-
-        # Store constructor arguments (re-used for processing)
-        self.fn = fn
-        self.args = args
-        self.kwargs = kwargs
-        self.signals = WorkerSignals()
-
-        # Add the callback to our kwargs
-        self.kwargs["progress_callback"] = self.signals.progress
-
-    @pyqtSlot()
-    def run(self):
-        """
-        Initialise the runner function with passed args, kwargs.
-        """
-
-        # Retrieve args/kwargs here; and fire processing using them
-        try:
-            result = self.fn(*self.args, **self.kwargs)
-        except:
-            traceback.print_exc()
-            exctype, value = sys.exc_info()[:2]
-            self.signals.error.emit((exctype, value, traceback.format_exc()))
-        else:
-            self.signals.result.emit(result)  # Return the result of the processing
-        finally:
-            self.signals.finished.emit()  # Done

pyrelacs/repro.toml

@@ -1,6 +0,0 @@
-[Sinus]
-duration = 10
-fs = 30000.0
-amplitude = 1
-freq = 10
-

pyrelacs/repros/input.py

@@ -1,54 +1,28 @@
-from typing import Optional
-
 import uldaq
-from IPython import embed
-import typer
-from typing_extensions import Annotated
 import matplotlib.pyplot as plt
 
 from pyrelacs.util.logging import config_logging
+from .repos import Repos
 
 log = config_logging()
 
 
-class ReadData:
+class ReadData(Repos):
     def __init__(self) -> None:
-        devices = uldaq.get_daq_device_inventory(uldaq.InterfaceType.USB)
-        log.debug(f"Found daq devices {len(devices)}, connecting to the first one")
-        self.daq_device = uldaq.DaqDevice(devices[0])
-        self.daq_device.connect()
-        log.debug("Connected")
+        super().__init__()
 
-    def read_analog_in(self, channel: Annotated[Optional[int], typer.Argument()] = 0):
+    def analog_in(self) -> None:
         # Get the Ananlog In device and Analog Info
-        ai_device = self.daq_device.get_ai_device()
-        ai_info = ai_device.get_info()
-        log.debug(
-            f"Analog info,\n Channels available {ai_info.get_num_chans()}, \n Max Samplerate: {ai_info.get_max_scan_rate()}"
-        )
-        buffer_len = 1_000_000
-        buf = uldaq.create_float_buffer(1, buffer_len)
-
-        er = ai_device.a_in_scan(
-            1,
-            1,
-            uldaq.AiInputMode.SINGLE_ENDED,
-            uldaq.Range.BIP10VOLTS,
-            buffer_len,
-            500_000,
-            uldaq.ScanOption.DEFAULTIO,
-            uldaq.AInScanFlag.DEFAULT,
-            data=buf,
+        data = self.read_analog_daq(
+            [0, 0],
+            10,
+            3000.0,
         )
-        ai_device.scan_wait(uldaq.WaitType.WAIT_UNTIL_DONE, timeout=-1)
-        log.debug("Scanning")
-
-        self.daq_device.disconnect()
-        self.daq_device.release()
-        plt.plot(buf)
+        plt.plot(data)
         plt.show()
+        self.disconnect_dac()
 
 
 if __name__ == "__main__":
     daq_input = ReadData()
-    typer.run(daq_input.read_analog_in)
+    daq_input.analog_in()

pyrelacs/repros/repos.py

@@ -1,5 +1,6 @@
+from ctypes import Array, c_double
 import uldaq
-from IPython import embed
+import numpy as np
 
 from pyrelacs.util.logging import config_logging
 
@@ -8,6 +9,58 @@ log = config_logging()
 
 class Repos:
     def __init__(self) -> None:
+        devices = uldaq.get_daq_device_inventory(uldaq.InterfaceType.USB)
+        log.debug(f"Found daq devices {len(devices)}, connecting to the first one")
+        if len(devices) == 0:
+            log.error("Did not found daq devices, please connect one")
+            exit(1)
+        self.daq_device = uldaq.DaqDevice(devices[0])
+        self.daq_device.connect()
+        log.debug("Connected")
+
+    def read_analog_daq(
+        self,
+        channels: list[int],
+        duration: int,
+        samplerate: float,
+        AiInputMode: uldaq.AiInputMode = uldaq.AiInputMode.SINGLE_ENDED,
+        Range: uldaq.Range = uldaq.Range.BIP10VOLTS,
+        ScanOption: uldaq.ScanOption = uldaq.ScanOption.DEFAULTIO,
+        AInScanFlag: uldaq.AInScanFlag = uldaq.AInScanFlag.DEFAULT,
+    ) -> Array[c_double]:
+
+        if len(channels) > 2:
+            log.error("Please provide only two channels")
+        # Get the Ananlog In device and Analog Info
+
+        ai_device = self.daq_device.get_ai_device()
+        ai_info = ai_device.get_info()
+
+        buffer_len = np.shape(np.arange(0, duration, 1 / samplerate))[0]
+        data_analog_input = uldaq.create_float_buffer(1, buffer_len)
+
+        er = ai_device.a_in_scan(
+            channels[0],
+            channels[1],
+            AiInputMode,
+            Range,
+            buffer_len,
+            samplerate,
+            ScanOption,
+            AInScanFlag,
+            data=data_analog_input,
+        )
+        ai_device.scan_wait(uldaq.WaitType.WAIT_UNTIL_DONE, timeout=-1)
+        log.info(f"Sampled with {er}")
+        log.debug("Scanning")
+
+        return data_analog_input
+
+    def disconnect_dac(self):
+        self.daq_device.disconnect()
+        self.daq_device.release()
+
+    def write_daq(self) -> None:
         pass
 
     def run_repo(self) -> None:
@@ -18,9 +71,3 @@ class Repos:
 
     def reload_repo(self) -> None:
         pass
-
-    def read_daq(self) -> None:
-        pass
-
-    def write_daq(self) -> None:
-        pass