fixing seg fault by returning data_analog_output
This commit is contained in:
parent
a9be09dc06
commit
26f43151a2
@ -1,7 +1,6 @@
|
|||||||
from ctypes import Array, c_double
|
from ctypes import Array, c_double
|
||||||
import time
|
import time
|
||||||
from typing import Union
|
from typing import Union
|
||||||
|
|
||||||
from IPython import embed
|
from IPython import embed
|
||||||
import numpy.typing as npt
|
import numpy.typing as npt
|
||||||
import uldaq
|
import uldaq
|
||||||
@ -26,10 +25,6 @@ class MccDac:
|
|||||||
self.dio_device = self.daq_device.get_dio_device()
|
self.dio_device = self.daq_device.get_dio_device()
|
||||||
log.debug("Connected")
|
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):
|
def connect_dac(self):
|
||||||
devices = uldaq.get_daq_device_inventory(uldaq.InterfaceType.USB)
|
devices = uldaq.get_daq_device_inventory(uldaq.InterfaceType.USB)
|
||||||
log.debug(f"Found daq devices {len(devices)}, connecting to the first one")
|
log.debug(f"Found daq devices {len(devices)}, connecting to the first one")
|
||||||
@ -85,17 +80,14 @@ class MccDac:
|
|||||||
Range: uldaq.Range = uldaq.Range.BIP10VOLTS,
|
Range: uldaq.Range = uldaq.Range.BIP10VOLTS,
|
||||||
ScanOption: uldaq.ScanOption = uldaq.ScanOption.DEFAULTIO,
|
ScanOption: uldaq.ScanOption = uldaq.ScanOption.DEFAULTIO,
|
||||||
AOutScanFlag: uldaq.AOutScanFlag = uldaq.AOutScanFlag.DEFAULT,
|
AOutScanFlag: uldaq.AOutScanFlag = uldaq.AOutScanFlag.DEFAULT,
|
||||||
):
|
) -> Array[c_double]:
|
||||||
assert len(channels) == 2, log.error("You can only provide two channels [0, 1]")
|
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)
|
buffer = c_double * len(data)
|
||||||
data_analog_output = buffer(*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(
|
err = self.ao_device.a_out_scan(
|
||||||
channels[0],
|
channels[0],
|
||||||
channels[1],
|
channels[1],
|
||||||
@ -106,12 +98,12 @@ class MccDac:
|
|||||||
AOutScanFlag,
|
AOutScanFlag,
|
||||||
data_analog_output,
|
data_analog_output,
|
||||||
)
|
)
|
||||||
# self.diggital_trigger()
|
except Exception as e:
|
||||||
# self.ao_device.scan_wait(uldaq.WaitType.WAIT_UNTIL_DONE, 11)
|
print(f"{e}")
|
||||||
# except Exception as e:
|
self.set_analog_to_zero()
|
||||||
# print(f"{e}")
|
self.disconnect_dac()
|
||||||
# self.set_analog_to_zero()
|
|
||||||
# self.disconnect_dac()
|
return data_analog_output
|
||||||
|
|
||||||
def set_analog_to_zero(self, channels: list[int] = [0, 1]):
|
def set_analog_to_zero(self, channels: list[int] = [0, 1]):
|
||||||
try:
|
try:
|
||||||
@ -130,18 +122,15 @@ class MccDac:
|
|||||||
log.error("disconnection dac")
|
log.error("disconnection dac")
|
||||||
self.disconnect_dac()
|
self.disconnect_dac()
|
||||||
|
|
||||||
def diggital_trigger(self, channel: int = 0) -> None:
|
def diggital_trigger(self) -> None:
|
||||||
bit_channel = self.read_bit(channel)
|
if not self.read_bit(channel=0):
|
||||||
log.debug(bit_channel)
|
self.write_bit(channel=0, bit=1)
|
||||||
if not bit_channel:
|
|
||||||
self.write_bit(channel, 1)
|
|
||||||
else:
|
else:
|
||||||
self.write_bit(channel, 0)
|
self.write_bit(channel=0, bit=0)
|
||||||
time.sleep(1)
|
time.time_ns()
|
||||||
self.write_bit(channel, 1)
|
self.write_bit(channel=0, bit=1)
|
||||||
|
|
||||||
def write_bit(self, channel: int = 0, bit: int = 1) -> None:
|
def write_bit(self, channel: int = 0, bit: int = 1) -> None:
|
||||||
log.debug(self.dio_device)
|
|
||||||
self.dio_device.d_config_bit(
|
self.dio_device.d_config_bit(
|
||||||
uldaq.DigitalPortType.AUXPORT, channel, uldaq.DigitalDirection.OUTPUT
|
uldaq.DigitalPortType.AUXPORT, channel, uldaq.DigitalDirection.OUTPUT
|
||||||
)
|
)
|
||||||
@ -150,7 +139,6 @@ class MccDac:
|
|||||||
)
|
)
|
||||||
|
|
||||||
def read_bit(self, channel: int = 0):
|
def read_bit(self, channel: int = 0):
|
||||||
log.debug(self.dio_device)
|
|
||||||
bit = self.dio_device.d_bit_in(uldaq.DigitalPortType.AUXPORT, channel)
|
bit = self.dio_device.d_bit_in(uldaq.DigitalPortType.AUXPORT, channel)
|
||||||
return bit
|
return bit
|
||||||
|
|
||||||
@ -205,7 +193,7 @@ class MccDac:
|
|||||||
SINFREQ = 1
|
SINFREQ = 1
|
||||||
t = np.arange(0, DURATION, 1 / SAMPLERATE)
|
t = np.arange(0, DURATION, 1 / SAMPLERATE)
|
||||||
data = AMPLITUDE * np.sin(2 * np.pi * SINFREQ * t)
|
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, 0, -2, -5, -10, -20, -50]
|
||||||
db_values = [0, -10, -20]
|
db_values = [0, -10, -20]
|
||||||
@ -219,9 +207,9 @@ class MccDac:
|
|||||||
else:
|
else:
|
||||||
self.set_attenuation_level(db_value, db_value)
|
self.set_attenuation_level(db_value, db_value)
|
||||||
|
|
||||||
stim, ao_device = self.write_analog_dac(
|
_ = self.write_analog(
|
||||||
data_channels,
|
data,
|
||||||
[0, 1],
|
[0, 0],
|
||||||
SAMPLERATE,
|
SAMPLERATE,
|
||||||
ScanOption=uldaq.ScanOption.EXTTRIGGER,
|
ScanOption=uldaq.ScanOption.EXTTRIGGER,
|
||||||
Range=uldaq.Range.BIP10VOLTS,
|
Range=uldaq.Range.BIP10VOLTS,
|
||||||
@ -229,7 +217,9 @@ class MccDac:
|
|||||||
self.diggital_trigger()
|
self.diggital_trigger()
|
||||||
|
|
||||||
try:
|
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:
|
except uldaq.ul_exception.ULException:
|
||||||
log.debug("Operation timed out")
|
log.debug("Operation timed out")
|
||||||
self.write_bit(channel=0, bit=0)
|
self.write_bit(channel=0, bit=0)
|
||||||
@ -244,7 +234,6 @@ class MccDac:
|
|||||||
|
|
||||||
log.info("Sleeping for 1 second, before next attenuation")
|
log.info("Sleeping for 1 second, before next attenuation")
|
||||||
time.sleep(1)
|
time.sleep(1)
|
||||||
self.deactivate_attenuator()
|
|
||||||
|
|
||||||
def set_attenuation_level(
|
def set_attenuation_level(
|
||||||
self,
|
self,
|
||||||
|
Loading…
Reference in New Issue
Block a user