Compare commits

..

No commits in common. "26f43151a27672ad662dce7ca22e0f595d6e2f84" and "b912159b760f77fdb066765fd07f6b138bcc1a68" have entirely different histories.

3 changed files with 112 additions and 44 deletions

View File

@ -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()

View File

@ -19,29 +19,60 @@ 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)]
stim = self.write_analog(
# 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,
[0, 0],
[1, 1],
SAMPLERATE,
ScanOption=uldaq.ScanOption.EXTTRIGGER,
)
data_channel_one = self.read_analog(
[0, 0], DURATION, 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.ai_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)
time.sleep(1)
self.set_analog_to_zero()
@ -59,9 +90,9 @@ class Calibration(MccDac):
if __name__ == "__main__":
SAMPLERATE = 40_000.0
DURATION = 5
AMPLITUDE = 0.5
AMPLITUDE = 1
SINFREQ = 10
cal = Calibration()
# cal.ccheck_attenuator()
cal.check_amplitude()
daq_input = Calibration()
# daq_input.run_calibration()
daq_input.check_attenuator()

View File

@ -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")
@ -80,14 +85,17 @@ 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}")
try:
log.info(self.ao_device)
err = self.ao_device.a_out_scan(
channels[0],
channels[1],
@ -98,12 +106,12 @@ class MccDac:
AOutScanFlag,
data_analog_output,
)
except Exception as e:
print(f"{e}")
self.set_analog_to_zero()
self.disconnect_dac()
return 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()
def set_analog_to_zero(self, channels: list[int] = [0, 1]):
try:
@ -122,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
)
@ -139,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
@ -193,7 +205,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]
@ -207,9 +219,9 @@ class MccDac:
else:
self.set_attenuation_level(db_value, db_value)
_ = self.write_analog(
data,
[0, 0],
stim, ao_device = self.write_analog_dac(
data_channels,
[0, 1],
SAMPLERATE,
ScanOption=uldaq.ScanOption.EXTTRIGGER,
Range=uldaq.Range.BIP10VOLTS,
@ -217,9 +229,7 @@ class MccDac:
self.diggital_trigger()
try:
self.ao_device.scan_wait(uldaq.WaitType.WAIT_UNTIL_DONE, 15)
self.write_bit(channel=0, bit=0)
self.set_analog_to_zero()
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)
@ -234,6 +244,7 @@ class MccDac:
log.info("Sleeping for 1 second, before next attenuation")
time.sleep(1)
self.deactivate_attenuator()
def set_attenuation_level(
self,