forked from awendt/pyrelacs
143 lines
3.8 KiB
Python
143 lines
3.8 KiB
Python
from ctypes import Array, c_double
|
|
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 Repos:
|
|
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()
|
|
log.debug("Connected")
|
|
|
|
def read_analog_daq(
|
|
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]:
|
|
if channels[0] == channels[1]:
|
|
channel_len = 1
|
|
else:
|
|
channel_len = len(channels)
|
|
assert len(channels) == 2, log.error("Please provide a list with two ints")
|
|
|
|
ai_device = self.daq_device.get_ai_device()
|
|
buffer_len = np.shape(np.arange(0, duration, 1 / samplerate))[0]
|
|
data_analog_input = uldaq.create_float_buffer(channel_len, buffer_len)
|
|
|
|
er = ai_device.a_in_scan(
|
|
channels[0],
|
|
channels[1],
|
|
AiInputMode,
|
|
Range,
|
|
buffer_len,
|
|
samplerate,
|
|
ScanOption,
|
|
AInScanFlag,
|
|
data=data_analog_input,
|
|
)
|
|
# ai_device.scan_wait(uldaq.WaitType.WAIT_UNTIL_DONE, timeout=-1)
|
|
|
|
return data_analog_input
|
|
|
|
def send_analog_dac(
|
|
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,
|
|
):
|
|
"""
|
|
|
|
Parameters
|
|
----------
|
|
data : Union[list, npt.NDArray]
|
|
|
|
channels : list[int]
|
|
|
|
duration : int
|
|
|
|
samplerate : float
|
|
|
|
AiInputMode : uldaq.AiInputMode
|
|
|
|
Range : uldaq.Range
|
|
|
|
ScanOption : uldaq.ScanOption
|
|
|
|
AInScanFlag : uldaq.AOutScanFlag
|
|
|
|
|
|
Returns
|
|
-------
|
|
Array[c_double]
|
|
ao_device
|
|
|
|
|
|
"""
|
|
buffer = c_double * len(data)
|
|
data_analog_output = buffer(*data)
|
|
|
|
log.debug(f"Created C_double data {data_analog_output}")
|
|
ao_device = self.daq_device.get_ao_device()
|
|
ao_info = ao_device.get_info()
|
|
|
|
err = 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, ao_device
|
|
|
|
def digital_trigger(self, portn: int = 0, data: int = 1) -> None:
|
|
log.info(f"{self.daq_device}")
|
|
dio_device = self.daq_device.get_dio_device()
|
|
|
|
dio_device.d_config_bit(
|
|
uldaq.DigitalPortType.AUXPORT, portn, uldaq.DigitalDirection.OUTPUT
|
|
)
|
|
|
|
dio_device.d_bit_out(uldaq.DigitalPortType.AUXPORT, bit_number=portn, data=data)
|
|
|
|
def disconnect_dac(self):
|
|
self.daq_device.disconnect()
|
|
self.daq_device.release()
|
|
|
|
def clean_up():
|
|
pass
|
|
|
|
def run_repo(self) -> None:
|
|
pass
|
|
|
|
def stop_repo(self) -> None:
|
|
pass
|
|
|
|
def reload_repo(self) -> None:
|
|
pass
|