Code documentation

Overview

This section provides comprehensive documentation for the LDAQ code, designed to facilitate data acquisition, generation, and visualization across a variety of hardware platforms. The documentation is organized into several key sections, each focusing on a specific aspect of the LDAQ system.

The “Acquisition” section introduces the base acquisition class along with hardware-specific acquisition classes, detailing the methods and properties associated with data acquisition from various devices. These devices range from simulated environments and serial communications to specific hardware integrations like National Instruments, Digilent, FLIR thermal cameras, and Basler cameras.

In the “Generation” section, the focus shifts to data generation. Here, the base generation class and various hardware-specific generation classes are documented, providing insights into how data can be generated and manipulated within the LDAQ framework.

The “Visualization” section delves into the capabilities of LDAQ in terms of data representation and user interface, highlighting the various features of the LDAQ Visualization class.

Under “Core,” the core functionalities of LDAQ are encapsulated, documenting the central operations and workflows that tie together acquisition, generation, and visualization.

The “Utilities” section encompasses additional tools and functions that aid in the operation of LDAQ, including National Instruments utilities and other general-purpose functions for loading and handling measurement data.

Acquisition

Base Acqusition Class

All acquisition classes inherit from the base class LDAQ.acquisition_base.BaseAcquisition. This class defines the basic interface for all acquisition classes. The following methods are defined:

class LDAQ.acquisition_base.BaseAcquisition

Parent acquisition class that should be used when creating new child acquisition source class. Child class should override methods the following methods:

• self.__init__()

• self.set_data_source()

• self.terminate_data_source()

• self.read_data()

• self.clear_buffer() (optional)

• self.get_sample_rate() (optional)

For further information on how to override these methods, see the listed methods docstrings.

Additionally, the __init__() or set_data_source() methods should override or be able to set the following attributes:

• self._channel_names_init - list of original data channels names from source

• self._channel_names_video_init - list of original video channels names from source

• self._channel_shapes_video_init - list of original video channels shapes from source

• self.sample_rate = 0 - sample rate of acquisition source

acquire()

Acquires data from acquisition source and also properly saves the data to pyTrigger ringbuffer. Additionally it also stops the measurement run and terminates acquisition source properly. This method is continuously called in the run_acquisition() method.

activate_trigger(all_sources: bool = True) → None

Sets trigger off. Useful if the acquisition class is trigered by another process. This trigger can also trigger other acquisition sources by setting property class

add_virtual_channel(virtual_channel_name: str, source_channels: int | str | list, function: callable, *args, **kwargs) → None

Add a virtual channel to the acquisition class.

Parameters:

• virtual_channel_name (str) – Name of the channel that will be created

• source_channel (int, str, list) – list of name strings or indices of the channels in self.channel_names_all on which function will be applied. optionally, if only one channel is used, it can be given as a string or index form self.channel_names_all

• function (function) – Function used on the channels. Takes channels’ arrays as input and has to return a numpy array where the first dimension is the number of samples. If returned array has 2 dimensions, it is treated as data source, if it has 3 dimensions, it is treated as video source. The first argument of the function can be a reference to the acquisition class itself. This is useful if the function needs to access some of the acquisition class’ attributes, for example data history.

• *args – additional arguments to be passed to the function (function passed as input argument to this method)

• **kwargs – additional keyword arguments to be passed to the function (function passed as input argument to this method)y

Example 1:

>>> def func(ch1, ch2): # ch1 and ch2 are numpy arrays
>>>     # ch1 and ch2 are of shape (n_samples, 1) and NOT (1, )
>>>     return ch1 + ch2 # returns numpy array of shape (n_samples, 1) or (n_samples, M, K)
>>> acq.add_virtual_channel('ch1+ch2', ['ch1', 'ch2'], func)

Example 2:

>>> def virtual_channel_func(self, ch1):
>>>     try:
>>>         # At virtual channel definition, retrieving data or channel indices is not yet possible for all channels.
>>>         # Therefore, we use a try-except block to properly execute channel definition.
>>>         time, data = self.get_data(N_points=2)
>>>         i_ch3 = self.get_channel_index("ch3", channel_type='data') # retrieve data index channel of ch1
>>>         ch3_prev = data[-1, i_ch3] # retrieve last value of ch1
>>>
>>>     except:
>>>         # at channel definition, no data is available yet. Therefore, we set the previous value to 0.
>>>         ch3_prev = 0
>>>
>>>     # cumulative sum of ch1:
>>>     ch1_cumsum = np.cumsum(ch1) + ch3_prev
>>>     return ch1_cumsum.reshape(-1,1) # reshape to (n_samples, 1)
>>>
>>> acq.add_virtual_channel('ch3', 'ch1', virtual_channel_func)

clear_buffer() → None

EDIT in child class (Optional).

The source buffer should be cleared with this method. It can either clear the buffer, or just read the data with self.read_data() and does not add/save data anywhere. By default, this method will read the data from the source and not add/save data anywhere.

Returns None.

get_channel_index(channel_name: str, channel_type: str = 'data') → int

Returns the index of the channel from either self.channel_names_all, self.channel_names or self.channel_names_video. The channel_type argument is used to specify which list to use. If index is used for retrieving channel data from array returned by self.get_data() then channel_type should depend on which type of data you are recieving.

Parameters:

• channel_name (str) – name of the channel

• channel_type (str) – type of the channel. Can be ‘all’, ‘data’ or ‘video’. Default is ‘data’.

Returns:

index of the channel

Return type:

channel_index (int)

get_data(N_points: int | str | None = None, data_to_return: str = 'data') → tuple

Reads and returns data from the buffer.

Parameters:

• N_points (int, str, None) – number of last N points to read from pyTrigger buffer. if N_points=”new”, then only new points will be retrieved. if None all samples are returned.

• data_to_return (str) – ‘video’, ‘data’ or ‘flattened’. If ‘data’, only data channels are returned as an array

Returns:

(time, data) - np.ndarray 1D time vector and measured data. Data shape depends on ‘data_to_return’ parameter:

• if ‘data’ is requested, returns array with shape (n_samples, n_data_channels).

• if ‘video’ is requested, returns list of 3D arrays with shape (n_samples, height, width).

• if ‘flattened’ is requested, returns flattened data array with shape (n_samples, n_ringbuffer_channels).

Return type:

tuple

IMPORTANT: if N_points = “new” is used for retrieving new results during measurement and Core() class is used for measurement control, then periodic saving should be turned OFF in Core() class. In other words, Core.run() method should be called with save_interval=None.

get_data_PLOT(data_to_return: str = 'data') → ndarray | list

Reads only new data from pyTrigger ring buffer and returns it.

This method is used only for plotting purposes and SHOULD NOT BE USED for any other purpose. Additionally, it does not return time vector, only data. See get_data() method for more details. Data returned in this method is the same as in get_data().

Returns:

• if ‘data’ is requested, returns array with shape (n_samples, n_data_channels).

• if ‘video’ is requested, returns list of 3D arrays with shape (n_samples, height, width).

• if ‘flattened’ is requested, returns flattened data array with shape (n_samples, n_ringbuffer_channels)

Return type:

np.ndarray, list

get_measurement_dict(N_points: int | str = None) → dict

Reads data from pyTrigger ring buffer using self.get_data() method and returns a dictionary containing all relevant information about the measurement.

Parameters:

N_points (None, int, str) – Number of points to get from pyTrigger ringbuffer. If type(N_points)==int then N_points last samples are returned. If N_points==’new’, only new points after trigger event are returned. If None, all samples are returned. Defaults to None.

Returns:

keys and values are the following:

• ’time’: 1D array

• ’channel_names’: list of channel names

• ’data’: 2D array (n_samples, n_data_channels)

• ’channel_names_video’: list of video channel names

• ’video’: list of 3D arrays (n_samples, height, width)

• ’sample_rate’: sample rate of acquisition

Return type:

dict

IMPORTANT: If N_points = “new” is used for retrieving new results during measurement and Core() class is used for measurement control, then periodic saving should be turned OFF in Core() class. In other words, Core.run() method should be called with save_interval=None.

get_sample_rate() → float

EDIT in child class (Optional).

Returns sample rate of acquisition class.

is_triggered() → bool

Checks if acquisition class has been triggered during measurement.

Returns:

True/False if triggered

Return type:

bool

read_data() → ndarray

EDIT in child class.

This method only reads data from the source and transforms data into standard format used by other methods. It is called within self.acquire() method which properly handles acquiring data and saves it into pyTrigger ring buffer.

Must ALWAYS return a 2D numpy array of shape (n_samples, n_columns).

IMPORTANT: If some of the channels are videos (2D array - so shape is (n_samples, n_pixels_width, n_pixels_height)), then data has to be reshaped to shape (n_samples, n_pixels_width*n_pixels_height). Then data from multiple sources have to be concatenated into one array of shape (n_samples, n_cols), where cols is the combined number of pixel of all video sources and number of channels in data sources.

For an example where data source has 2 video sources with resolution 300x200 and 2 data channels, the final shape returned by this methods should be (n_samples, 300*200*2+2).

For video sources, the shape of the video is automatically stored in self.channel_shapes_video_init when self.set_data_source() is called. When data is retrieved from the source, it is reshaped to (n_samples, n_pixels_width, n_pixels_height).

Returns:

2D numpy array of shape (n_samples, n_columns)

Return type:

data (np.ndarray)

reset_trigger(all_sources: bool = True) → None

Resets trigger.

Parameters:

all_sources (bool) – if True, resets trigger of all acquisition sources. If False, resets only this acquisition source. Currently, this parameter is not used, it is always set to True.

run_acquisition(run_time: float = None, run_in_background: bool = False) → None

Runs acquisition. This method is used to start the acquisition.

Parameters:

• run_time (float) – number of seconds for which the acquisition will run.

• run_in_background (bool) – if True, acquisition will run in a separate thread.

Returns:

None

save(name: str, root: str = '', timestamp: bool = True, comment: str = None) → None

Save acquired data.

Parameters:

• name (str) – filename

• root (str, optional) – directory to save to. Defaults to ‘’.

• timestamp (bool, optional) – include timestamp before ‘filename’. Defaults to True.

• comment (str, optional) – commentary on the saved file. Defaults to None.

Returns:

None

set_continuous_mode(boolean: bool = True, measurement_duration: float = None) → None

Sets continuous mode of the acquisition.

If True, acquisition will run indefinitely until externally stopped. If False, acquisition will run for a specified time.

Parameters:

• boolean (bool, optional) – Defaults to True.

• measurement_duration (float, optional) – If not None, sets the duration of the measurement in seconds. It does NOT update ring buffer size. Defaults to None. Based on measurement_duration, the number of total samples to be acquired is calculated. In this case the ring buffer size can be different to the number of samples to be acquired. If None, measurement duration measurement will not stop until externally stopped. This means that after the ring buffer is filled, the oldest data will be overwritten by the newest data.

Returns:

None

Examples

>>> # Setting continuous mode to True, setting buffer size to 10 seconds of data, measurement will run indefinitely:
>>> acq.set_trigger(level=0, channel=0, duration=10.) # this will trigger measurement right away, buffer size will be 10 seconds
>>> acq.set_continuous_mode(True, measurement_duration=None) # acquisition will run indefinitely until externally stopped

>>> # Setting continuous mode to True, setting buffer size to 5 seconds of data, measurment will run for 10 seconds:
>>> acq.set_trigger(level=0, channel=0, duration=5.) # this will trigger measurement right away, buffer size will be 5 seconds
>>> acq.set_continuous_mode(True, measurement_duration=10.) # acquisition will run for 10 seconds, but buffer will store only 5 seconds of data

set_data_source() → None

EDIT in child class.

Properly sets acquisition source before measurement is started. Requirements for this method:

• Should call super().set_data_source() AT THE END of the method.

• Should be set up in a way that it is able to be called multiple times in a row without issues.

• Should set up self._channel_names_init and self._channel_names_video_init if not set in __init__() method.

VIDEO source only:

• Should set self._channel_shapes_video_init which is a list of tuples with shapes of each video channel that will be recieved from acquisition source. This is required for proper operation of the class.

• the order of the shapes in self._channel_shapes_video_init should be the same as the order of the channels in self._channel_names_video_init.

set_trigger(level: float, channel: str | int, duration: float | int = 1, duration_unit: str = 'seconds', presamples: int = 0, type: str = 'abs') → None

Set parameters for triggering the measurement.

Only one trigger channel is supported at the moment. Additionally trigger can only be set on ‘data’ channels. If trigger is needed on ‘video’ channels, a ‘data’ virtual channel has to be created using ‘add_virtual_channel()’ method, and then trigger can be set on this virtual channel.

Parameters:

• level (float) – trigger level

• channel (int, str) – trigger channel (int or str). If str, it must be one of the channel names. If int, index from self.channel_names (‘data’ channels) has to be provided (NOTE: see the difference between self.channel_names and self.channel_names_all).

• duration (float, int, optional) – duration of the acquisition after trigger (in seconds or samples). Defaults to 1.

• duration_unit (str, optional) – ‘seconds’ or ‘samples’. Defaults to ‘seconds’.

• presamples (int, optional) – number of presamples to save. Defaults to 0.

• type (str, optional) – trigger type: up, down or abs. Defaults to ‘abs’.

Returns:

None

stop() → None

Stops acquisition run.

terminate_data_source() → None

EDIT in child class.

Properly closes/disconnects acquisition source after the measurement. The method should be able to handle mutliple calls in a row.

update_trigger_parameters(**kwargs) → None

Updates trigger settings. See ‘set_trigger()’ method for possible parameters.

Hardware-Specific Acqusition Classes

class LDAQ.simulator.SimulatedAcquisition(acquisition_name=None, multi_processing=False)

Simulated acquisition class that can be used when no source is present.

clear_buffer()

Clears serial buffer.

static data_generator_multiprocessing(connection, stop_event, sample_rate, ser_function, fun_args)

This function runs in a separate process and generates data (2D numpy arrays), and maintains a buffer of generated data.

data_generator_threading(stop_event, sample_rate, fun_or_arr, fun_args)

This function runs in a separate process and generates data (2D numpy arrays), and maintains a buffer of generated data.

read_data()

Reads data from simulated data source.

Returns:

data from serial port with shape (n_samples, n_channels).

Return type:

np.ndarray

set_data_source(initiate_data_source=True)

Initializes simulated data source

set_simulated_data(fun_or_array, channel_names=None, sample_rate=None, args=())

sets simulated data to be returned by read_data() method. This should also update self._channel_names_init list.

NOTE: The function ‘fun’ should also include all library imports needed for its execution if multiprocessing=True. This is due to serialization limitations of the function of ‘dill’ library in order to be able to pass the function to the child process. For example, if the function ‘fun’ uses numpy, it should be imported.

Parameters:

• fun_or_array (function, np.ndarray) – function that returns numpy array with shape (n_samples, n_channels) or numpy array with shape (n_samples, n_channels) that will be repeated in a loop.

• channel_names (list, optional) – list of channel names. Defaults to None, in which case the names “channel_0”, “channel_1”, … are used.

• sample_rate (int, optional) – sample rate of the simulated data. Defaults to None, in which case the sample rate of 1000 Hz is used.

• args (tuple, optional) – arguments for the function. Defaults to ().

Example:

>>> def simulate_signal(t, f1, f2):
>>>     import numpy as np
>>>     sig1 = np.sin(2*np.pi*f1*t) + np.random.rand(len(t))*0.3
>>>     sig2 = np.cos(2*np.pi*f2*t) + np.random.rand(len(t))*0.3
>>>     return np.array([sig1, sig2]).T
>>>
>>> acq_simulated = LDAQ.simulator.SimulatedAcquisition(acquisition_name='sim')
>>> acq_simulated.set_simulated_data(simulate_signal, channel_names=["ch1", "ch2"], sample_rate=100000, args=(84, 120))
>>> acq_simulated.run_acquisition(5.0)

set_simulated_video(fun_or_array, channel_name_video=None, sample_rate=None, args=())

sets simulated video to be returned by read_data() method. This should also update self._channel_names_video_init and self._channel_shapes_video_init lists.

NOTE: if simulator acqusition is running with multi_processing=True, The function ‘fun’ should also include all library imports needed for its execution.

This is due to serialization limitations of the function of ‘dill’ library in order to be able to pass the function to the child process. For example, if the function ‘fun’ uses numpy, it should be imported.

Parameters:

• fun_or_array (function, np.ndarray) – function that returns numpy array with shape (n_samples, width, height) or numpy array with shape (n_samples, width, height) that will be repeated in a loop.

• channel_name_video (str, optional) – name of the video channel. Defaults to None, in which case the name “video” is used.

• sample_rate (int, optional) – sample rate of the simulated data. Defaults to None, in which case the sample rate of 30 Hz is used.

• args (tuple, optional) – arguments for the function. Defaults to ().

terminate_data_source()

Terminates simulated data source

class LDAQ.national_instruments.NIAcquisition(task_name: str | object, acquisition_name: str | None = None)

National Instruments Acquisition class, compatible with any NI acquisition device that is supported by NI-DAQmx library.

To use this class, you need to install NI-DAQmx library found on this link: https://www.ni.com/en/support/downloads/drivers/download.ni-daq-mx.html#494676

Installation instructions:

• Download NI-DAQmx from the link listed above.

• Install NI-DAQmx.

clear_buffer()

Clears the buffer of the device.

clear_task()

Clear a task.

read_data()

Reads data from device buffer and returns it.

Returns:

numpy array with shape (n_samples, n_channels)

Return type:

np.ndarray

run_acquisition(run_time=None, run_in_background=False)

Runs acquisition. This is the method one should call to start the acquisition.

Parameters:

• run_time (float) – number of seconds for which the acquisition will run.

• run_in_background (bool) – if True, acquisition will run in a separate thread.

Returns:

None

set_data_source()

Sets the acquisition device to properly start the acquisition. This function is called before the acquisition is started. It is used to properly initialize the device and set the data source channels and virtual channels.

terminate_data_source()

Properly closes the data source.

class LDAQ.serial_communication.SerialAcquisition(port, baudrate, byte_sequence, timeout=1, start_bytes=b'', end_bytes=b'\n', write_start_bytes=None, write_end_bytes=None, pretest_time=None, sample_rate=None, channel_names=None, acquisition_name=None)

General Purpose Class for Serial Communication.

clear_buffer()

Clears serial buffer.

get_sample_rate(run_pretest=False)

Returns acquisition sample rate or estimates sample rate if run_pretest is True.

Parameters:

run_pretest (bool, optional) – If True, then runs pretest to estimate sample rate. Defaults to False.

Returns:

estimated sample rate

Return type:

float

read_data()

reads data from serial port and returns it as numpy array with shape (n_samples, n_channels).

Returns:

data from serial port with shape (n_samples, n_channels).

Return type:

np.ndarray

set_channel_names()

Sets default channel names if none were passed to the class.

set_data_source(write_start_bytes=False)

Initializes serial connection, sets channels and virtual channels.

Parameters:

• write_start_bytes (bool, optional) – If True, then writes write_start_bytes to serial port

• False. ((passed in init method). Defaults to) –

set_unpack_data_settings()

Converts byte_sequence to string passed to struct unpack method.

terminate_data_source()

Closes serial connection.

write_to_serial(write_bytes, delay_ms=10)

Writes data to serial port.

Parameters:

• write_bytes (list, tuple, bytes, bytearray) – bytes to be written to serial port. If list/tuple, then elements have to be of type byte/bytearray. Writes each encoded bstring with ‘delay_ms’ delay.

• delay_ms (int, optional) – Delay between writing bytes. Defaults to 10 ms.

class LDAQ.digilent.WaveFormsAcquisition(acquisition_name=None, channels=[0, 1], sample_rate=10000, channel_names=None, device_number=None)

This is a class for acquiring data from Digilent Analog Discovery 2, using WaveForms SDK.

To use this class, you need to install WaveForms found on this link: https://digilent.com/shop/software/digilent-waveforms/

Installation instructions:

• Download WaveForms from the link listed above.

• Install WaveForms.

clear_buffer()

The source buffer should be cleared with this method. Either actually clears the buffer, or just reads the data with self.read_data() and does not add/save data anywhere.

Returns None.

configure_channels(input_range=None)

Specify min and max value range for each channel.

Parameters:

input_range (dict) – dictionary with channel index as key and tuple of min and max values as value. channel indices have to be the same as self.channel_idx (or channels input parameters in init) For example: {0:(-10, 10), 1:(-5, 5)} -> channel 0 has range -10 to 10 V and channel 1 has range -5 to 5 V. By default, all channels have range -10 to 10 V.

get_sample_rate()

Returns sample rate of acquisition class. This function is also useful to compute sample_rate estimation if no sample rate is given

Returns self.sample_rate

read_data()

This method only reads data from the source and transforms data into standard format used by other methods. This method is called within self.acquire() method which properly handles acquiring data and saves it into pyTrigger ring buffer.

Must return a 2D numpy array of shape (n_samples, n_columns).

set_data_source()

Properly sets acquisition source before measurement is started. Should be set up in a way that it is able to be called multiple times in a row without issues.

terminate_data_source()

Properly closes acquisition source after the measurement.

Returns None.

class LDAQ.flir.FLIRThermalCamera(acquisition_name=None, channel_name_IR_cam='temperature_field', IRtype='LINEAR_10MK')

Acquisition class for FLIR thermal camera (A50).

This class is adapted from examples for thermal A50 camera provided by FLIR, found on their website (login required): https://www.flir.com/support-center/iis/machine-vision/downloads/spinnaker-sdk-download/spinnaker-sdk–download-files/

Installation steps:

• Install Spinnaker SDK (i.e. SpinnakerSDK_FULL_3.1.0.79_x64.exe, found on provided link)

• Install PySpin (python wrapper for Spinnaker SDK). On the website listed above, there are multiple build wheels listed under “Lastest Windows Python Spinnaker SDK”. Choose the one that matches your python version and architecture (i.e. spinnaker_python-3.1.0.79-cp310-cp310-win_amd64.zip for python 3.10 64-bit - this is also the version used for development of this class)

clear_buffer()

The source buffer should be cleared with this method. Either actually clears the buffer, or just reads the data with self.read_data() and does not add/save data anywhere.

Returns None.

get_sample_rate()

Returns sample rate of acquisition class. This function is also useful to compute sample_rate estimation if no sample rate is given

Returns self.sample_rate

read_data()

This method only reads data from the source and transforms data into standard format used by other methods. This method is called within self.acquire() method which properly handles acquiring data and saves it into pyTrigger ring buffer.

Must return a 2D numpy array of shape (n_samples, n_columns).

set_IRtype(IRtype)

This function sest the IR type to be used by the camera.

Sets the IR type to either:

• LINEAR_10MK: 10mK temperature resolution

• LINEAR_100MK: 100mK temperature resolution

• RADIOMETRIC: capture radiometric data and manually convert to temperature (currently not recommended)

Parameters:

IRtype (str) – IR type to be used by the camera (either LINEAR_10MK, LINEAR_100MK or RADIOMETRIC)

set_data_source()

Properly sets acquisition source before measurement is started. Should be set up in a way that it is able to be called multiple times in a row without issues.

terminate_data_source()

Properly closes acquisition source after the measurement.

Returns None.

class LDAQ.basler.BaslerCamera(acquisition_name=None, sample_rate=10, channel_name_camera='Camera', offset=(0, 0), size=(3000, 2000), subsample=1, pixel_format='Mono12', exposure_time_ms=4.0)

Acquisition class for Basler camera using pypylon library.

Link to required programs: https://www.baslerweb.com/en/downloads/software-downloads/#type=pylonsoftware;language=all;version=7.3.0

Installation steps:

• Download and install pylon 7.3.0 Camera Software Suite Windows software and choose developer option during installation

• Install python library with pip install pypylon

clear_buffer()

The source buffer should be cleared with this method. Either actually clears the buffer, or just reads the data with self.read_data() and does not add/save data anywhere.

Returns None.

get_sample_rate()

Returns sample rate of acquisition class. This function is also useful to compute sample_rate estimation if no sample rate is given

Returns self.sample_rate

read_data()

This method only reads data from the source and transforms data into standard format used by other methods. This method is called within self.acquire() method which properly handles acquiring data and saves it into pyTrigger ring buffer.

Must return a 2D numpy array of shape (n_samples, n_columns).

set_data_source(start_grabbing=True)

Properly sets acquisition source before measurement is started. Should be set up in a way that it is able to be called multiple times in a row without issues.

terminate_data_source()

Properly closes acquisition source after the measurement.

Returns None.

Generation

Base Generation Class

All generation classes inherit from the base class LDAQ.generation_base.BaseGeneration. This class defines the basic interface for all generation classes. The following methods are defined:

class LDAQ.generation_base.BaseGeneration

Base class for signal generation. Used for generating signals.

add_delay(delay)

Adds delay before generation starts

Parameters:

delay (float) – Delay in seconds

generate()

EDIT in child class. The child should call methods that generate the signal.

run_generation(delay=None, block=False)

Runs generation. If block is True, the generation will block the main thread until generation is stopped.

Parameters:

• delay (float, optional) – Delay in seconds before generation starts. If None, no delay is added or previous delay is used. Defaults to None.

• block (bool, optional) – If True, the generation will block the main thread until generation is stopped. Defaults to False.

set_data_source()

EDIT in child class. The child should call methods that set the signal.

stop()

Stops the generation.

terminate_data_source()

EDIT in child class. The child should call methods that terminates and exits signal generation device correctly.

Hardware-Specific Generation Classes

class LDAQ.national_instruments.NIGeneration(task_name, signal=None, generation_name=None)

clear_task()

Clears NI output task.

generate()

Generates the signal.

set_data_source(initiate=True)

Sets the data source for the generation.

Parameters:

initiate (bool, optional) – intitiate NI task. Defaults to True.

set_generation_signal(signal)

sets signal that will be generated, and repeated in a loop.

Parameters:

signal (np.ndarray) – numpy array with shape (n_samples, n_channels) or (n_samples,).

terminate_data_source()

Terminates the data source for the generation.

Visualization

class LDAQ.Visualization(refresh_rate: int = 100, max_points_to_refresh: int = 10000, sequential_plot_updates: bool = False)

add_image(source: str, channel: str | int, function: str | callable | None = None, refresh_rate: int = 100, colormap: str = 'CET-L17') → None

Add an image plot to the visualization for the specified source and channel.

Parameters:

• source (str) – The name of the source to add the image plot to.

• channel (str/int) – The name of the channel to add the image plot to.

• function (function/str, optional) – A function or string to apply to the image data before plotting. Defaults to None.

• refresh_rate (int, optional) – The number of milliseconds between updates of the plot. Defaults to 100.

• colormap (str, optional) – The colormap to use for the plot. Defaults to ‘CET-L17’ (suitable for thermal images), for grayscale use ‘gray’. Any CET colormap can be used.

This method adds an image plot to the visualization for the specified source and channel. The function argument can be used to apply a custom function to the image data before plotting. If function is not specified or is not a callable function or a string, the identity function is used. If function is a string, it is looked up in the INBUILT_FUNCTIONS dictionary. The refresh_rate argument specifies the number of milliseconds between updates of the plot. The colormap argument specifies the colormap to use for the plot.

If source is not already in self.plots, a new entry is created for it. If channel is not already in the entry for source in self.plots, a new plot is created for it.

This method modifies the plots and color_map attributes of the Visualization object in-place.

add_lines(position: Tuple[int, int], source: str, channels: int | str | tuple | list, function: callable | str | None = None, nth: int | str = 'auto', refresh_rate: int | None = None, t_span: int | float | None = None) → None

Build the layout dictionary.

Parameters:

• position (tuple) – The position of the subplot. Example: (0, 0).

• source (str) – The source of the data. Name that was given to the Acquisition object.

• channels (int/str/tuple/list) – The channels from the source to be plotted. Can also be a list of tuples of integers to plot channel vs. channel. Example: [(0, 1), (2, 3)]. For more details, see example below and documentation.

• function (function/str, optional) – The function to be applied to the data before plotting. If channels is a list of tuples, the function is applied to each tuple separately. Defaults to None.

• nth (int/str, optional) – The nth sample to be plotted. If nth is "auto", the nth sample is computed automatically. Defaults to "auto".

• refresh_rate (int, optional) – The refresh rate of the subplot in milliseconds. If this argument is not specified, the refresh rate defined in the Visualization is used. Defaults to None.

• t_span (int/float, optional) – The length of the time axis. If this option is not specified, it is computed from the xlim. Defaults to None.

Channels

If the channels argument is an integer, the data from the channel with the specified index will be plotted.

If the channels argument is a list of integers, the data from the channels with the specified indices will be plotted:

>>> vis.add_lines(position=(0, 0), source='DataSource', channels=[0, 1])

To plot channel vs. channel the channels argument is a tuple of two integers:

>>> vis.add_lines(position=(0, 0), source='DataSource', channels=(0, 1))

The first integer is the index of the x-axis and the second integer is the index of the y-axis.

Multiple channel vs. channel plots can be added to the same subplot:

>>> vis.add_lines(position=(0, 0), source='DataSource', channels=[(0, 1), (2, 3)])

The function argument

The data can be processed on-the-fly by a specified function.

The function can be specified by the user. To use the built-in functions, a string is passed to the function argument. An example of a built-in function is “fft” which computes the [Fast Fourier Transform](https://numpy.org/doc/stable/reference/generated/numpy.fft.rfft.html) of the data with indices 0 and 1:

>>> vis.add_lines(position=(0, 0), source='DataSource', channels=[0, 1], function='fft')

To build a custom function, the function must be defined as follows:

>>> def function(self, channel_data):
        '''
        Args:
            self: instance of the acquisition object (has to be there so the function is called properly)
            channel_data (dict): A dictionary containing the channel data.
        '''
        return channel_data**2

The self argument in the custom function referes to the instance of the acquisition object. This connection can be used to access the properties of the acquisition object, e.g. sample rate. The channel_data argument is a list of numpy arrays, where each array corresponds to the data from one channel. The data is acquired in the order specified in the channels argument.

For the example above, the custom function is called for each channel separetely, the channel_data is a one-dimensional numpy array. To add mutiple channels to the channel_data argument, the channels argument is modified as follows:

>>> vis.add_lines(position=(0, 0), source='DataSource', channels=[(0, 1)], function=function)

The function is now passed the channel_data with shape (N, 2) where N is the number of samples. The function can also return a 2D numpy array with shape (N, 2) where the first column is the x-axis and the second column is the y-axis. An example of such a function is:

>>> def function(self, channel_data):
        '''
        Args:
            self: instance of the acquisition object (has to be there so the function is called properly)
            channel_data (np.ndarray): A 2D channel data array of size (N, 2).
        Returns:
            np.ndarray: A 2D array np.array([x, y]).T that will be plotted on the subplot.
        '''
        ch0, ch1 = channel_data.T
        x =  np.arange(len(ch1)) / self.acquisition.sample_rate # time array
        y = ch1**2 + ch0 - 10
        return np.array([x, y]).T

config_subplot(position: Tuple[int, int], xlim: Tuple[float, float] | None = None, ylim: Tuple[float, float] | None = None, t_span: float | None = None, axis_style: str | None = 'linear', title: str | None = None, rowspan: int = 1, colspan: int = 1) → None

Configure a subplot at position position.

Parameters:

• position (tuple) – Tuple of two integers, the position of the subplot in the layout.

• xlim (tuple, optional) – Tuple of two floats, the limits of the x-axis. If not given, the limits are set to (0, 1).

• ylim (tuple, optional) – Tuple of two floats, the limits of the y-axis. Defaults to None.

• t_span (int/float, optional) – The length of the time axis. If this option is not specified, it is computed from the xlim. Defaults to None.

• axis_style (str, optional) – The style of the axis. Can be “linear”, “semilogx”, “semilogy” or “loglog”. Defaults to “linear”.

• title (str, optional) – The title of the subplot. Defaults to None.

• rowspan (int, optional) – The number of rows the subplot spans. Defaults to 1.

• colspan (int, optional) – The number of columns the subplot spans. Defaults to 1.

This method configures a subplot at position position with the specified options. The xlim, ylim, t_span, axis_style, title, rowspan and colspan options are stored in the subplot_options dictionary of the Visualization object.

create_ring_buffers()

Create and initialize the ring buffers for all plots in self.plots.

For each source in self.plots, this method creates a RingBuffer2D object with the appropriate number of rows and channels, based on the t_span and sample_rate options in self.plots and the corresponding acquisition. If the acquisition has video channels, this method also initializes a list of random images for each video channel. If a source does not have any channels, a RingBuffer2D object with one row and one channel is created.

This method modifies the ring_buffers and new_images attributes of the Visualization object in-place.

Core

class LDAQ.core.Core(acquisitions, generations=None, controls=None, visualization=None)

add_check_events(*args)

Takes functions as input arguments that take only “self” argument and returns True/False. If any of the provided functions is True, the acquisition will be stopped. This can be used to add additional functionalities to the measurement, such as stopping the measurement if a certain condition is met.

Each time this function is called, the previous additional check functions are erased.

Example

>>> def check_temperature(self):
>>>     acq = self.acquisitions[0] # take 1st acquisition source
>>>     temperature = acq.get_measurement_dict()['data'][:, 0] # take 1st channel
>>>     # stop the measurement if temperature is above 50 degrees:
>>>     if temperature > 50:
>>>         return True
>>>     else:
>>>         return False

get_measurement_dict(N_seconds=None)

Returns measured data from all sources.

Parameters:

N_seconds (float, str, optional) – last number of seconds of the measurement. if “new” then only new data is returned. Defaults to None. When Core() class is run with run() method and periodic saving, N_seconds=”new” should not be used as it will cause data loss.

Returns:

Measurement dictionary.

• 1st level keys: Acquisition names.

• Values: Acquisition dictionaries with the following keys and types:

  • time: 1D array

  • channel_names: Corresponding to self.channel_names

  • data: 2D array with shape (n_samples, n_data_channels)

  • channel_names_video: Corresponding to self.channel_names_video

  • video: List of 3D arrays with shape (n_samples, height, width)

  • sample_rate: Corresponding to self.sample_rate

Return type:

dict

run(measurement_duration=None, autoclose=True, autostart=False, save_interval=None, run_name='Run', root='', save_channels=None, verbose=2)

Runs the measurement with acquisition and generation sources that have already been set. This entails setting configuration and making acquiring and generation threads for each source, as well as starting and stopping them when needed. If visualization has been defined, it will run in a separate thread.

Parameters:

• measurement_duration (float) – measurement duration in seconds, from trigger event of any of the sources. If None, the measurement runs for the duration specified with set_trigger() method and the measurement_duration is None, measurement will take place for duration specified in set_trigger(). Default is None.

• autoclose (bool) – whether the visualization should close automatically or not. Default is True.

• autostart (bool) – whether the measurement should start automatically or not. If True, start as soon as all the acquisition sources are ready. This is not recommended when measuring with different acquisition sources, as the delay between the sources is generally increased. Defaults to False.

• save_interval (float) – data is saved every ‘save_periodically’ seconds. Defaults to None, meaning data is not saved.

• run_name (str) – name of the run. This name is used to save measurements when periodic saving is turned on. Default is “Run”.

• root (str) – root directory where measurements are saved. Default is empty string.

• save_channels (list) – list of channels (strings) to save when periodic saving is turned on (save_interval != None). For each acquisition source a check if specified channels exists will be performed. If None, all channels are saved. Default is None.

• verbose (int) – 0 (print nothing), 1 (print status) or 2 (print status and hotkey legend, if used without visualization). Default is 2.

Returns:

None

save_measurement(name=None, root=None, timestamp=True, comment=None, save_channels=None)

Save acquired data from all sources into one dictionary saved as pickle. See get_measurement_dict() method for the

structure of the dictionary.

Parameters:

• name (str, optional) – filename, if None filename defaults to run name specified in run() method. Defaults to None.

• root (str, optional) – directory to save to. Defaults to None.

• timestamp (bool, optional) – include timestamp before ‘filename’. Defaults to True.

• comment (str, optional) – comment on the saved file. Defaults to None.

• save_channels (list) – list of channels (strings) to save when periodic saving is turned on (save_interval != None). For each acquisition source a check if specified channels exists will be performed. If None, all channels are saved. Default is None.

Returns:

path to the saved file

Return type:

str

set_trigger(source, channel, level, duration, duration_unit='seconds', presamples=0, trigger_type='abs')

Set parameters for triggering the measurement.

Parameters:

• source (int, str) – Index (position in the ‘self.acquisitions’ list) or name of the acquisition source as a string (‘acquisition.acquisition_name’) for which trigger is to be set.

• channel (int, str) – trigger channel (int or str). If str, it must be one of the channel names. If int, index from self.channel_names (‘data’ channels) has to be provided (NOTE: see the difference between self.channel_names and self.channel_names_all).

• level (float) – trigger level

• duration (float, int, optional) – duration of the acquisition after trigger (in seconds or samples). Defaults to 1.

• duration_unit (str, optional) – ‘seconds’ or ‘samples’. Defaults to ‘seconds’.

• presamples (int, optional) – number of presamples to save. Defaults to 0.

• type (str, optional) – trigger type: up, down or abs. Defaults to ‘abs’.

Returns:

None

NOTE: only one trigger channel is supported at the moment. Additionally trigger can only be set on ‘data’ channels. If trigger is needed on ‘video’ channels, a ‘data’ virtual channel has to be created using ‘add_virtual_channel()’ method, and then trigger can be set on this virtual channel.

start_acquisition()

Starts acquisitions sources.

stop_acquisition_and_generation()

Stops all acquisition and generation sources.

Utilities

National Instruments

class LDAQ.national_instruments.NITask(task_name: str, sample_rate: float, settings_file: str | None = None)

acquire(wait_4_all_samples=False)

Acquires the data from the task.

add_channel(channel_name: str, device_ind: int, channel_ind: int, sensitivity: float | None = None, sensitivity_units: str | None = None, units: str | None = None, serial_nr: str | None = None, scale: float | None = None, min_val: float | None = None, max_val: float | None = None) → None

Add a channel to the task. The channel is not actually added to the task until the task is initiated.

Parameters:

• channel_name – name of the channel.

• device_ind – index of the device. To see all devices, see self.device_list attribute.

• channel_ind – index of the channel on the device.

• sensitivity – sensitivity of the sensor.

• sensitivity_units – units of the sensitivity.

• units – output units.

• serial_nr – serial number of the sensor. If specified, the sensitivity, sensitivity_units and units are read from the settings file.

• scale – scale the signal. If specified, the sensitivity, sensitivity_units are ignored. The prescaled units are assumed to be Volts, the scaled units are assumed to be units. The scale can be float or a tuple. If float, this is the slope of the linear scale and y-interception is at 0. If tuple, the first element is the slope and the second element is the y-interception.

• min_val – minimum value of the signal. If None, the default value is used.

• max_val – maximum value of the signal. If None, the default value is used.

initiate(start_task: bool = True) → None

Initiate the task.

Parameters:

start_task – start the task after initiating it.

save(clear_task: bool = True) → None

Save the task to the system (NI MAX).

If the task is not initiated yet, it will be initiated.

Parameters:

clear_task – Whether to clear the task after saving. Defaults to True.

class LDAQ.national_instruments.NITaskOutput(task_name: str, sample_rate: float, samples_per_channel: int | None = None)

add_channel(channel_name: str, device_ind: int, channel_ind: int, min_val: float = -10.0, max_val: float = 10.0) → None

Add a channel to the task.

Parameters:

• channel_name – Name of the channel.

• device_ind – Index of the device. To see all devices, see self.device_list attribute.

• channel_ind – Index of the channel on the device.

• min_val – Minimum value of the channel. Defaults to -10.

• max_val – Maximum value of the channel. Defaults to 10.

initiate(start_task: bool = True) → None

Initiate the task.

Parameters:

start_task – Whether to start the task after initiating it. Defaults to True.

Other

LDAQ.utils.load_measurement(name: str, directory: str = '')

Loads a measurement from a pickle file.

Parameters:

• name (str) – name of the measurement file

• directory (str) – directory where the measurement file is located

Returns:

dictionary containing the measurement

Return type:

measurement (dict)

LDAQ.load_measurement_multiple_files(directory: str = None, contains: str = '')

Loads all measurements with name matching the string and combines them into one measurement dictionary.

Parameters:

• directory (str) – directory where the measurement files are located

• contains (str) – string that the measurement files should contain in their name

Returns:

dictionary containing concatenated measurement datra from multiple files.

Return type:

measurement (dict)