Merge branch 'main' of https://whale.am28.uni-tuebingen.de/git/Awendt/pyrelacs

pyrelacs/repros/attcs3310.py

@@ -0,0 +1,26 @@
+import time
+
+import uldaq
+from IPython import embed
+from pyrelacs.repros.repos import MccDac
+from pyrelacs.util.logging import config_logging
+import numpy as np
+
+log = config_logging()
+
+
+class Attenuator(MccDac):
+    def __init__(self) -> None:
+        super().__init__()
+
+
+if __name__ == "__main__":
+    SAMPLERATE = 40_000.0
+    DURATION = 5
+    AMPLITUDE = 1
+    SINFREQ = 1
+
+    att = Attenuator()
+    # att.set_attenuation_level(db_channel1=5, db_channel2=5)
+
+    att.check_attenuator()

pyrelacs/repros/calbi.py

@@ -1,14 +1,18 @@
 import ctypes
+import faulthandler
+
 import time
 
 import uldaq
 from IPython import embed
-from pyrelacs.repros.repos import MccDac
-from pyrelacs.util.logging import config_logging
 import numpy as np
 import matplotlib.pyplot as plt
 
+from pyrelacs.repros.mccdac import MccDac
+from pyrelacs.util.logging import config_logging
+
 log = config_logging()
+faulthandler.enable()
 
 
 class Calibration(MccDac):
@@ -17,11 +21,11 @@ class Calibration(MccDac):
 
     def run_calibration(self):
         # Stimulus
-        time = np.arange(0, DURATION, 1 / SAMPLERATE)
-        data = AMPLITUDE * np.sin(2 * np.pi * SINFREQ * time)
+        t = np.arange(0, DURATION, 1 / SAMPLERATE)
+        data = AMPLITUDE * np.sin(2 * np.pi * SINFREQ * t)
         # sending stimulus
 
-        stim, ao_device = self.write_analog_dac(
+        self.write_analog(
             data, [0, 0], SAMPLERATE, ScanOption=uldaq.ScanOption.EXTTRIGGER
         )
 
@@ -29,23 +33,66 @@ class Calibration(MccDac):
         #     [0, 1], DURATION, SAMPLERATE, ScanOption=uldaq.ScanOption.EXTTRIGGER
         # )
 
-        # trigger the 0 channel to start the aqcuisition
-        self.write_bit(channel=0, bit=1)
+        self.diggital_trigger()
         try:
-            ao_device.scan_wait(uldaq.WaitType.WAIT_UNTIL_DONE, 15)
+            self.ao_device.scan_wait(uldaq.WaitType.WAIT_UNTIL_DONE, 15)
         except uldaq.ul_exception.ULException:
             log.debug("Operation timed out")
+            # reset the diggital trigger
             self.write_bit(channel=0, bit=0)
-            self.disconnect_dac()
-            self.connect_dac()
+            time.sleep(1)
+            self.set_analog_to_zero()
+
+    def check_amplitude(self):
+        # write to ananlog 1
+        t = np.arange(0, DURATION, 1 / SAMPLERATE)
+        data = AMPLITUDE * np.sin(2 * np.pi * SINFREQ * t)
+
+        # sending stimulus
+        log.debug(f"{data}, {data.shape}")
+        # self.set_attenuation_level(db_channel1=0.0, db_channel2=0.0)
+        # self.set_analog_to_zero()
+        # time.sleep(1)
+        log.debug(self.ao_device)
+        embed()
+        exit()
+
+        self.write_analog(
+            data,
+            [1, 1],
+            SAMPLERATE,
+            ScanOption=uldaq.ScanOption.EXTTRIGGER,
+        )
+        # data_channel_one = self.read_analog(
+        #     [0, 0], DURATION, SAMPLERATE, ScanOption=uldaq.ScanOption.EXTTRIGGER
+        # )
+        log.debug(self.ao_device)
+        time.sleep(1)
+
+        self.diggital_trigger()
+
+        try:
+            self.ao_device.scan_wait(uldaq.WaitType.WAIT_UNTIL_DONE, 15)
+            self.write_bit(channel=0, bit=0)
+            time.sleep(1)
             self.set_analog_to_zero()
+        except uldaq.ul_exception.ULException:
+            log.debug("Operation timed out")
+            # reset the diggital trigger
+            self.write_bit(channel=0, bit=0)
+            time.sleep(1)
+            self.set_analog_to_zero()
+            self.disconnect_dac()
+        embed()
+        exit()
 
 
 if __name__ == "__main__":
     SAMPLERATE = 40_000.0
-    DURATION = 10
+    DURATION = 5
     AMPLITUDE = 1
-    SINFREQ = 100
+    SINFREQ = 10
+
     daq_input = Calibration()
     # daq_input.run_calibration()
     daq_input.check_attenuator()

pyrelacs/repros/mccdac.py

@@ -1,6 +1,7 @@
 from ctypes import Array, c_double
 import time
 from typing import Union
+
 from IPython import embed
 import numpy.typing as npt
 import uldaq
@@ -25,6 +26,10 @@ class MccDac:
         self.dio_device = self.daq_device.get_dio_device()
         log.debug("Connected")
 
+        # log.debug("Activate Attenuator")
+        # self.set_attenuation_level(db_channel1=0.0, db_channel2=0.0)
+        # self.set_analog_to_zero()
+
     def connect_dac(self):
         devices = uldaq.get_daq_device_inventory(uldaq.InterfaceType.USB)
         log.debug(f"Found daq devices {len(devices)}, connecting to the first one")
@@ -38,7 +43,7 @@ class MccDac:
         self.dio_device = self.daq_device.get_dio_device()
         log.debug("Connected")
 
-    def read_analog_daq(
+    def read_analog(
         self,
         channels: list[int],
         duration: int,
@@ -72,7 +77,7 @@ class MccDac:
 
         return data_analog_input
 
-    def write_analog_dac(
+    def write_analog(
         self,
         data: Union[list, npt.NDArray],
         channels: list[int],
@@ -81,48 +86,16 @@ class MccDac:
         ScanOption: uldaq.ScanOption = uldaq.ScanOption.DEFAULTIO,
         AOutScanFlag: uldaq.AOutScanFlag = uldaq.AOutScanFlag.DEFAULT,
     ):
-        """
-
-        Parameters
-        ----------
-        data : Union[list, npt.NDArray]
-
-        channels : list[int]
-
-        duration : int
-
-        samplerate : float
-
-        AiInputMode : uldaq.AiInputMode
-
-        Range : uldaq.Range
-
-        ScanOption : uldaq.ScanOption
-
-        AInScanFlag : uldaq.AOutScanFlag
-
-
-        Returns
-        -------
-        Array[c_double]
-        ao_device
-        """
-
         assert len(channels) == 2, log.error("You can only provide two channels [0, 1]")
 
-        if channels[0] != channels[1]:
-            buffer_len_channels = 2
-        else:
-            buffer_len_channels = 1
+        log.debug(f"{len(data)}, {type(data)}")
 
-        buffer = c_double * (len(data) * buffer_len_channels)
-        assert len(data) != len(data) * buffer_len_channels, log.error(
-            "Data must be duplicated for both channels data[len(data_channel) + len(data_channel1)]..."
-        )
+        buffer = c_double * len(data)
         data_analog_output = buffer(*data)
-
         log.debug(f"Created C_double data {data_analog_output}")
-        try:
+
+        log.info(self.ao_device)
+
         err = self.ao_device.a_out_scan(
             channels[0],
             channels[1],
@@ -133,14 +106,12 @@ class MccDac:
             AOutScanFlag,
             data_analog_output,
         )
-        except Exception as e:
-            print(f"{e}")
-            self.set_analog_to_zero()
-            self.disconnect_dac()
-
-        # ao_device.scan_wait(uldaq.WaitType.WAIT_UNTIL_DONE, 11)
-
-        return data_analog_output, self.ao_device
+        # self.diggital_trigger()
+        # self.ao_device.scan_wait(uldaq.WaitType.WAIT_UNTIL_DONE, 11)
+        # except Exception as e:
+        #     print(f"{e}")
+        #     self.set_analog_to_zero()
+        #     self.disconnect_dac()
 
     def set_analog_to_zero(self, channels: list[int] = [0, 1]):
         try:
@@ -159,15 +130,18 @@ class MccDac:
             log.error("disconnection dac")
             self.disconnect_dac()
 
-    def diggital_trigger(self) -> None:
-        if not self.read_bit(channel=0):
-            self.write_bit(channel=0, bit=1)
+    def diggital_trigger(self, channel: int = 0) -> None:
+        bit_channel = self.read_bit(channel)
+        log.debug(bit_channel)
+        if not bit_channel:
+            self.write_bit(channel, 1)
         else:
-            self.write_bit(channel=0, bit=0)
-            time.time_ns()
-            self.write_bit(channel=0, bit=1)
+            self.write_bit(channel, 0)
+            time.sleep(1)
+            self.write_bit(channel, 1)
 
     def write_bit(self, channel: int = 0, bit: int = 1) -> None:
+        log.debug(self.dio_device)
         self.dio_device.d_config_bit(
             uldaq.DigitalPortType.AUXPORT, channel, uldaq.DigitalDirection.OUTPUT
         )
@@ -176,6 +150,7 @@ class MccDac:
         )
 
     def read_bit(self, channel: int = 0):
+        log.debug(self.dio_device)
         bit = self.dio_device.d_bit_in(uldaq.DigitalPortType.AUXPORT, channel)
         return bit
 
@@ -209,14 +184,6 @@ class MccDac:
         )
         return data_digital_input
 
-    def write_byte(self, channel: int, byte: list[int]):
-        self.dio_device.d_config_bit(
-            uldaq.DigitalPortType.AUXPORT, channel, uldaq.DigitalDirection.OUTPUT
-        )
-        for bit in byte:
-            self.dio_device.d_bit_out(uldaq.DigitalPortType.AUXPORT, channel, bit)
-            time.sleep(1)
-
     def disconnect_dac(self):
         self.daq_device.disconnect()
         self.daq_device.release()

pyrelacs/repros/output.py

@@ -13,22 +13,13 @@ log = config_logging()
 class Output_daq(MccDac):
     def __init__(self) -> None:
         super().__init__()
-        # devices = uldaq.get_daq_device_inventory(uldaq.InterfaceType.USB)
-        # self.daq_device = uldaq.DaqDevice(devices[0])
-        # self.daq_device.connect()
 
     def write_daq(self):
         log.debug("running repro")
         time = np.arange(0, 10, 1 / 30_000.0)
-        data = 2 * np.sin(2 * np.pi * 1 * time)
-        self.send_analog_dac(
-            data, [0, 0], 30_000, ScanOption=uldaq.ScanOption.EXTTRIGGER
-        )
-
-    def trigger(self):
-        self.digital_trigger(1)
-        self.daq_device.disconnect()
-        self.daq_device.release()
+        data = 1 * np.sin(2 * np.pi * 1 * time)
+        self.write_analog(data, [0, 0], 30_000, ScanOption=uldaq.ScanOption.EXTTRIGGER)
+        self.diggital_trigger()
 
 
 if __name__ == "__main__":