From 541e6c63f3b2c9dbb2d851362a30f2a77f981d58 Mon Sep 17 00:00:00 2001 From: Zach Hindes Date: Tue, 5 Mar 2024 17:26:24 -0600 Subject: [PATCH 1/3] Add Digital Stream Writer tests --- tests/component/test_stream_writers_do.py | 618 ++++++++++++++++++++++ 1 file changed, 618 insertions(+) create mode 100644 tests/component/test_stream_writers_do.py diff --git a/tests/component/test_stream_writers_do.py b/tests/component/test_stream_writers_do.py new file mode 100644 index 00000000..1affa9fe --- /dev/null +++ b/tests/component/test_stream_writers_do.py @@ -0,0 +1,618 @@ +import ctypes +import math +from typing import Callable, List + +import numpy +import pytest + +import nidaqmx +from nidaqmx.constants import LineGrouping +from nidaqmx.stream_writers import DigitalMultiChannelWriter, DigitalSingleChannelWriter +from nidaqmx.utils import flatten_channel_string + + +def _start_do_task(task: nidaqmx.Task, is_port: bool = False, num_chans: int = 1) -> None: + # we'll be doing simple on-demand, so start the task... + task.start() + + # ... and set the output to a known initial value + if is_port: + if num_chans == 8: + task.write(0) + else: + task.write([0 for _ in range(num_chans)]) + else: + if num_chans == 1: + task.write(False) + else: + task.write([False for _ in range(num_chans)]) + + +@pytest.fixture +def do_single_line_task( + generate_task: Callable[[], nidaqmx.Task], real_x_series_device: nidaqmx.system.Device +) -> nidaqmx.Task: + task = generate_task() + task.do_channels.add_do_chan( + real_x_series_device.do_lines[0].name, line_grouping=LineGrouping.CHAN_FOR_ALL_LINES + ) + _start_do_task(task) + return task + + +@pytest.fixture +def do_single_channel_multi_line_task( + generate_task: Callable[[], nidaqmx.Task], real_x_series_device: nidaqmx.system.Device +) -> nidaqmx.Task: + task = generate_task() + chan = task.do_channels.add_do_chan( + flatten_channel_string(real_x_series_device.do_lines[:8].name), + line_grouping=LineGrouping.CHAN_FOR_ALL_LINES, + ) + _start_do_task(task, num_chans=chan.do_num_lines) + return task + + +@pytest.fixture +def do_multi_channel_multi_line_task( + generate_task: Callable[[], nidaqmx.Task], real_x_series_device: nidaqmx.system.Device +) -> nidaqmx.Task: + task = generate_task() + task.do_channels.add_do_chan( + flatten_channel_string(real_x_series_device.do_lines[:8].name), + line_grouping=LineGrouping.CHAN_PER_LINE, + ) + _start_do_task(task, num_chans=task.number_of_channels) + return task + + +@pytest.fixture +def do_port0_task( + generate_task: Callable[[], nidaqmx.Task], real_x_series_device: nidaqmx.system.Device +) -> nidaqmx.Task: + task = generate_task() + # X Series port 0 has either 32 or 8 lines. The former can only be used with 32-bit writes. The + # latter can be used with any sized port write. + task.do_channels.add_do_chan( + real_x_series_device.do_ports[0].name, + line_grouping=LineGrouping.CHAN_FOR_ALL_LINES, + ) + _start_do_task(task, is_port=True) + return task + + +@pytest.fixture +def do_port1_task( + generate_task: Callable[[], nidaqmx.Task], real_x_series_device: nidaqmx.system.Device +) -> nidaqmx.Task: + task = generate_task() + # X Series port 1 has 8 lines, and can be used with any sized port write. + task.do_channels.add_do_chan( + real_x_series_device.do_ports[1].name, + line_grouping=LineGrouping.CHAN_FOR_ALL_LINES, + ) + _start_do_task(task, is_port=True) + return task + + +@pytest.fixture +def do_multi_channel_port_task( + generate_task: Callable[[], nidaqmx.Task], real_x_series_device: nidaqmx.system.Device +) -> nidaqmx.Task: + task = generate_task() + # X Series port 1 has 8 lines, and can be used with any sized port write + task.do_channels.add_do_chan( + real_x_series_device.do_ports[1].name, + line_grouping=LineGrouping.CHAN_FOR_ALL_LINES, + ) + # X Series port 2 has 8 lines, and can be used with any sized port write + task.do_channels.add_do_chan( + real_x_series_device.do_ports[2].name, + line_grouping=LineGrouping.CHAN_FOR_ALL_LINES, + ) + _start_do_task(task, is_port=True, num_chans=task.number_of_channels) + return task + + +def _start_di_task(task: nidaqmx.Task) -> None: + # don't reserve the lines, so we can loop them back + for chan in task.di_channels: + chan.di_tristate = False + # we'll be doing simple on-demand, so start the task now + task.start() + + +@pytest.fixture +def di_single_line_loopback_task( + generate_task: Callable[[], nidaqmx.Task], + real_x_series_device: nidaqmx.system.Device, +) -> nidaqmx.Task: + task = generate_task() + task.di_channels.add_di_chan( + real_x_series_device.di_lines[0].name, line_grouping=LineGrouping.CHAN_FOR_ALL_LINES + ) + _start_di_task(task) + return task + + +@pytest.fixture +def di_multi_line_loopback_task( + generate_task: Callable[[], nidaqmx.Task], real_x_series_device: nidaqmx.system.Device +) -> nidaqmx.Task: + task = generate_task() + task.di_channels.add_di_chan( + flatten_channel_string(real_x_series_device.di_lines[:8].name), + line_grouping=LineGrouping.CHAN_FOR_ALL_LINES, + ) + _start_di_task(task) + return task + + +@pytest.fixture +def di_port0_loopback_task( + generate_task: Callable[[], nidaqmx.Task], real_x_series_device: nidaqmx.system.Device +) -> nidaqmx.Task: + task = generate_task() + task.di_channels.add_di_chan( + real_x_series_device.di_ports[0].name, + line_grouping=LineGrouping.CHAN_FOR_ALL_LINES, + ) + _start_di_task(task) + return task + + +@pytest.fixture +def di_port1_loopback_task( + generate_task: Callable[[], nidaqmx.Task], real_x_series_device: nidaqmx.system.Device +) -> nidaqmx.Task: + task = generate_task() + task.di_channels.add_di_chan( + real_x_series_device.di_ports[1].name, + line_grouping=LineGrouping.CHAN_FOR_ALL_LINES, + ) + _start_di_task(task) + return task + + +@pytest.fixture +def di_multi_channel_port_loopback_task( + generate_task: Callable[[], nidaqmx.Task], real_x_series_device: nidaqmx.system.Device +) -> nidaqmx.Task: + task = generate_task() + task.di_channels.add_di_chan( + real_x_series_device.di_ports[1].name, + line_grouping=LineGrouping.CHAN_FOR_ALL_LINES, + ) + task.di_channels.add_di_chan( + real_x_series_device.di_ports[2].name, + line_grouping=LineGrouping.CHAN_FOR_ALL_LINES, + ) + _start_di_task(task) + return task + + +def _get_num_lines_in_task(task: nidaqmx.Task) -> int: + return sum([chan.do_num_lines for chan in task.channels]) + + +def _get_digital_data_for_sample(num_lines: int, sample_number: int) -> int: + result = 0 + # "Count" from 0 in binary within each group of 8 lines, like simulated data. + for _ in range((num_lines + 7) // 8): + result = (result << 8) | sample_number + + line_mask = (2**num_lines) - 1 + return result & line_mask + + +def _get_digital_data(num_lines: int, num_samples: int) -> List[int]: + return [ + _get_digital_data_for_sample(num_lines, sample_number) + for sample_number in range(num_samples) + ] + + +def _get_digital_port_data_for_sample(task: nidaqmx.Task, sample_number: int) -> List[int]: + return [ + _get_digital_data_for_sample(chan.do_num_lines, sample_number) for chan in task.channels + ] + + +def _get_digital_port_data_port_major(task: nidaqmx.Task, num_samples: int) -> List[List[int]]: + return [_get_digital_data(chan.do_num_lines, num_samples) for chan in task.channels] + + +def _get_digital_port_data_sample_major(task: nidaqmx.Task, num_samples: int) -> List[List[int]]: + result = _get_digital_port_data_port_major(task, num_samples) + return numpy.transpose(result).tolist() + + +def _int_to_bool_array(num_lines: int, input: int) -> numpy.typing.NDArray[numpy.bool_]: + result = numpy.full(num_lines, True, dtype=numpy.bool_) + for bit in range(num_lines): + result[bit] = (input & (1 << bit)) != 0 + return result + + +def test___digital_single_channel_writer___write_one_sample_one_line___updates_output( + do_single_line_task: nidaqmx.Task, + di_single_line_loopback_task: nidaqmx.Task, +) -> None: + writer = DigitalSingleChannelWriter(do_single_line_task.out_stream) + sample = True + + writer.write_one_sample_one_line(sample) + + assert di_single_line_loopback_task.read() == sample + + +def test___digital_single_channel_writer___write_one_sample_multi_line___updates_output( + do_single_channel_multi_line_task: nidaqmx.Task, + di_multi_line_loopback_task: nidaqmx.Task, +) -> None: + writer = DigitalSingleChannelWriter(do_single_channel_multi_line_task.out_stream) + num_lines = _get_num_lines_in_task(do_single_channel_multi_line_task) + samples_to_write = 256 + + # "sweep" up to the final value, the only one we'll validate + for datum in _get_digital_data(num_lines, samples_to_write): + writer.write_one_sample_multi_line(_int_to_bool_array(num_lines, datum)) + + assert di_multi_line_loopback_task.read() == datum + + +def test___digital_single_channel_writer___write_one_sample_multi_line_with_wrong_dtype___raises_error_with_correct_dtype( + do_single_channel_multi_line_task: nidaqmx.Task, +) -> None: + writer = DigitalSingleChannelWriter(do_single_channel_multi_line_task.out_stream) + num_lines = _get_num_lines_in_task(do_single_channel_multi_line_task) + sample = numpy.full(num_lines, math.inf, dtype=numpy.float64) + + with pytest.raises((ctypes.ArgumentError, TypeError)) as exc_info: + writer.write_one_sample_multi_line(sample) + + assert "bool" in exc_info.value.args[0] + + +def test___digital_single_channel_writer___write_one_sample_port_byte___updates_output( + do_port1_task: nidaqmx.Task, + di_port1_loopback_task: nidaqmx.Task, +) -> None: + writer = DigitalSingleChannelWriter(do_port1_task.out_stream) + num_lines = _get_num_lines_in_task(do_port1_task) + samples_to_write = 256 + + # "sweep" up to the final value, the only one we'll validate + for datum in _get_digital_data(num_lines, samples_to_write): + writer.write_one_sample_port_byte(datum) + + assert di_port1_loopback_task.read() == datum + + +def test___digital_single_channel_writer___write_many_sample_port_byte___updates_output( + do_port1_task: nidaqmx.Task, + di_port1_loopback_task: nidaqmx.Task, +) -> None: + writer = DigitalSingleChannelWriter(do_port1_task.out_stream) + num_lines = _get_num_lines_in_task(do_port1_task) + samples_to_write = 256 + data = numpy.array(_get_digital_data(num_lines, samples_to_write), dtype=numpy.uint8) + + # "sweep" up to the final value, the only one we'll validate + writer.write_many_sample_port_byte(data) + + assert di_port1_loopback_task.read() == data[-1] + + +def test___digital_single_channel_writer___write_many_sample_port_byte_with_wrong_dtype___raises_error_with_correct_dtype( + do_port1_task: nidaqmx.Task, +) -> None: + writer = DigitalSingleChannelWriter(do_port1_task.out_stream) + samples_to_write = 256 + data = numpy.full(samples_to_write, math.inf, dtype=numpy.float64) + + with pytest.raises((ctypes.ArgumentError, TypeError)) as exc_info: + writer.write_many_sample_port_byte(data) + + assert "uint8" in exc_info.value.args[0] + + +def test___digital_single_channel_writer___write_one_sample_port_uint16___updates_output( + do_port1_task: nidaqmx.Task, + di_port1_loopback_task: nidaqmx.Task, +) -> None: + writer = DigitalSingleChannelWriter(do_port1_task.out_stream) + num_lines = _get_num_lines_in_task(do_port1_task) + samples_to_write = 256 + + # "sweep" up to the final value, the only one we'll validate + for datum in _get_digital_data(num_lines, samples_to_write): + writer.write_one_sample_port_uint16(datum) + + assert di_port1_loopback_task.read() == datum + + +def test___digital_single_channel_writer___write_many_sample_port_uint16___updates_output( + do_port1_task: nidaqmx.Task, + di_port1_loopback_task: nidaqmx.Task, +) -> None: + writer = DigitalSingleChannelWriter(do_port1_task.out_stream) + num_lines = _get_num_lines_in_task(do_port1_task) + samples_to_write = 256 + data = numpy.array(_get_digital_data(num_lines, samples_to_write), dtype=numpy.uint16) + + # "sweep" up to the final value, the only one we'll validate + writer.write_many_sample_port_uint16(data) + + assert di_port1_loopback_task.read() == data[-1] + + +def test___digital_single_channel_writer___write_many_sample_port_uint16_with_wrong_dtype___raises_error_with_correct_dtype( + do_port1_task: nidaqmx.Task, +) -> None: + writer = DigitalSingleChannelWriter(do_port1_task.out_stream) + samples_to_write = 256 + data = numpy.full(samples_to_write, math.inf, dtype=numpy.float64) + + with pytest.raises((ctypes.ArgumentError, TypeError)) as exc_info: + writer.write_many_sample_port_uint16(data) + + assert "uint16" in exc_info.value.args[0] + + +def test___digital_single_channel_writer___write_one_sample_port_uint32___updates_output( + do_port0_task: nidaqmx.Task, + di_port0_loopback_task: nidaqmx.Task, +) -> None: + writer = DigitalSingleChannelWriter(do_port0_task.out_stream) + num_lines = _get_num_lines_in_task(do_port0_task) + samples_to_write = 256 + + # "sweep" up to the final value, the only one we'll validate + for datum in _get_digital_data(num_lines, samples_to_write): + writer.write_one_sample_port_uint32(datum) + + assert di_port0_loopback_task.read() == datum + + +def test___digital_single_channel_writer___write_many_sample_port_uint32___updates_output( + do_port0_task: nidaqmx.Task, + di_port0_loopback_task: nidaqmx.Task, +) -> None: + writer = DigitalSingleChannelWriter(do_port0_task.out_stream) + num_lines = _get_num_lines_in_task(do_port0_task) + samples_to_write = 256 + data = numpy.array(_get_digital_data(num_lines, samples_to_write), dtype=numpy.uint32) + + # "sweep" up to the final value, the only one we'll validate + writer.write_many_sample_port_uint32(data) + + assert di_port0_loopback_task.read() == data[-1] + + +def test___digital_single_channel_writer___write_many_sample_port_uint32_with_wrong_dtype___raises_error_with_correct_dtype( + do_port0_task: nidaqmx.Task, +) -> None: + writer = DigitalSingleChannelWriter(do_port0_task.out_stream) + samples_to_write = 256 + data = numpy.full(samples_to_write, math.inf, dtype=numpy.float64) + + with pytest.raises((ctypes.ArgumentError, TypeError)) as exc_info: + writer.write_many_sample_port_uint32(data) + + assert "uint32" in exc_info.value.args[0] + + +def test___digital_multi_channel_writer___write_one_sample_one_line___updates_output( + do_single_line_task: nidaqmx.Task, + di_single_line_loopback_task: nidaqmx.Task, +) -> None: + writer = DigitalMultiChannelWriter(do_single_line_task.out_stream) + num_lines = _get_num_lines_in_task(do_single_line_task) + samples_to_write = 256 + + # "sweep" up to the final value, the only one we'll validate + for datum in _get_digital_data(num_lines, samples_to_write): + writer.write_one_sample_one_line(_int_to_bool_array(num_lines, datum)) + + assert di_single_line_loopback_task.read() == datum + + +def test___digital_multi_channel_writer___write_one_sample_multi_line___updates_output( + do_multi_channel_multi_line_task: nidaqmx.Task, + di_multi_line_loopback_task: nidaqmx.Task, +) -> None: + writer = DigitalMultiChannelWriter(do_multi_channel_multi_line_task.out_stream) + num_channels = do_multi_channel_multi_line_task.number_of_channels + samples_to_write = 256 + + # "sweep" up to the final value, the only one we'll validate + for datum in _get_digital_data(num_channels, samples_to_write): + data_to_write = _int_to_bool_array(num_channels, datum).reshape((num_channels, 1)) + writer.write_one_sample_multi_line(data_to_write) + + assert di_multi_line_loopback_task.read() == datum + + +def test___digital_multi_channel_writer___write_one_sample_multi_line_with_wrong_dtype___raises_error_with_correct_dtype( + do_multi_channel_multi_line_task: nidaqmx.Task, +) -> None: + writer = DigitalMultiChannelWriter(do_multi_channel_multi_line_task.out_stream) + num_channels = do_multi_channel_multi_line_task.number_of_channels + sample = numpy.full((num_channels, 1), math.inf, dtype=numpy.float64) + + with pytest.raises((ctypes.ArgumentError, TypeError)) as exc_info: + writer.write_one_sample_multi_line(sample) + + assert "bool" in exc_info.value.args[0] + + +def test___digital_multi_channel_writer___write_one_sample_port_byte___updates_output( + do_multi_channel_port_task: nidaqmx.Task, + di_multi_channel_port_loopback_task: nidaqmx.Task, +) -> None: + writer = DigitalMultiChannelWriter(do_multi_channel_port_task.out_stream) + samples_to_write = 256 + + # "sweep" up to the final value, the only one we'll validate + for datum in _get_digital_port_data_sample_major(do_multi_channel_port_task, samples_to_write): + writer.write_one_sample_port_byte(numpy.array(datum, dtype=numpy.uint8)) + + assert di_multi_channel_port_loopback_task.read() == datum + + +def test___digital_multi_channel_writer___write_one_sample_port_byte_with_wrong_dtype___raises_error_with_correct_dtype( + do_multi_channel_port_task: nidaqmx.Task, +) -> None: + writer = DigitalMultiChannelWriter(do_multi_channel_port_task.out_stream) + num_channels = do_multi_channel_port_task.number_of_channels + data = numpy.full(num_channels, math.inf, dtype=numpy.float64) + + with pytest.raises((ctypes.ArgumentError, TypeError)) as exc_info: + writer.write_one_sample_port_byte(data) + + assert "uint8" in exc_info.value.args[0] + + +def test___digital_multi_channel_writer___write_many_sample_port_byte___updates_output( + do_multi_channel_port_task: nidaqmx.Task, + di_multi_channel_port_loopback_task: nidaqmx.Task, +) -> None: + writer = DigitalMultiChannelWriter(do_multi_channel_port_task.out_stream) + samples_to_write = 256 + + # "sweep" up to the final value, the only one we'll validate + data = _get_digital_port_data_port_major(do_multi_channel_port_task, samples_to_write) + writer.write_many_sample_port_byte(numpy.array(data, dtype=numpy.uint8)) + + assert di_multi_channel_port_loopback_task.read() == _get_digital_port_data_for_sample( + do_multi_channel_port_task, samples_to_write - 1 + ) + + +def test___digital_multi_channel_writer___write_many_sample_port_byte_with_wrong_dtype___raises_error_with_correct_dtype( + do_multi_channel_port_task: nidaqmx.Task, +) -> None: + writer = DigitalMultiChannelWriter(do_multi_channel_port_task.out_stream) + num_channels = do_multi_channel_port_task.number_of_channels + samples_to_write = 256 + data = numpy.full((num_channels, samples_to_write), math.inf, dtype=numpy.float64) + + with pytest.raises((ctypes.ArgumentError, TypeError)) as exc_info: + writer.write_many_sample_port_byte(data) + + assert "uint8" in exc_info.value.args[0] + + +def test___digital_multi_channel_writer___write_one_sample_port_uint16___updates_output( + do_multi_channel_port_task: nidaqmx.Task, + di_multi_channel_port_loopback_task: nidaqmx.Task, +) -> None: + writer = DigitalMultiChannelWriter(do_multi_channel_port_task.out_stream) + samples_to_write = 256 + + # "sweep" up to the final value, the only one we'll validate + for datum in _get_digital_port_data_sample_major(do_multi_channel_port_task, samples_to_write): + writer.write_one_sample_port_uint16(numpy.array(datum, dtype=numpy.uint16)) + + assert di_multi_channel_port_loopback_task.read() == datum + + +def test___digital_multi_channel_writer___write_one_sample_port_uint16_with_wrong_dtype___raises_error_with_correct_dtype( + do_multi_channel_port_task: nidaqmx.Task, +) -> None: + writer = DigitalMultiChannelWriter(do_multi_channel_port_task.out_stream) + num_channels = do_multi_channel_port_task.number_of_channels + data = numpy.full(num_channels, math.inf, dtype=numpy.float64) + + with pytest.raises((ctypes.ArgumentError, TypeError)) as exc_info: + writer.write_one_sample_port_uint16(data) + + assert "uint16" in exc_info.value.args[0] + + +def test___digital_multi_channel_writer___write_many_sample_port_uint16___updates_output( + do_multi_channel_port_task: nidaqmx.Task, + di_multi_channel_port_loopback_task: nidaqmx.Task, +) -> None: + writer = DigitalMultiChannelWriter(do_multi_channel_port_task.out_stream) + samples_to_write = 256 + + # "sweep" up to the final value, the only one we'll validate + data = _get_digital_port_data_port_major(do_multi_channel_port_task, samples_to_write) + writer.write_many_sample_port_uint16(numpy.array(data, dtype=numpy.uint16)) + + assert di_multi_channel_port_loopback_task.read() == _get_digital_port_data_for_sample( + do_multi_channel_port_task, samples_to_write - 1 + ) + + +def test___digital_multi_channel_writer___write_many_sample_port_uint16_with_wrong_dtype___raises_error_with_correct_dtype( + do_multi_channel_port_task: nidaqmx.Task, +) -> None: + writer = DigitalMultiChannelWriter(do_multi_channel_port_task.out_stream) + num_channels = do_multi_channel_port_task.number_of_channels + samples_to_write = 256 + data = numpy.full((num_channels, samples_to_write), math.inf, dtype=numpy.float64) + + with pytest.raises((ctypes.ArgumentError, TypeError)) as exc_info: + writer.write_many_sample_port_uint16(data) + + assert "uint16" in exc_info.value.args[0] + + +def test___digital_multi_channel_writer___write_one_sample_port_uint32___updates_output( + do_multi_channel_port_task: nidaqmx.Task, + di_multi_channel_port_loopback_task: nidaqmx.Task, +) -> None: + writer = DigitalMultiChannelWriter(do_multi_channel_port_task.out_stream) + samples_to_write = 256 + + # "sweep" up to the final value, the only one we'll validate + for datum in _get_digital_port_data_sample_major(do_multi_channel_port_task, samples_to_write): + writer.write_one_sample_port_uint32(numpy.array(datum, dtype=numpy.uint32)) + + assert di_multi_channel_port_loopback_task.read() == datum + + +def test___digital_multi_channel_writer___write_one_sample_port_uint32_with_wrong_dtype___raises_error_with_correct_dtype( + do_multi_channel_port_task: nidaqmx.Task, +) -> None: + writer = DigitalMultiChannelWriter(do_multi_channel_port_task.out_stream) + num_channels = do_multi_channel_port_task.number_of_channels + data = numpy.full(num_channels, math.inf, dtype=numpy.float64) + + with pytest.raises((ctypes.ArgumentError, TypeError)) as exc_info: + writer.write_one_sample_port_uint32(data) + + assert "uint32" in exc_info.value.args[0] + + +def test___digital_multi_channel_writer___write_many_sample_port_uint32___updates_output( + do_multi_channel_port_task: nidaqmx.Task, + di_multi_channel_port_loopback_task: nidaqmx.Task, +) -> None: + writer = DigitalMultiChannelWriter(do_multi_channel_port_task.out_stream) + samples_to_write = 256 + + # "sweep" up to the final value, the only one we'll validate + data = _get_digital_port_data_port_major(do_multi_channel_port_task, samples_to_write) + writer.write_many_sample_port_uint32(numpy.array(data, dtype=numpy.uint32)) + + assert di_multi_channel_port_loopback_task.read() == _get_digital_port_data_for_sample( + do_multi_channel_port_task, samples_to_write - 1 + ) + + +def test___digital_multi_channel_writer___write_many_sample_port_uint32_with_wrong_dtype___raises_error_with_correct_dtype( + do_multi_channel_port_task: nidaqmx.Task, +) -> None: + writer = DigitalMultiChannelWriter(do_multi_channel_port_task.out_stream) + num_channels = do_multi_channel_port_task.number_of_channels + samples_to_write = 256 + data = numpy.full((num_channels, samples_to_write), math.inf, dtype=numpy.float64) + + with pytest.raises((ctypes.ArgumentError, TypeError)) as exc_info: + writer.write_many_sample_port_uint32(data) + + assert "uint32" in exc_info.value.args[0] From 4105e38c541c9214997598f177a916990ac01037 Mon Sep 17 00:00:00 2001 From: Zach Hindes Date: Tue, 5 Mar 2024 17:30:53 -0600 Subject: [PATCH 2/3] cleanup --- tests/component/test_stream_writers_do.py | 7 ++----- 1 file changed, 2 insertions(+), 5 deletions(-) diff --git a/tests/component/test_stream_writers_do.py b/tests/component/test_stream_writers_do.py index 1affa9fe..61bfeece 100644 --- a/tests/component/test_stream_writers_do.py +++ b/tests/component/test_stream_writers_do.py @@ -12,10 +12,8 @@ def _start_do_task(task: nidaqmx.Task, is_port: bool = False, num_chans: int = 1) -> None: - # we'll be doing simple on-demand, so start the task... + # We'll be doing on-demand, so start the task and drive all lines low task.start() - - # ... and set the output to a known initial value if is_port: if num_chans == 8: task.write(0) @@ -115,10 +113,9 @@ def do_multi_channel_port_task( def _start_di_task(task: nidaqmx.Task) -> None: - # don't reserve the lines, so we can loop them back + # Don't reserve the lines, so we can read what DO is writing. for chan in task.di_channels: chan.di_tristate = False - # we'll be doing simple on-demand, so start the task now task.start() From f0e5a5bde7c99c6a8c6758ce3f441d25867336bb Mon Sep 17 00:00:00 2001 From: Zach Hindes Date: Wed, 6 Mar 2024 10:42:16 -0600 Subject: [PATCH 3/3] code review --- tests/component/test_stream_readers_di.py | 4 ++-- tests/component/test_stream_writers_ao.py | 12 ++++++------ tests/component/test_stream_writers_do.py | 13 ++++++------- 3 files changed, 14 insertions(+), 15 deletions(-) diff --git a/tests/component/test_stream_readers_di.py b/tests/component/test_stream_readers_di.py index 6aeb9da2..68995e2c 100644 --- a/tests/component/test_stream_readers_di.py +++ b/tests/component/test_stream_readers_di.py @@ -28,7 +28,7 @@ def di_single_channel_multi_line_task( task: nidaqmx.Task, sim_6363_device: nidaqmx.system.Device ) -> nidaqmx.Task: task.di_channels.add_di_chan( - flatten_channel_string(sim_6363_device.di_lines[:8].name), + flatten_channel_string(sim_6363_device.di_lines.channel_names[:8]), line_grouping=LineGrouping.CHAN_FOR_ALL_LINES, ) return task @@ -39,7 +39,7 @@ def di_multi_channel_multi_line_task( task: nidaqmx.Task, sim_6363_device: nidaqmx.system.Device ) -> nidaqmx.Task: task.di_channels.add_di_chan( - flatten_channel_string(sim_6363_device.di_lines[:8].name), + flatten_channel_string(sim_6363_device.di_lines.channel_names[:8]), line_grouping=LineGrouping.CHAN_PER_LINE, ) return task diff --git a/tests/component/test_stream_writers_ao.py b/tests/component/test_stream_writers_ao.py index 8cb4e797..87bc12be 100644 --- a/tests/component/test_stream_writers_ao.py +++ b/tests/component/test_stream_writers_ao.py @@ -95,7 +95,7 @@ def ao_multi_channel_task( task.start() # set the output to a known initial value - task.write([0.0 for _ in range(num_chans)]) + task.write([0.0] * num_chans) return task @@ -201,7 +201,7 @@ def test___analog_multi_channel_writer___write_many_sample___updates_output( data = numpy.ascontiguousarray( numpy.transpose( numpy.linspace( - [0.0 for _ in range(num_channels)], + [0.0] * num_channels, expected, num=samples_to_write, dtype=numpy.float64, @@ -241,7 +241,7 @@ def test___analog_unscaled_writer___write_int16___updates_output( data = numpy.ascontiguousarray( numpy.transpose( numpy.linspace( - [0 for _ in range(num_channels)], + [0] * num_channels, [_volts_to_codes(v) for v in expected], num=samples_to_write, dtype=numpy.int16, @@ -283,7 +283,7 @@ def test___analog_unscaled_writer___write_int32___updates_output( data = numpy.ascontiguousarray( numpy.transpose( numpy.linspace( - [0 for _ in range(num_channels)], + [0] * num_channels, [_volts_to_codes(v) for v in expected], num=samples_to_write, dtype=numpy.int32, @@ -325,7 +325,7 @@ def test___analog_unscaled_writer___write_uint16___updates_output( data = numpy.ascontiguousarray( numpy.transpose( numpy.linspace( - [0 for _ in range(num_channels)], + [0] * num_channels, [_volts_to_codes(v) for v in expected], num=samples_to_write, dtype=numpy.uint16, @@ -367,7 +367,7 @@ def test___analog_unscaled_writer___write_uint32___updates_output( data = numpy.ascontiguousarray( numpy.transpose( numpy.linspace( - [0 for _ in range(num_channels)], + [0] * num_channels, [_volts_to_codes(v) for v in expected], num=samples_to_write, dtype=numpy.uint32, diff --git a/tests/component/test_stream_writers_do.py b/tests/component/test_stream_writers_do.py index 61bfeece..2f245e5a 100644 --- a/tests/component/test_stream_writers_do.py +++ b/tests/component/test_stream_writers_do.py @@ -18,12 +18,12 @@ def _start_do_task(task: nidaqmx.Task, is_port: bool = False, num_chans: int = 1 if num_chans == 8: task.write(0) else: - task.write([0 for _ in range(num_chans)]) + task.write([0] * num_chans) else: if num_chans == 1: task.write(False) else: - task.write([False for _ in range(num_chans)]) + task.write([False] * num_chans) @pytest.fixture @@ -44,7 +44,7 @@ def do_single_channel_multi_line_task( ) -> nidaqmx.Task: task = generate_task() chan = task.do_channels.add_do_chan( - flatten_channel_string(real_x_series_device.do_lines[:8].name), + flatten_channel_string(real_x_series_device.do_lines.channel_names[:8]), line_grouping=LineGrouping.CHAN_FOR_ALL_LINES, ) _start_do_task(task, num_chans=chan.do_num_lines) @@ -57,7 +57,7 @@ def do_multi_channel_multi_line_task( ) -> nidaqmx.Task: task = generate_task() task.do_channels.add_do_chan( - flatten_channel_string(real_x_series_device.do_lines[:8].name), + flatten_channel_string(real_x_series_device.do_lines.channel_names[:8]), line_grouping=LineGrouping.CHAN_PER_LINE, ) _start_do_task(task, num_chans=task.number_of_channels) @@ -114,8 +114,7 @@ def do_multi_channel_port_task( def _start_di_task(task: nidaqmx.Task) -> None: # Don't reserve the lines, so we can read what DO is writing. - for chan in task.di_channels: - chan.di_tristate = False + task.di_channels.all.di_tristate = False task.start() @@ -138,7 +137,7 @@ def di_multi_line_loopback_task( ) -> nidaqmx.Task: task = generate_task() task.di_channels.add_di_chan( - flatten_channel_string(real_x_series_device.di_lines[:8].name), + flatten_channel_string(real_x_series_device.di_lines.channel_names[:8]), line_grouping=LineGrouping.CHAN_FOR_ALL_LINES, ) _start_di_task(task)