from ctypes import Array, c_double import time from typing import Union from IPython import embed import numpy.typing as npt import uldaq import numpy as np from pyrelacs.util.logging import config_logging log = config_logging() class MccDac: def __init__(self) -> None: devices = uldaq.get_daq_device_inventory(uldaq.InterfaceType.USB) log.debug(f"Found daq devices {len(devices)}, connecting to the first one") if len(devices) == 0: log.error("Did not found daq devices, please connect one") exit(1) self.daq_device = uldaq.DaqDevice(devices[0]) self.daq_device.connect() self.ai_device = self.daq_device.get_ai_device() self.ao_device = self.daq_device.get_ao_device() 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") if len(devices) == 0: log.error("Did not found daq devices, please connect one") exit(1) self.daq_device = uldaq.DaqDevice(devices[0]) self.daq_device.connect() self.ai_device = self.daq_device.get_ai_device() self.ao_device = self.daq_device.get_ao_device() self.dio_device = self.daq_device.get_dio_device() log.debug("Connected") def read_analog( self, channels: list[int], duration: int, samplerate: float, AiInputMode: uldaq.AiInputMode = uldaq.AiInputMode.SINGLE_ENDED, Range: uldaq.Range = uldaq.Range.BIP10VOLTS, 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]: buffer_len_channels = 2 else: buffer_len_channels = 1 buffer_len = np.shape(np.arange(0, duration, 1 / samplerate))[0] data_analog_input = uldaq.create_float_buffer(buffer_len_channels, buffer_len) er = self.ai_device.a_in_scan( channels[0], channels[1], AiInputMode, Range, buffer_len, samplerate, ScanOption, AInScanFlag, data=data_analog_input, ) return data_analog_input def write_analog( self, data: Union[list, npt.NDArray], channels: list[int], samplerate: float, Range: uldaq.Range = uldaq.Range.BIP10VOLTS, ScanOption: uldaq.ScanOption = uldaq.ScanOption.DEFAULTIO, AOutScanFlag: uldaq.AOutScanFlag = uldaq.AOutScanFlag.DEFAULT, ): 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) try: 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() def set_analog_to_zero(self, channels: list[int] = [0, 1]): try: err = self.ao_device.a_out_list( channels[0], channels[1], [ uldaq.Range.BIP10VOLTS, uldaq.Range.BIP10VOLTS, ], uldaq.AOutListFlag.DEFAULT, [0, 0], ) except Exception as e: log.error("f{e}") log.error("disconnection dac") self.disconnect_dac() def diggital_trigger(self, channel: int = 0) -> None: bit_channel = self.read_bit(channel) if not bit_channel: self.write_bit(channel, 1) else: self.write_bit(channel, 0) time.time_ns() 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 ) self.dio_device.d_bit_out( uldaq.DigitalPortType.AUXPORT, bit_number=channel, data=bit ) 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 def read_digitalio( self, channels: list[int], duration, samplerate, ScanOptions: uldaq.ScanOption = uldaq.ScanOption.DEFAULTIO, DInScanFlag: uldaq.DInScanFlag = uldaq.DInScanFlag.DEFAULT, ): if channels[0] == channels[1]: channel_len = 1 else: channel_len = len(channels) buffer_len = np.shape(np.arange(0, duration, 1 / samplerate))[0] data_digital_input = uldaq.create_int_buffer(channel_len, buffer_len) self.dio_device.d_config_port( uldaq.DigitalPortType.AUXPORT, uldaq.DigitalDirection.INPUT ) scan_rate = self.dio_device.d_in_scan( uldaq.DigitalPortType.AUXPORT0, uldaq.DigitalPortType.AUXPORT0, len(data_digital_input), samplerate, ScanOptions, DInScanFlag, data_digital_input, ) return data_digital_input def disconnect_dac(self): self.daq_device.disconnect() self.daq_device.release() def check_attenuator(self): """ ident : attdev-1 strobepin : 6 datainpin : 5 dataoutpin: -1 cspin : 4 mutepin : 7 zcenpin : -1 """ SAMPLERATE = 40_000.0 DURATION = 5 AMPLITUDE = 1 SINFREQ = 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) self.write_analog( data_channels, [0, 1], SAMPLERATE, ScanOption=uldaq.ScanOption.EXTTRIGGER, Range=uldaq.Range.BIP10VOLTS, ) 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") 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)