fixing seg fault by returning data_analog_output

pyrelacs/repros/attcs3310.py

@@ -1,26 +0,0 @@
-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

@@ -19,60 +19,29 @@ class Calibration(MccDac):
     def __init__(self) -> None:
         super().__init__()
 
-    def run_calibration(self):
-        # Stimulus
-        t = np.arange(0, DURATION, 1 / SAMPLERATE)
-        data = AMPLITUDE * np.sin(2 * np.pi * SINFREQ * t)
-        # sending stimulus
-
-        self.write_analog(
-            data, [0, 0], SAMPLERATE, ScanOption=uldaq.ScanOption.EXTTRIGGER
-        )
-
-        # read_data = self.read_analog_daq(
-        #     [0, 1], DURATION, SAMPLERATE, ScanOption=uldaq.ScanOption.EXTTRIGGER
-        # )
-
-        self.diggital_trigger()
-        try:
-            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)
-            time.sleep(1)
-            self.set_analog_to_zero()
-
     def check_amplitude(self):
+        self.set_attenuation_level(db_channel1=0.0, db_channel2=0.0)
         # write to ananlog 1
         t = np.arange(0, DURATION, 1 / SAMPLERATE)
         data = AMPLITUDE * np.sin(2 * np.pi * SINFREQ * t)
+        data_channels = np.zeros(2 * len(data))
+        # c = [(i,for i,j in zip(data, data)]
 
-        # 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(
+        stim = self.write_analog(
             data,
-            [1, 1],
+            [0, 0],
             SAMPLERATE,
             ScanOption=uldaq.ScanOption.EXTTRIGGER,
         )
-        # data_channel_one = self.read_analog(
-        #     [0, 0], DURATION, SAMPLERATE, ScanOption=uldaq.ScanOption.EXTTRIGGER
-        # )
-        log.debug(self.ao_device)
+        data_channel_one = self.read_analog(
+            [0, 0], DURATION, SAMPLERATE, ScanOption=uldaq.ScanOption.EXTTRIGGER
+        )
         time.sleep(1)
 
         self.diggital_trigger()
 
         try:
-            self.ao_device.scan_wait(uldaq.WaitType.WAIT_UNTIL_DONE, 15)
+            self.ai_device.scan_wait(uldaq.WaitType.WAIT_UNTIL_DONE, 15)
             self.write_bit(channel=0, bit=0)
             time.sleep(1)
             self.set_analog_to_zero()
@@ -90,9 +59,9 @@ class Calibration(MccDac):
 if __name__ == "__main__":
     SAMPLERATE = 40_000.0
     DURATION = 5
-    AMPLITUDE = 1
+    AMPLITUDE = 0.5
     SINFREQ = 10
 
-    daq_input = Calibration()
-    # daq_input.run_calibration()
-    daq_input.check_attenuator()
+    cal = Calibration()
+    # cal.ccheck_attenuator()
+    cal.check_amplitude()

pyrelacs/repros/mccdac.py

@@ -1,7 +1,6 @@
 from ctypes import Array, c_double
 import time
 from typing import Union
-
 from IPython import embed
 import numpy.typing as npt
 import uldaq
@@ -26,10 +25,6 @@ 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")
@@ -85,33 +80,30 @@ class MccDac:
         Range: uldaq.Range = uldaq.Range.BIP10VOLTS,
         ScanOption: uldaq.ScanOption = uldaq.ScanOption.DEFAULTIO,
         AOutScanFlag: uldaq.AOutScanFlag = uldaq.AOutScanFlag.DEFAULT,
-    ):
+    ) -> Array[c_double]:
         assert len(channels) == 2, log.error("You can only provide two channels [0, 1]")
 
-        log.debug(f"{len(data)}, {type(data)}")
-
         buffer = c_double * len(data)
         data_analog_output = buffer(*data)
-        log.debug(f"Created C_double data {data_analog_output}")
 
-        log.info(self.ao_device)
+        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,
-        )
-        # 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()
+        return data_analog_output
 
     def set_analog_to_zero(self, channels: list[int] = [0, 1]):
         try:
@@ -130,18 +122,15 @@ class MccDac:
             log.error("disconnection dac")
             self.disconnect_dac()
 
-    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)
+    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)
-            time.sleep(1)
-            self.write_bit(channel, 1)
+            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:
-        log.debug(self.dio_device)
         self.dio_device.d_config_bit(
             uldaq.DigitalPortType.AUXPORT, channel, uldaq.DigitalDirection.OUTPUT
         )
@@ -150,7 +139,6 @@ 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
 
@@ -205,7 +193,7 @@ class MccDac:
         SINFREQ = 1
         t = np.arange(0, DURATION, 1 / SAMPLERATE)
         data = AMPLITUDE * np.sin(2 * np.pi * SINFREQ * t)
-        data_channels = np.concatenate((data, data))
+        # data_channels = np.concatenate((data, data))
 
         db_values = [0, 0, -2, -5, -10, -20, -50]
         db_values = [0, -10, -20]
@@ -219,9 +207,9 @@ class MccDac:
             else:
                 self.set_attenuation_level(db_value, db_value)
 
-            stim, ao_device = self.write_analog_dac(
-                data_channels,
-                [0, 1],
+            _ = self.write_analog(
+                data,
+                [0, 0],
                 SAMPLERATE,
                 ScanOption=uldaq.ScanOption.EXTTRIGGER,
                 Range=uldaq.Range.BIP10VOLTS,
@@ -229,7 +217,9 @@ class MccDac:
             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)
+                self.write_bit(channel=0, bit=0)
+                self.set_analog_to_zero()
             except uldaq.ul_exception.ULException:
                 log.debug("Operation timed out")
                 self.write_bit(channel=0, bit=0)
@@ -244,7 +234,6 @@ class MccDac:
 
             log.info("Sleeping for 1 second, before next attenuation")
             time.sleep(1)
-        self.deactivate_attenuator()
 
     def set_attenuation_level(
         self,