pyrelacs/repros/attcs3310.py

@@ -1,130 +0,0 @@
-import ctypes
-import time
-
-import uldaq
-from IPython import embed
-from pyrelacs.repros.repos import Repos
-from pyrelacs.util.logging import config_logging
-import numpy as np
-import matplotlib.pyplot as plt
-
-log = config_logging()
-
-
-class Attenuator(Repos):
-    def __init__(self) -> None:
-        super().__init__()
-
-    def check_attenuator(self):
-        """
-        ident     : attdev-1
-        strobepin : 6
-        datainpin : 5
-        dataoutpin: -1
-        cspin     : 4
-        mutepin   : 7
-        zcenpin   : -1
-        """
-        t = np.arange(0, DURATION, 1 / SAMPLERATE)
-        data = AMPLITUDE * np.sin(2 * np.pi * SINFREQ * t)
-        data_channels = np.concatenate((data, data))
-
-        db_values = [0, 0, -2, -5, -10, -20, -50]
-        for i, db_value in enumerate(db_values):
-            log.info(f"Attenuating the Channels, with {db_value}")
-            if i == 1:
-                log.info("Muting the Channels")
-                self.set_attenuation_level(
-                    db_value, db_value, mute_channel1=True, mute_channel2=True
-                )
-            else:
-                self.set_attenuation_level(db_value, db_value)
-
-            stim, ao_device = self.write_analog_dac(
-                data_channels,
-                [0, 1],
-                SAMPLERATE,
-                ScanOption=uldaq.ScanOption.EXTTRIGGER,
-                Range=uldaq.Range.BIP10VOLTS,
-            )
-            self.diggital_trigger()
-
-            try:
-                ao_device.scan_wait(uldaq.WaitType.WAIT_UNTIL_DONE, 15)
-            except uldaq.ul_exception.ULException:
-                log.debug("Operation timed out")
-                self.write_bit(channel=0, bit=0)
-                self.disconnect_dac()
-                self.connect_dac()
-                self.set_analog_to_zero()
-            finally:
-                self.write_bit(channel=0, bit=0)
-                self.disconnect_dac()
-                self.connect_dac()
-                self.set_analog_to_zero()
-
-            log.info("Sleeping for 1 second, before next attenuation")
-            time.sleep(1)
-        self.deactivate_attenuator()
-
-    def set_attenuation_level(
-        self,
-        db_channel1: float = 5.0,
-        db_channel2: float = 5.0,
-        mute_channel1: bool = False,
-        mute_channel2: bool = False,
-    ):
-        """
-        ident     : attdev-1
-        strobepin : 6
-        datainpin : 5
-        dataoutpin: -1
-        cspin     : 4
-        mutepin   : 7
-        zcenpin   : -1
-        """
-
-        self.activate_attenuator()
-        hardware_possible_db = np.arange(-95.5, 32.0, 0.5)
-        byte_number = np.arange(1, 256)
-        byte_number_db1 = byte_number[hardware_possible_db == db_channel1][0]
-        binary_db1 = np.binary_repr(byte_number_db1, width=8)
-        byte_number_db2 = byte_number[hardware_possible_db == db_channel2][0]
-        binary_db2 = np.binary_repr(byte_number_db2, width=8)
-        if mute_channel1:
-            log.info("Muting channel one")
-            binary_db1 = "00000000"
-        if mute_channel2:
-            log.info("Muting channel one")
-            binary_db2 = "00000000"
-
-        channels_db = binary_db1 + binary_db2
-        self.write_bit(channel=4, bit=0)
-        for b in channels_db:
-            self.write_bit(channel=5, bit=int(b))
-            time.time_ns()
-            self.write_bit(channel=6, bit=1)
-            time.time_ns()
-            self.write_bit(channel=6, bit=0)
-            time.time_ns()
-        self.write_bit(channel=4, bit=1)
-
-    def activate_attenuator(self):
-        for ch, b in zip([4, 5, 6, 7], [1, 0, 0, 1]):
-            self.write_bit(channel=ch, bit=b)
-
-    def deactivate_attenuator(self):
-        # mute should be enabled for starting calibration
-        self.write_bit(channel=7, bit=0)
-
-
-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,5 +1,4 @@
 import ctypes
-import time
 
 import uldaq
 from IPython import embed
@@ -21,31 +20,28 @@ class Calibration(Repos):
         data = AMPLITUDE * np.sin(2 * np.pi * SINFREQ * time)
         # sending stimulus
 
-        stim, ao_device = self.write_analog_dac(
+        stim, ao_device = self.send_analog_dac(
             data, [0, 0], SAMPLERATE, ScanOption=uldaq.ScanOption.EXTTRIGGER
         )
-
         # read_data = self.read_analog_daq(
         #     [0, 1], DURATION, SAMPLERATE, ScanOption=uldaq.ScanOption.EXTTRIGGER
         # )
+        data = self.read_digitalio([0, 0], DURATION, SAMPLERATE)
+        self.digital_trigger()
+        ao_device.scan_wait(uldaq.WaitType.WAIT_UNTIL_DONE, -1)
+        self.digital_trigger(data=0)
 
-        # trigger the 0 channel to start the aqcuisition
+        self.disconnect_dac()
-        self.write_bit(channel=0, bit=1)
+        self.connect_dac()
-        try:
+        self.set_analog_to_zero()
-            ao_device.scan_wait(uldaq.WaitType.WAIT_UNTIL_DONE, 15)
+        embed()
-        except uldaq.ul_exception.ULException:
+        exit()
-            log.debug("Operation timed out")
-            self.write_bit(channel=0, bit=0)
-            self.disconnect_dac()
-            self.connect_dac()
-            self.set_analog_to_zero()
 
 
 if __name__ == "__main__":
     SAMPLERATE = 40_000.0
-    DURATION = 10
+    DURATION = 5
     AMPLITUDE = 1
-    SINFREQ = 100
+    SINFREQ = 1
     daq_input = Calibration()
-    # daq_input.run_calibration()
+    daq_input.run_calibration()
-    daq_input.attenuator()

pyrelacs/repros/repos.py

@@ -1,5 +1,4 @@
 from ctypes import Array, c_double
-import time
 from typing import Union
 from IPython import embed
 import numpy.typing as npt
@@ -48,15 +47,14 @@ class Repos:
         ScanOption: uldaq.ScanOption = uldaq.ScanOption.DEFAULTIO,
         AInScanFlag: uldaq.AInScanFlag = uldaq.AInScanFlag.DEFAULT,
     ) -> Array[c_double]:
-        assert len(channels) == 2, log.error("You can only provide two channels [0, 1]")
+        if channels[0] == channels[1]:
-
+            channel_len = 1
-        if channels[0] != channels[1]:
-            buffer_len_channels = 2
         else:
-            buffer_len_channels = 1
+            channel_len = len(channels)
+        assert len(channels) == 2, log.error("Please provide a list with two ints")
 
         buffer_len = np.shape(np.arange(0, duration, 1 / samplerate))[0]
-        data_analog_input = uldaq.create_float_buffer(buffer_len_channels, buffer_len)
+        data_analog_input = uldaq.create_float_buffer(channel_len, buffer_len)
 
         er = self.ai_device.a_in_scan(
             channels[0],
@@ -69,10 +67,11 @@ class Repos:
             AInScanFlag,
             data=data_analog_input,
         )
+        # ai_device.scan_wait(uldaq.WaitType.WAIT_UNTIL_DONE, timeout=-1)
 
         return data_analog_input
 
-    def write_analog_dac(
+    def send_analog_dac(
         self,
         data: Union[list, npt.NDArray],
         channels: list[int],
@@ -106,79 +105,52 @@ class Repos:
         -------
         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
 
-        buffer = c_double * (len(data) * buffer_len_channels)
+        """
-        assert len(data) != len(data) * buffer_len_channels, log.error(
+        buffer = c_double * len(data)
-            "Data must be duplicated for both channels data[len(data_channel) + len(data_channel1)]..."
-        )
         data_analog_output = buffer(*data)
 
         log.debug(f"Created C_double data {data_analog_output}")
-        try:
-            err = self.ao_device.a_out_scan(
-                channels[0],
-                channels[1],
-                Range,
-                int(len(data)),
-                samplerate,
-                ScanOption,
-                AOutScanFlag,
-                data_analog_output,
-            )
-        except Exception as e:
-            print(f"{e}")
-            self.set_analog_to_zero()
-            self.disconnect_dac()
 
+        err = self.ao_device.a_out_scan(
+            channels[0],
+            channels[1],
+            Range,
+            int(len(data)),
+            samplerate,
+            ScanOption,
+            AOutScanFlag,
+            data_analog_output,
+        )
+        log.info(f"The actual scan rate was {err}")
         # ao_device.scan_wait(uldaq.WaitType.WAIT_UNTIL_DONE, 11)
 
         return data_analog_output, self.ao_device
 
     def set_analog_to_zero(self, channels: list[int] = [0, 1]):
-        try:
+        err = self.ao_device.a_out_list(
-            err = self.ao_device.a_out_list(
+            channels[0],
-                channels[0],
+            channels[1],
-                channels[1],
+            [
-                [
+                uldaq.Range.BIP10VOLTS,
-                    uldaq.Range.BIP10VOLTS,
+                uldaq.Range.BIP10VOLTS,
-                    uldaq.Range.BIP10VOLTS,
+            ],
-                ],
+            uldaq.AOutListFlag.DEFAULT,
-                uldaq.AOutListFlag.DEFAULT,
+            [0, 0],
-                [0, 0],
+        )
-            )
+
-        except Exception as e:
+    def digital_trigger(self, channel: int = 0, data: int = 1) -> None:
-            log.error("f{e}")
+        log.info(f"{self.daq_device}")
-            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)
-        else:
-            self.write_bit(channel=0, bit=0)
-            time.time_ns()
-            self.write_bit(channel=0, bit=1)
 
-    def write_bit(self, channel: int = 0, bit: int = 1) -> None:
         self.dio_device.d_config_bit(
             uldaq.DigitalPortType.AUXPORT, channel, uldaq.DigitalDirection.OUTPUT
         )
+
         self.dio_device.d_bit_out(
-            uldaq.DigitalPortType.AUXPORT, bit_number=channel, data=bit
+            uldaq.DigitalPortType.AUXPORT, bit_number=channel, data=data
         )
 
-    def read_bit(self, channel: int = 0):
-        bit = self.dio_device.d_bit_in(uldaq.DigitalPortType.AUXPORT, channel)
-        return bit
-
     def read_digitalio(
         self,
         channels: list[int],
@@ -209,14 +181,6 @@ class Repos:
         )
         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()