minipyrelacs/pyrelacs/repros/repos.py

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()
        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 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_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")

        buffer_len = np.shape(np.arange(0, duration, 1 / samplerate))[0]
        data_analog_input = uldaq.create_float_buffer(channel_len, buffer_len)

        er = self.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}")

        err = self.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, self.ao_device

    def set_analog_to_zero(self, channels: list[int] = [0, 1]):
        err = self.ao_device.a_out_list(
            channels[0],
            channels[1],
            [
                uldaq.Range.BIP10VOLTS,
                uldaq.Range.BIP10VOLTS,
            ],
            uldaq.AOutListFlag.DEFAULT,
            [0, 0],
        )

    def digital_trigger(self, channel: int = 0, data: int = 1) -> None:
        log.info(f"{self.daq_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=data
        )

    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 run_repo(self) -> None:
        pass

    def stop_repo(self) -> None:
        pass

    def reload_repo(self) -> None:
        pass