diff --git a/poetry.lock b/poetry.lock index a072728f..4518cc66 100644 --- a/poetry.lock +++ b/poetry.lock @@ -2465,14 +2465,14 @@ toml = ">=0.10.1" [[package]] name = "nitypes" -version = "1.0.0" +version = "1.1.0.dev1" description = "Data types for NI Python APIs" optional = false python-versions = "<4.0,>=3.9" groups = ["main"] files = [ - {file = "nitypes-1.0.0-py3-none-any.whl", hash = "sha256:545dff978040fd92a68876023b8b6ff666c1d861ead1b3e7771108754bd57345"}, - {file = "nitypes-1.0.0.tar.gz", hash = "sha256:dc5ecb8b5f52570c2311daaa60ce6dca9712d7151236ef1f5314ad6ac02e0bdc"}, + {file = "nitypes-1.1.0.dev1-py3-none-any.whl", hash = "sha256:d98ad6e3f8b92db76b5c1c584431fa27d3e74cce6e20464e43ee0117b02fe089"}, + {file = "nitypes-1.1.0.dev1.tar.gz", hash = "sha256:50b23e00cc6960996656c4c9ef0ca71dd267fc5c9ca481077b682c29190aa2d3"}, ] [package.dependencies] @@ -4121,4 +4121,4 @@ grpc = ["grpcio", "ni-grpcdevice-v1-proto", "ni-protobuf-types", "protobuf"] [metadata] lock-version = "2.1" python-versions = ">=3.9,<4.0" -content-hash = "f7d53512dc12a25658f1a2bb319e6f2c4fe1c2c037b19f7b6de53646f9f72ed8" +content-hash = "80db624a7860c7c0497305f8f9d2c685b6745075dc55ae030381d12157129d22" diff --git a/pyproject.toml b/pyproject.toml index 14c03b32..79ba78bd 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -63,7 +63,7 @@ click = ">=8.0.0" distro = { version = ">=1.9.0", platform = "linux" } requests = ">=2.25.0" typing_extensions = { version = ">=4.0.0" } -nitypes = {version=">=1.0.0"} +nitypes = {version=">=1.1.0.dev1"} [tool.poetry.group.codegen.dependencies] Mako = "^1.2" diff --git a/tests/component/_digital_utils.py b/tests/component/_digital_utils.py index 80be6526..42393331 100644 --- a/tests/component/_digital_utils.py +++ b/tests/component/_digital_utils.py @@ -10,6 +10,7 @@ from nitypes.waveform.typing import AnyDigitalState, TDigitalState import nidaqmx +import nidaqmx.system _D = TypeVar("_D", bound=numpy.generic) @@ -119,9 +120,9 @@ def _get_digital_port_data_sample_major(task: nidaqmx.Task, num_samples: int) -> return numpy.transpose(result).tolist() -def _bool_array_to_int(bool_array: numpy.typing.NDArray[numpy.bool_]) -> int: +def _bool_array_to_int_lsb(bool_array: numpy.typing.NDArray[numpy.bool_]) -> int: result = 0 - # Simulated data is little-endian + # Interpret data as little-endian for bit in bool_array[::-1]: result = (result << 1) | int(bit) return result @@ -129,7 +130,7 @@ def _bool_array_to_int(bool_array: numpy.typing.NDArray[numpy.bool_]) -> int: def _bool_array_to_int_msb(bool_array: numpy.typing.NDArray[numpy.bool_]) -> int: result = 0 - # Data from ports is big-endian (see AB#3178052) + # Interpret data as big-endian for bit in bool_array: result = (result << 1) | int(bit) return result @@ -144,10 +145,10 @@ def _int_to_bool_array(num_lines: int, input: int) -> numpy.typing.NDArray[numpy def _get_waveform_data(waveform: DigitalWaveform[Any]) -> list[int]: assert isinstance(waveform, DigitalWaveform) - return [_bool_array_to_int(sample) for sample in waveform.data] + return [_bool_array_to_int_lsb(sample) for sample in waveform.data] -def _get_waveform_data_msb(waveform: DigitalWaveform[Any]) -> list[int]: +def _get_waveform_port_data(waveform: DigitalWaveform[Any]) -> list[int]: assert isinstance(waveform, DigitalWaveform) return [_bool_array_to_int_msb(sample) for sample in waveform.data] @@ -232,3 +233,20 @@ def _read_and_copy( ) -> numpy.typing.NDArray[_D]: read_func(array) return array.copy() + + +def _validate_waveform_signals( + device: nidaqmx.system.device.Device, + waveform: DigitalWaveform[Any], + lines: list[int] | range, # signal index to line index mapping +) -> None: + lines_list = list(lines) + signal_count = waveform.signal_count + sample_count = waveform.sample_count + assert signal_count == len(lines_list) + for signal_index in range(signal_count): + line_index = lines_list[signal_index] + signal = waveform.signals[signal_index] + assert signal.signal_index == signal_index + assert signal.name == device.di_lines[line_index].name + assert signal.data.tolist() == _get_expected_data_for_line(sample_count, line_index) diff --git a/tests/component/stream_readers/test_digital_multi_channel_reader.py b/tests/component/stream_readers/test_digital_multi_channel_reader.py index 18667432..5d25593e 100644 --- a/tests/component/stream_readers/test_digital_multi_channel_reader.py +++ b/tests/component/stream_readers/test_digital_multi_channel_reader.py @@ -13,9 +13,9 @@ import nidaqmx.system from nidaqmx._feature_toggles import WAVEFORM_SUPPORT, FeatureNotSupportedError from nidaqmx.constants import ( + READ_ALL_AVAILABLE, AcquisitionType, LineGrouping, - READ_ALL_AVAILABLE, ReallocationPolicy, WaveformAttributeMode, ) @@ -23,14 +23,16 @@ from nidaqmx.stream_readers import DaqError, DigitalMultiChannelReader from nidaqmx.utils import flatten_channel_string from tests.component._digital_utils import ( - _bool_array_to_int, - _get_expected_data_for_line, + _bool_array_to_int_lsb, _get_digital_data, + _get_expected_data_for_line, _get_expected_digital_port_data_port_major, _get_expected_digital_port_data_sample_major, _get_num_di_lines_in_task, _get_waveform_data, + _get_waveform_port_data, _read_and_copy, + _validate_waveform_signals, ) from tests.component._utils import _is_timestamp_close_to_now @@ -45,7 +47,7 @@ def test___digital_multi_channel_reader___read_one_sample_one_line___returns_val data = [_read_and_copy(reader.read_one_sample_one_line, sample) for _ in range(samples_to_read)] - assert [_bool_array_to_int(sample) for sample in data] == _get_digital_data( + assert [_bool_array_to_int_lsb(sample) for sample in data] == _get_digital_data( num_lines, samples_to_read ) @@ -75,7 +77,7 @@ def test___digital_multi_channel_reader___read_one_sample_multi_line___returns_v _read_and_copy(reader.read_one_sample_multi_line, sample) for _ in range(samples_to_read) ] - assert [_bool_array_to_int(sample[:, 0]) for sample in data] == _get_digital_data( + assert [_bool_array_to_int_lsb(sample[:, 0]) for sample in data] == _get_digital_data( num_channels, samples_to_read ) @@ -93,7 +95,7 @@ def test___digital_multi_channel_reader___read_one_sample_multi_line_jagged___re ] assert [ - [_bool_array_to_int(sample[chan, :]) for chan in range(num_channels)] for sample in data + [_bool_array_to_int_lsb(sample[chan, :]) for chan in range(num_channels)] for sample in data ] == _get_expected_digital_port_data_sample_major( di_multi_channel_port_uint32_task, samples_to_read ) @@ -330,6 +332,7 @@ def test___digital_multi_channel_multi_line_reader___read_waveforms_feature_disa def test___digital_multi_channel_multi_line_reader___read_waveforms___returns_valid_waveforms( di_multi_chan_multi_line_timing_task: nidaqmx.Task, + sim_6363_device: nidaqmx.system.Device, ) -> None: reader = DigitalMultiChannelReader(di_multi_chan_multi_line_timing_task.in_stream) num_channels = di_multi_chan_multi_line_timing_task.number_of_channels @@ -350,6 +353,7 @@ def test___digital_multi_channel_multi_line_reader___read_waveforms___returns_va assert waveform.timing.sample_interval == ht_timedelta(seconds=1 / 1000) assert waveform.channel_name == di_multi_chan_multi_line_timing_task.di_channels[chan].name assert waveform.sample_count == samples_to_read + _validate_waveform_signals(sim_6363_device, waveform, [chan]) def test___digital_multi_channel_different_lines_reader___read_waveforms___returns_valid_waveforms( @@ -378,29 +382,19 @@ def test___digital_multi_channel_different_lines_reader___read_waveforms___retur assert waveforms[0].sample_count == samples_to_read assert waveforms[0].timing.sample_interval == ht_timedelta(seconds=1 / 1000) assert waveforms[0].channel_name == di_multi_chan_diff_lines_timing_task.di_channels[0].name - assert waveforms[0]._get_signal_names() == [ - sim_6363_device.di_lines[0].name, - ] + _validate_waveform_signals(sim_6363_device, waveforms[0], [0]) assert _get_waveform_data(waveforms[1]) == [0, 0, 1, 1, 2, 2, 3, 3, 0, 0] assert _is_timestamp_close_to_now(waveforms[1].timing.timestamp) assert waveforms[1].sample_count == samples_to_read assert waveforms[1].timing.sample_interval == ht_timedelta(seconds=1 / 1000) assert waveforms[1].channel_name == di_multi_chan_diff_lines_timing_task.di_channels[1].name - assert waveforms[1]._get_signal_names() == [ - sim_6363_device.di_lines[1].name, - sim_6363_device.di_lines[2].name, - ] + _validate_waveform_signals(sim_6363_device, waveforms[1], [2, 1]) assert _get_waveform_data(waveforms[2]) == [0, 0, 0, 0, 0, 0, 0, 0, 1, 1] assert _is_timestamp_close_to_now(waveforms[2].timing.timestamp) assert waveforms[2].sample_count == samples_to_read assert waveforms[2].timing.sample_interval == ht_timedelta(seconds=1 / 1000) assert waveforms[2].channel_name == di_multi_chan_diff_lines_timing_task.di_channels[2].name - assert waveforms[2]._get_signal_names() == [ - sim_6363_device.di_lines[3].name, - sim_6363_device.di_lines[4].name, - sim_6363_device.di_lines[5].name, - sim_6363_device.di_lines[6].name, - ] + _validate_waveform_signals(sim_6363_device, waveforms[2], [6, 5, 4, 3]) def test___digital_multi_channel_lines_and_port_reader___read_waveforms___returns_valid_waveforms( @@ -429,43 +423,22 @@ def test___digital_multi_channel_lines_and_port_reader___read_waveforms___return assert _is_timestamp_close_to_now(waveforms[0].timing.timestamp) assert waveforms[0].sample_count == samples_to_read assert waveforms[0].channel_name == di_multi_chan_lines_and_port_task.di_channels[0].name - assert waveforms[0]._get_signal_names() == [ - sim_6363_device.di_lines[0].name, - ] + _validate_waveform_signals(sim_6363_device, waveforms[0], [0]) assert _get_waveform_data(waveforms[1]) == [0, 0, 1, 1, 2, 2, 3, 3, 0, 0] assert _is_timestamp_close_to_now(waveforms[1].timing.timestamp) assert waveforms[1].sample_count == samples_to_read assert waveforms[1].channel_name == di_multi_chan_lines_and_port_task.di_channels[1].name - assert waveforms[1]._get_signal_names() == [ - sim_6363_device.di_lines[1].name, - sim_6363_device.di_lines[2].name, - ] + _validate_waveform_signals(sim_6363_device, waveforms[1], [2, 1]) assert _get_waveform_data(waveforms[2]) == [0, 0, 0, 0, 0, 0, 0, 0, 1, 1] assert _is_timestamp_close_to_now(waveforms[2].timing.timestamp) assert waveforms[2].sample_count == samples_to_read assert waveforms[2].channel_name == di_multi_chan_lines_and_port_task.di_channels[2].name - assert waveforms[2]._get_signal_names() == [ - sim_6363_device.di_lines[3].name, - sim_6363_device.di_lines[4].name, - sim_6363_device.di_lines[5].name, - sim_6363_device.di_lines[6].name, - ] - # Note, the data on the port's waveform is MSB instead of LSB because of bug AB#3178052 - # When that bug is fixed, these asserts should be updated - assert _get_waveform_data(waveforms[3]) == [0, 128, 64, 192, 32, 160, 96, 224, 16, 144] + _validate_waveform_signals(sim_6363_device, waveforms[2], [6, 5, 4, 3]) + assert _get_waveform_port_data(waveforms[3]) == [0, 1, 2, 3, 4, 5, 6, 7, 8, 9] assert _is_timestamp_close_to_now(waveforms[3].timing.timestamp) assert waveforms[3].sample_count == samples_to_read assert waveforms[3].channel_name == di_multi_chan_lines_and_port_task.di_channels[3].name - assert waveforms[3]._get_signal_names() == [ - sim_6363_device.di_lines[39].name, - sim_6363_device.di_lines[38].name, - sim_6363_device.di_lines[37].name, - sim_6363_device.di_lines[36].name, - sim_6363_device.di_lines[35].name, - sim_6363_device.di_lines[34].name, - sim_6363_device.di_lines[33].name, - sim_6363_device.di_lines[32].name, - ] + _validate_waveform_signals(sim_6363_device, waveforms[3], range(32, 40)) def test___digital_multi_channel_different_lines_reader___read_mismatched_waveforms___throws_exception( @@ -490,6 +463,7 @@ def test___digital_multi_channel_different_lines_reader___read_mismatched_wavefo def test___digital_multi_channel_multi_line_reader___read_waveforms_no_args___returns_valid_waveforms( di_multi_chan_multi_line_timing_task: nidaqmx.Task, + sim_6363_device: nidaqmx.system.Device, ) -> None: reader = DigitalMultiChannelReader(di_multi_chan_multi_line_timing_task.in_stream) num_channels = di_multi_chan_multi_line_timing_task.number_of_channels @@ -509,10 +483,12 @@ def test___digital_multi_channel_multi_line_reader___read_waveforms_no_args___re assert waveform.timing.sample_interval == ht_timedelta(seconds=1 / 1000) assert waveform.channel_name == di_multi_chan_multi_line_timing_task.di_channels[chan].name assert waveform.sample_count == 50 + _validate_waveform_signals(sim_6363_device, waveform, [chan]) def test___digital_multi_channel_multi_line_reader___read_waveforms_in_place___populates_valid_waveforms( di_multi_chan_multi_line_timing_task: nidaqmx.Task, + sim_6363_device: nidaqmx.system.Device, ) -> None: reader = DigitalMultiChannelReader(di_multi_chan_multi_line_timing_task.in_stream) num_channels = di_multi_chan_multi_line_timing_task.number_of_channels @@ -533,6 +509,7 @@ def test___digital_multi_channel_multi_line_reader___read_waveforms_in_place___p assert waveform.timing.sample_interval == ht_timedelta(seconds=1 / 1000) assert waveform.channel_name == di_multi_chan_multi_line_timing_task.di_channels[chan].name assert waveform.sample_count == samples_to_read + _validate_waveform_signals(sim_6363_device, waveform, [chan]) def test___digital_multi_channel_multi_line_reader___reuse_waveform_in_place___overwrites_data_timing_and_attributes( @@ -561,6 +538,7 @@ def _make_multi_channel_multi_line_reader(lines_start, rate): assert _get_waveform_data(waveform) == _get_expected_data_for_line(sample_count, chan) assert waveform.timing.sample_interval == ht_timedelta(seconds=1 / 1000) assert waveform.channel_name == sim_6363_device.di_lines[chan].name + _validate_waveform_signals(sim_6363_device, waveform, [chan]) reader1.read_waveforms(waveforms, sample_count) timestamps1 = [wf.timing.timestamp for wf in waveforms] @@ -568,6 +546,7 @@ def _make_multi_channel_multi_line_reader(lines_start, rate): assert _get_waveform_data(waveform) == _get_expected_data_for_line(sample_count, chan + 1) assert waveform.timing.sample_interval == ht_timedelta(seconds=1 / 2000) assert waveform.channel_name == sim_6363_device.di_lines[chan + 1].name + _validate_waveform_signals(sim_6363_device, waveform, [chan + 1]) for ts0, ts1 in zip(timestamps0, timestamps1): assert ts1 > ts0 @@ -590,6 +569,7 @@ def test___digital_multi_channel_multi_line_reader___read_into_undersized_wavefo def test___digital_multi_channel_multi_line_reader___read_into_undersized_waveforms___returns_valid_waveforms( di_multi_chan_multi_line_timing_task: nidaqmx.Task, + sim_6363_device: nidaqmx.system.Device, ) -> None: reader = DigitalMultiChannelReader(di_multi_chan_multi_line_timing_task.in_stream) num_channels = di_multi_chan_multi_line_timing_task.number_of_channels @@ -610,6 +590,7 @@ def test___digital_multi_channel_multi_line_reader___read_into_undersized_wavefo assert waveform.timing.sample_interval == ht_timedelta(seconds=1 / 1000) assert waveform.channel_name == di_multi_chan_multi_line_timing_task.di_channels[chan].name assert waveform.sample_count == samples_to_read + _validate_waveform_signals(sim_6363_device, waveform, [chan]) def test___digital_multi_channel_reader___reuse_waveform_in_place_with_different_sample_counts___populates_valid_waveforms( @@ -641,25 +622,31 @@ def _make_multi_channel_reader(chan_a_index, chan_b_index, samps_per_chan): reader0.read_waveforms(waveforms, 5) assert waveforms[0].sample_count == 5 assert _get_waveform_data(waveforms[0]) == _get_expected_data_for_line(5, 0) + _validate_waveform_signals(sim_6363_device, waveforms[0], [0]) assert waveforms[0].channel_name == f"{sim_6363_device.name}/port0/line0" assert waveforms[1].sample_count == 5 assert _get_waveform_data(waveforms[1]) == _get_expected_data_for_line(5, 1) + _validate_waveform_signals(sim_6363_device, waveforms[1], [1]) assert waveforms[1].channel_name == f"{sim_6363_device.name}/port0/line1" reader1.read_waveforms(waveforms, 10) assert waveforms[0].sample_count == 10 assert _get_waveform_data(waveforms[0]) == _get_expected_data_for_line(10, 2) + _validate_waveform_signals(sim_6363_device, waveforms[0], [2]) assert waveforms[0].channel_name == f"{sim_6363_device.name}/port0/line2" assert waveforms[1].sample_count == 10 assert _get_waveform_data(waveforms[1]) == _get_expected_data_for_line(10, 3) + _validate_waveform_signals(sim_6363_device, waveforms[1], [3]) assert waveforms[1].channel_name == f"{sim_6363_device.name}/port0/line3" reader2.read_waveforms(waveforms, 15) assert waveforms[0].sample_count == 15 assert _get_waveform_data(waveforms[0]) == _get_expected_data_for_line(15, 4) + _validate_waveform_signals(sim_6363_device, waveforms[0], [4]) assert waveforms[0].channel_name == f"{sim_6363_device.name}/port0/line4" assert waveforms[1].sample_count == 15 assert _get_waveform_data(waveforms[1]) == _get_expected_data_for_line(15, 5) + _validate_waveform_signals(sim_6363_device, waveforms[1], [5]) assert waveforms[1].channel_name == f"{sim_6363_device.name}/port0/line5" @@ -707,6 +694,7 @@ def test___digital_multi_channel_multi_line_reader_with_timing_flag___read_wavef def test___digital_multi_channel_multi_line_reader_with_extended_properties_flag___read_waveforms___only_includes_extended_properties( di_multi_chan_multi_line_timing_task: nidaqmx.Task, + sim_6363_device: nidaqmx.system.Device, ) -> None: in_stream = di_multi_chan_multi_line_timing_task.in_stream in_stream.waveform_attribute_mode = WaveformAttributeMode.EXTENDED_PROPERTIES @@ -728,10 +716,12 @@ def test___digital_multi_channel_multi_line_reader_with_extended_properties_flag assert waveform.timing.sample_interval_mode == SampleIntervalMode.NONE assert waveform.channel_name == di_multi_chan_multi_line_timing_task.di_channels[chan].name assert waveform.sample_count == samples_to_read + _validate_waveform_signals(sim_6363_device, waveform, [chan]) def test___digital_multi_channel_multi_line_reader_with_both_flags___read_waveforms___includes_both_timing_and_extended_properties( di_multi_chan_multi_line_timing_task: nidaqmx.Task, + sim_6363_device: nidaqmx.system.Device, ) -> None: in_stream = di_multi_chan_multi_line_timing_task.in_stream in_stream.waveform_attribute_mode = ( @@ -757,6 +747,7 @@ def test___digital_multi_channel_multi_line_reader_with_both_flags___read_wavefo assert waveform.timing.sample_interval == ht_timedelta(seconds=1 / 1000) assert waveform.channel_name == di_multi_chan_multi_line_timing_task.di_channels[chan].name assert waveform.sample_count == samples_to_read + _validate_waveform_signals(sim_6363_device, waveform, [chan]) def test___digital_multi_channel_multi_line_reader_with_none_flag___read_waveforms___minimal_waveform_data( @@ -795,6 +786,7 @@ def test___digital_multi_channel_multi_line_reader_with_none_flag___read_wavefor def test___digital_multi_channel_multi_line_reader___read_waveform_all_dtypes___returns_valid_waveform( di_multi_chan_multi_line_timing_task: nidaqmx.Task, dtype, + sim_6363_device: nidaqmx.system.Device, ) -> None: in_stream = di_multi_chan_multi_line_timing_task.in_stream reader = DigitalMultiChannelReader(in_stream) @@ -809,10 +801,12 @@ def test___digital_multi_channel_multi_line_reader___read_waveform_all_dtypes___ assert len(waveforms) == num_channels for chan, waveform in enumerate(waveforms): assert _get_waveform_data(waveform) == _get_expected_data_for_line(num_samples, chan) + _validate_waveform_signals(sim_6363_device, waveform, [chan]) def test___digital_multi_channel_multi_line_reader___read_waveforms_read_all_available___returns_valid_waveforms( di_multi_chan_multi_line_timing_task: nidaqmx.Task, + sim_6363_device: nidaqmx.system.Device, ) -> None: reader = DigitalMultiChannelReader(di_multi_chan_multi_line_timing_task.in_stream) num_channels = di_multi_chan_multi_line_timing_task.number_of_channels @@ -832,3 +826,4 @@ def test___digital_multi_channel_multi_line_reader___read_waveforms_read_all_ava assert waveform.timing.sample_interval == ht_timedelta(seconds=1 / 1000) assert waveform.channel_name == di_multi_chan_multi_line_timing_task.di_channels[chan].name assert waveform.sample_count == samples_read + _validate_waveform_signals(sim_6363_device, waveform, [chan]) diff --git a/tests/component/stream_readers/test_digital_single_channel_reader.py b/tests/component/stream_readers/test_digital_single_channel_reader.py index 1a01b560..bdbc83fa 100644 --- a/tests/component/stream_readers/test_digital_single_channel_reader.py +++ b/tests/component/stream_readers/test_digital_single_channel_reader.py @@ -13,9 +13,9 @@ import nidaqmx.system from nidaqmx._feature_toggles import WAVEFORM_SUPPORT, FeatureNotSupportedError from nidaqmx.constants import ( + READ_ALL_AVAILABLE, AcquisitionType, LineGrouping, - READ_ALL_AVAILABLE, ReallocationPolicy, WaveformAttributeMode, ) @@ -23,13 +23,14 @@ from nidaqmx.stream_readers import DaqError, DigitalSingleChannelReader from nidaqmx.utils import flatten_channel_string from tests.component._digital_utils import ( - _bool_array_to_int, + _bool_array_to_int_lsb, _get_digital_data, _get_expected_data_for_line, _get_num_di_lines_in_task, _get_waveform_data, - _get_waveform_data_msb, + _get_waveform_port_data, _read_and_copy, + _validate_waveform_signals, ) from tests.component._utils import _is_timestamp_close_to_now @@ -58,7 +59,7 @@ def test___digital_single_channel_reader___read_one_sample_multi_line___returns_ _read_and_copy(reader.read_one_sample_multi_line, sample) for _ in range(samples_to_read) ] - assert [_bool_array_to_int(sample) for sample in data] == _get_digital_data( + assert [_bool_array_to_int_lsb(sample) for sample in data] == _get_digital_data( num_lines, samples_to_read ) @@ -216,6 +217,7 @@ def test___digital_single_line_reader___read_waveform_feature_disabled___raises_ def test___digital_single_line_reader___read_waveform___returns_valid_waveform( di_single_line_timing_task: nidaqmx.Task, + sim_6363_device: nidaqmx.system.Device, ) -> None: reader = DigitalSingleChannelReader(di_single_line_timing_task.in_stream) samples_to_read = 10 @@ -229,10 +231,12 @@ def test___digital_single_line_reader___read_waveform___returns_valid_waveform( assert waveform.timing.sample_interval == ht_timedelta(seconds=1 / 1000) assert waveform.timing.sample_interval_mode == SampleIntervalMode.REGULAR assert waveform.channel_name == di_single_line_timing_task.di_channels[0].name + _validate_waveform_signals(sim_6363_device, waveform, [0]) def test___digital_single_channel_multi_line_reader___read_waveform___returns_valid_waveform( di_single_channel_multi_line_timing_task: nidaqmx.Task, + sim_6363_device: nidaqmx.system.Device, ) -> None: reader = DigitalSingleChannelReader(di_single_channel_multi_line_timing_task.in_stream) samples_to_read = 10 @@ -247,10 +251,12 @@ def test___digital_single_channel_multi_line_reader___read_waveform___returns_va assert waveform.timing.sample_interval == ht_timedelta(seconds=1 / 1000) assert waveform.timing.sample_interval_mode == SampleIntervalMode.REGULAR assert waveform.channel_name == di_single_channel_multi_line_timing_task.di_channels[0].name + _validate_waveform_signals(sim_6363_device, waveform, [7, 6, 5, 4, 3, 2, 1, 0]) def test___digital_single_line_reader___read_waveform_no_args___returns_valid_waveform( di_single_line_timing_task: nidaqmx.Task, + sim_6363_device: nidaqmx.system.Device, ) -> None: reader = DigitalSingleChannelReader(di_single_line_timing_task.in_stream) waveform = DigitalWaveform(50) @@ -263,10 +269,12 @@ def test___digital_single_line_reader___read_waveform_no_args___returns_valid_wa assert waveform.timing.sample_interval == ht_timedelta(seconds=1 / 1000) assert waveform.timing.sample_interval_mode == SampleIntervalMode.REGULAR assert waveform.channel_name == di_single_line_timing_task.di_channels[0].name + _validate_waveform_signals(sim_6363_device, waveform, [0]) def test___digital_single_channel_multi_line_reader___read_waveform_no_args___returns_valid_waveform( di_single_channel_multi_line_timing_task: nidaqmx.Task, + sim_6363_device: nidaqmx.system.Device, ) -> None: reader = DigitalSingleChannelReader(di_single_channel_multi_line_timing_task.in_stream) num_lines = _get_num_di_lines_in_task(di_single_channel_multi_line_timing_task) @@ -280,10 +288,12 @@ def test___digital_single_channel_multi_line_reader___read_waveform_no_args___re assert waveform.timing.sample_interval == ht_timedelta(seconds=1 / 1000) assert waveform.timing.sample_interval_mode == SampleIntervalMode.REGULAR assert waveform.channel_name == di_single_channel_multi_line_timing_task.di_channels[0].name + _validate_waveform_signals(sim_6363_device, waveform, range(num_lines)[::-1]) def test___digital_single_line_reader___read_waveform_in_place___returns_valid_waveform( di_single_line_timing_task: nidaqmx.Task, + sim_6363_device: nidaqmx.system.Device, ) -> None: reader = DigitalSingleChannelReader(di_single_line_timing_task.in_stream) waveform = DigitalWaveform(50) @@ -295,10 +305,12 @@ def test___digital_single_line_reader___read_waveform_in_place___returns_valid_w assert _is_timestamp_close_to_now(waveform.timing.timestamp) assert waveform.timing.sample_interval == ht_timedelta(seconds=1 / 1000) assert waveform.channel_name == di_single_line_timing_task.di_channels[0].name + _validate_waveform_signals(sim_6363_device, waveform, [0]) def test___digital_single_channel_multi_line_reader___read_waveform_in_place___returns_valid_waveform( di_single_channel_multi_line_timing_task: nidaqmx.Task, + sim_6363_device: nidaqmx.system.Device, ) -> None: reader = DigitalSingleChannelReader(di_single_channel_multi_line_timing_task.in_stream) num_lines = _get_num_di_lines_in_task(di_single_channel_multi_line_timing_task) @@ -311,6 +323,7 @@ def test___digital_single_channel_multi_line_reader___read_waveform_in_place___r assert _is_timestamp_close_to_now(waveform.timing.timestamp) assert waveform.timing.sample_interval == ht_timedelta(seconds=1 / 1000) assert waveform.channel_name == di_single_channel_multi_line_timing_task.di_channels[0].name + _validate_waveform_signals(sim_6363_device, waveform, range(num_lines)[::-1]) def test___digital_single_line_reader___reuse_waveform_in_place___overwrites_data_timing_and_attributes( @@ -335,12 +348,14 @@ def _make_single_line_reader(chan_index, rate): assert _get_waveform_data(waveform) == _get_expected_data_for_line(sample_count, 0) assert waveform.timing.sample_interval == ht_timedelta(seconds=1 / 1000) assert waveform.channel_name == sim_6363_device.di_lines[0].name + _validate_waveform_signals(sim_6363_device, waveform, [0]) reader1.read_waveform(waveform, sample_count) timestamp1 = waveform.timing.timestamp assert _get_waveform_data(waveform) == _get_expected_data_for_line(sample_count, 1) assert waveform.timing.sample_interval == ht_timedelta(seconds=1 / 2000) assert waveform.channel_name == sim_6363_device.di_lines[1].name + _validate_waveform_signals(sim_6363_device, waveform, [1]) assert timestamp1 > timestamp0 @@ -370,12 +385,14 @@ def _make_single_channel_multi_line_reader(lines_start, rate): assert _get_waveform_data(waveform) == [0, 1, 2, 3, 4, 5, 6, 7, 8, 9] assert waveform.timing.sample_interval == ht_timedelta(seconds=1 / 1000) assert waveform.channel_name == f"{sim_6363_device.di_lines[0].name}..." + _validate_waveform_signals(sim_6363_device, waveform, [3, 2, 1, 0]) reader1.read_waveform(waveform, sample_count) timestamp1 = waveform.timing.timestamp assert _get_waveform_data(waveform) == [0, 0, 1, 1, 2, 2, 3, 3, 4, 4] assert waveform.timing.sample_interval == ht_timedelta(seconds=1 / 2000) assert waveform.channel_name == f"{sim_6363_device.di_lines[1].name}..." + _validate_waveform_signals(sim_6363_device, waveform, [4, 3, 2, 1]) assert timestamp1 > timestamp0 @@ -398,6 +415,7 @@ def test___digital_single_line_reader___read_into_undersized_waveform_without_re def test___digital_single_line_reader___read_into_undersized_waveform___returns_valid_waveform( di_single_line_timing_task: nidaqmx.Task, + sim_6363_device: nidaqmx.system.Device, ) -> None: reader = DigitalSingleChannelReader(di_single_line_timing_task.in_stream) samples_to_read = 10 @@ -411,6 +429,7 @@ def test___digital_single_line_reader___read_into_undersized_waveform___returns_ assert waveform.timing.sample_interval == ht_timedelta(seconds=1 / 1000) assert waveform.timing.sample_interval_mode == SampleIntervalMode.REGULAR assert waveform.channel_name == di_single_line_timing_task.di_channels[0].name + _validate_waveform_signals(sim_6363_device, waveform, [0]) def test___digital_single_channel_reader___reuse_waveform_in_place_with_different_sample_counts___populates_valid_waveforms( @@ -435,20 +454,23 @@ def _make_single_channel_reader(chan_index, samps_per_chan): assert waveform.sample_count == 5 assert _get_waveform_data(waveform) == _get_expected_data_for_line(5, 0) assert waveform.channel_name == f"{sim_6363_device.name}/port0/line0" + _validate_waveform_signals(sim_6363_device, waveform, [0]) reader1.read_waveform(waveform, 10) assert waveform.sample_count == 10 assert _get_waveform_data(waveform) == _get_expected_data_for_line(10, 1) assert waveform.channel_name == f"{sim_6363_device.name}/port0/line1" + _validate_waveform_signals(sim_6363_device, waveform, [1]) reader2.read_waveform(waveform, 15) assert waveform.sample_count == 15 assert _get_waveform_data(waveform) == _get_expected_data_for_line(15, 2) assert waveform.channel_name == f"{sim_6363_device.name}/port0/line2" + _validate_waveform_signals(sim_6363_device, waveform, [2]) def test___digital_single_line_reader___read_waveform_high_sample_rate___returns_correct_sample_interval( - di_single_line_high_rate_task: nidaqmx.Task, + di_single_line_high_rate_task: nidaqmx.Task, sim_6363_device: nidaqmx.system.Device ) -> None: reader = DigitalSingleChannelReader(di_single_line_high_rate_task.in_stream) samples_to_read = 50 @@ -462,6 +484,7 @@ def test___digital_single_line_reader___read_waveform_high_sample_rate___returns assert waveform.timing.sample_interval == ht_timedelta(seconds=1 / 10_000_000) assert waveform.sample_count == samples_to_read assert waveform.channel_name == di_single_line_high_rate_task.di_channels[0].name + _validate_waveform_signals(sim_6363_device, waveform, [0]) def test___digital_single_line_reader_with_timing_flag___read_waveform___only_includes_timing_data( @@ -483,7 +506,7 @@ def test___digital_single_line_reader_with_timing_flag___read_waveform___only_in def test___digital_single_line_reader_with_extended_properties_flag___read_waveform___only_includes_extended_properties( - di_single_line_timing_task: nidaqmx.Task, + di_single_line_timing_task: nidaqmx.Task, sim_6363_device: nidaqmx.system.Device ) -> None: in_stream = di_single_line_timing_task.in_stream in_stream.waveform_attribute_mode = WaveformAttributeMode.EXTENDED_PROPERTIES @@ -496,10 +519,11 @@ def test___digital_single_line_reader_with_extended_properties_flag___read_wavef assert _get_waveform_data(waveform) == _get_digital_data(1, 50) assert waveform.timing.sample_interval_mode == SampleIntervalMode.NONE assert waveform.channel_name == di_single_line_timing_task.di_channels[0].name + _validate_waveform_signals(sim_6363_device, waveform, [0]) def test___digital_single_line_reader_with_both_flags___read_waveform___includes_both_timing_and_extended_properties( - di_single_line_timing_task: nidaqmx.Task, + di_single_line_timing_task: nidaqmx.Task, sim_6363_device: nidaqmx.system.Device ) -> None: in_stream = di_single_line_timing_task.in_stream in_stream.waveform_attribute_mode = ( @@ -516,6 +540,7 @@ def test___digital_single_line_reader_with_both_flags___read_waveform___includes assert waveform.timing.sample_interval_mode == SampleIntervalMode.REGULAR assert waveform.timing.sample_interval == ht_timedelta(seconds=1 / 1000) assert waveform.channel_name == di_single_line_timing_task.di_channels[0].name + _validate_waveform_signals(sim_6363_device, waveform, [0]) def test___digital_single_line_reader_with_none_flag___read_waveform___minimal_waveform_data( @@ -535,7 +560,7 @@ def test___digital_single_line_reader_with_none_flag___read_waveform___minimal_w def test___digital_single_channel_port_uint32_reader___read_waveform___returns_valid_waveform( - di_single_channel_port_uint32_timing_task: nidaqmx.Task, + di_single_channel_port_uint32_timing_task: nidaqmx.Task, sim_6363_device: nidaqmx.system.Device ) -> None: reader = DigitalSingleChannelReader(di_single_channel_port_uint32_timing_task.in_stream) num_lines = 32 @@ -545,13 +570,12 @@ def test___digital_single_channel_port_uint32_reader___read_waveform___returns_v samples_read = reader.read_waveform(waveform, num_samples) assert samples_read == num_samples - assert _get_waveform_data_msb(waveform) == _get_digital_data( - num_lines, num_samples - ) # TODO: AB#3178052 - change to _get_waveform_data() + assert _get_waveform_port_data(waveform) == _get_digital_data(num_lines, num_samples) assert _is_timestamp_close_to_now(waveform.timing.timestamp) assert waveform.timing.sample_interval == ht_timedelta(seconds=1 / 1000) assert waveform.timing.sample_interval_mode == SampleIntervalMode.REGULAR assert waveform.channel_name == di_single_channel_port_uint32_timing_task.di_channels[0].name + _validate_waveform_signals(sim_6363_device, waveform, range(num_lines)) def test___digital_single_channel_lines_and_port___read_waveform___returns_valid_waveform( @@ -566,24 +590,21 @@ def test___digital_single_channel_lines_and_port___read_waveform___returns_valid samples_read = reader.read_waveform(waveform, samples_to_read) assert samples_read == samples_to_read - # Note, the data on the port's waveform is MSB instead of LSB because of bug AB#3178052 - # When that bug is fixed, these asserts should be updated - assert _get_waveform_data(waveform) == [0, 1025, 514, 1539, 260, 1285, 774, 1799, 128, 1153] + assert _get_waveform_port_data(waveform) == [ + 0b00000000000, + 0b10000000001, + 0b01000000010, + 0b11000000011, + 0b00100000100, + 0b10100000101, + 0b01100000110, + 0b11100000111, + 0b00000001000, + 0b10000001001, + ] assert waveform.sample_count == samples_to_read assert waveform.channel_name == di_single_chan_lines_and_port_task.di_channels[0].name - assert waveform._get_signal_names() == [ - sim_6363_device.di_lines[0].name, - sim_6363_device.di_lines[1].name, - sim_6363_device.di_lines[2].name, - sim_6363_device.di_lines[39].name, - sim_6363_device.di_lines[38].name, - sim_6363_device.di_lines[37].name, - sim_6363_device.di_lines[36].name, - sim_6363_device.di_lines[35].name, - sim_6363_device.di_lines[34].name, - sim_6363_device.di_lines[33].name, - sim_6363_device.di_lines[32].name, - ] + _validate_waveform_signals(sim_6363_device, waveform, [32, 33, 34, 35, 36, 37, 38, 39, 2, 1, 0]) @pytest.mark.parametrize( @@ -596,6 +617,7 @@ def test___digital_single_channel_lines_and_port___read_waveform___returns_valid ) def test___digital_single_channel_multi_line_reader___read_waveform_all_dtypes___returns_valid_waveform( di_single_channel_multi_line_timing_task: nidaqmx.Task, + sim_6363_device: nidaqmx.system.Device, dtype, ) -> None: reader = DigitalSingleChannelReader(di_single_channel_multi_line_timing_task.in_stream) @@ -607,10 +629,12 @@ def test___digital_single_channel_multi_line_reader___read_waveform_all_dtypes__ assert samples_read == num_samples assert _get_waveform_data(waveform) == _get_digital_data(num_lines, num_samples) + _validate_waveform_signals(sim_6363_device, waveform, [7, 6, 5, 4, 3, 2, 1, 0]) def test___digital_single_line_reader___read_waveform_read_all_available___returns_valid_waveform( di_single_line_timing_task: nidaqmx.Task, + sim_6363_device: nidaqmx.system.Device, ) -> None: reader = DigitalSingleChannelReader(di_single_line_timing_task.in_stream) waveform = DigitalWaveform(100) @@ -623,3 +647,4 @@ def test___digital_single_line_reader___read_waveform_read_all_available___retur assert waveform.timing.sample_interval == ht_timedelta(seconds=1 / 1000) assert waveform.timing.sample_interval_mode == SampleIntervalMode.REGULAR assert waveform.channel_name == di_single_line_timing_task.di_channels[0].name + _validate_waveform_signals(sim_6363_device, waveform, [0]) diff --git a/tests/component/stream_writers/test_digital_multi_channel_writer.py b/tests/component/stream_writers/test_digital_multi_channel_writer.py index 22c520c7..02355c9f 100644 --- a/tests/component/stream_writers/test_digital_multi_channel_writer.py +++ b/tests/component/stream_writers/test_digital_multi_channel_writer.py @@ -24,7 +24,7 @@ _get_digital_port_data_sample_major, _get_num_do_lines_in_task, _get_waveform_data, - _get_waveform_data_msb, + _get_waveform_port_data, _int_to_bool_array, _start_do_task, ) @@ -369,9 +369,9 @@ def test___digital_multi_channel_writer___write_waveforms_ports___outputs_match_ actual_value = di_multi_channel_port_loopback_task.read() assert actual_value[0] != actual_value[1] assert actual_value == [ - _get_waveform_data_msb(waveforms[0])[-1], - _get_waveform_data_msb(waveforms[1])[-1], - ] # TODO: AB#3178052 - change to _get_waveform_data() + _get_waveform_port_data(waveforms[0])[-1], + _get_waveform_port_data(waveforms[1])[-1], + ] def test___digital_multi_channel_writer___write_waveforms_port_and_lines___outputs_match_final_values( @@ -395,9 +395,9 @@ def test___digital_multi_channel_writer___write_waveforms_port_and_lines___outpu assert samples_written == num_samples actual_value = di_multi_channel_port_and_lines_loopback_task.read() assert actual_value == [ - _get_waveform_data_msb(waveforms[0])[-1], + _get_waveform_port_data(waveforms[0])[-1], _get_waveform_data(waveforms[1])[-1], - ] # TODO: AB#3178052 - change to _get_waveform_data() + ] def test___digital_multi_channel_writer___write_waveforms_with_non_contiguous_data___outputs_match_final_values( diff --git a/tests/component/stream_writers/test_digital_single_channel_writer.py b/tests/component/stream_writers/test_digital_single_channel_writer.py index a786463c..371b3574 100644 --- a/tests/component/stream_writers/test_digital_single_channel_writer.py +++ b/tests/component/stream_writers/test_digital_single_channel_writer.py @@ -17,7 +17,7 @@ _get_digital_data, _get_num_do_lines_in_task, _get_waveform_data, - _get_waveform_data_msb, + _get_waveform_port_data, _int_to_bool_array, ) @@ -426,9 +426,7 @@ def test___digital_single_channel_writer___write_waveform_port_uint8___outputs_m actual_value = di_port1_loopback_task.read() assert samples_written == num_samples assert waveform.signal_count == num_lines - assert ( - actual_value == _get_waveform_data_msb(waveform)[i - 1] - ) # TODO: AB#3178052 - change to _get_waveform_data() + assert actual_value == _get_waveform_port_data(waveform)[i - 1] def test___digital_single_channel_writer___write_waveform_port_uint32___outputs_match_final_values( @@ -450,6 +448,4 @@ def test___digital_single_channel_writer___write_waveform_port_uint32___outputs_ actual_value = di_port0_loopback_task.read() assert samples_written == num_samples assert waveform.signal_count == num_lines - assert ( - actual_value == _get_waveform_data_msb(waveform)[i - 1] - ) # TODO: AB#3178052 - change to _get_waveform_data() + assert actual_value == _get_waveform_port_data(waveform)[i - 1] diff --git a/tests/component/task/test_task_read_waveform_di.py b/tests/component/task/test_task_read_waveform_di.py index d141044a..f1757091 100644 --- a/tests/component/task/test_task_read_waveform_di.py +++ b/tests/component/task/test_task_read_waveform_di.py @@ -8,29 +8,36 @@ from tests.component._digital_utils import ( _get_expected_data_for_line, _get_waveform_data, + _get_waveform_port_data, + _validate_waveform_signals, ) def test___digital_single_channel___read_waveform___returns_valid_waveform( di_single_channel_timing_task: nidaqmx.Task, + sim_6363_device: nidaqmx.system.Device, ) -> None: waveform = di_single_channel_timing_task.read_waveform() assert isinstance(waveform, DigitalWaveform) assert _get_waveform_data(waveform) == _get_expected_data_for_line(50, 0) + _validate_waveform_signals(sim_6363_device, waveform, [0]) def test___digital_single_channel___read_waveform_one_sample___returns_waveform_with_one_sample( di_single_channel_timing_task: nidaqmx.Task, + sim_6363_device: nidaqmx.system.Device, ) -> None: waveform = di_single_channel_timing_task.read_waveform(1) assert isinstance(waveform, DigitalWaveform) assert _get_waveform_data(waveform) == _get_expected_data_for_line(1, 0) + _validate_waveform_signals(sim_6363_device, waveform, [0]) def test___digital_single_channel___read_waveform_many_sample___returns_waveform_with_many_samples( di_single_channel_timing_task: nidaqmx.Task, + sim_6363_device: nidaqmx.system.Device, ) -> None: samples_to_read = 10 @@ -38,10 +45,12 @@ def test___digital_single_channel___read_waveform_many_sample___returns_waveform assert isinstance(waveform, DigitalWaveform) assert _get_waveform_data(waveform) == _get_expected_data_for_line(samples_to_read, 0) + _validate_waveform_signals(sim_6363_device, waveform, [0]) def test___digital_single_channel___read_waveform_too_many_samples___returns_waveform_with_correct_number_of_samples( di_single_channel_timing_task: nidaqmx.Task, + sim_6363_device: nidaqmx.system.Device, ) -> None: samples_to_read = 100 samples_available = 50 @@ -51,6 +60,7 @@ def test___digital_single_channel___read_waveform_too_many_samples___returns_wav assert isinstance(waveform, DigitalWaveform) assert waveform.sample_count == samples_available assert _get_waveform_data(waveform) == _get_expected_data_for_line(samples_available, 0) + _validate_waveform_signals(sim_6363_device, waveform, [0]) def test___digital_single_channel___read_waveform_lines_and_port___returns_valid_waveform( @@ -61,28 +71,26 @@ def test___digital_single_channel___read_waveform_lines_and_port___returns_valid waveform = di_single_chan_lines_and_port_task.read_waveform(samples_to_read) - # Note, the data on the port's waveform is MSB instead of LSB because of bug AB#3178052 - # When that bug is fixed, these asserts should be updated - assert _get_waveform_data(waveform) == [0, 1025, 514, 1539, 260, 1285, 774, 1799, 128, 1153] + assert _get_waveform_port_data(waveform) == [ + 0b00000000000, + 0b10000000001, + 0b01000000010, + 0b11000000011, + 0b00100000100, + 0b10100000101, + 0b01100000110, + 0b11100000111, + 0b00000001000, + 0b10000001001, + ] assert waveform.sample_count == samples_to_read assert waveform.channel_name == di_single_chan_lines_and_port_task.di_channels[0].name - assert waveform._get_signal_names() == [ - sim_6363_device.di_lines[0].name, - sim_6363_device.di_lines[1].name, - sim_6363_device.di_lines[2].name, - sim_6363_device.di_lines[39].name, - sim_6363_device.di_lines[38].name, - sim_6363_device.di_lines[37].name, - sim_6363_device.di_lines[36].name, - sim_6363_device.di_lines[35].name, - sim_6363_device.di_lines[34].name, - sim_6363_device.di_lines[33].name, - sim_6363_device.di_lines[32].name, - ] + _validate_waveform_signals(sim_6363_device, waveform, [32, 33, 34, 35, 36, 37, 38, 39, 2, 1, 0]) def test___digital_multi_channel___read_waveform___returns_valid_waveforms( di_multi_channel_timing_task: nidaqmx.Task, + sim_6363_device: nidaqmx.system.Device, ) -> None: num_channels = di_multi_channel_timing_task.number_of_channels @@ -91,12 +99,14 @@ def test___digital_multi_channel___read_waveform___returns_valid_waveforms( assert isinstance(waveforms, list) assert len(waveforms) == num_channels assert all(isinstance(waveform, DigitalWaveform) for waveform in waveforms) - for chan_index, waveform in enumerate(waveforms): - assert _get_waveform_data(waveform) == _get_expected_data_for_line(50, chan_index) + for chan, waveform in enumerate(waveforms): + assert _get_waveform_data(waveform) == _get_expected_data_for_line(50, chan) + _validate_waveform_signals(sim_6363_device, waveform, [chan]) def test___digital_multi_channel___read_waveform_one_sample___returns_waveforms_with_single_sample( di_multi_channel_timing_task: nidaqmx.Task, + sim_6363_device: nidaqmx.system.Device, ) -> None: num_channels = di_multi_channel_timing_task.number_of_channels @@ -105,12 +115,14 @@ def test___digital_multi_channel___read_waveform_one_sample___returns_waveforms_ assert isinstance(waveforms, list) assert len(waveforms) == num_channels assert all(isinstance(waveform, DigitalWaveform) for waveform in waveforms) - for chan_index, waveform in enumerate(waveforms): - assert _get_waveform_data(waveform) == _get_expected_data_for_line(1, chan_index) + for chan, waveform in enumerate(waveforms): + assert _get_waveform_data(waveform) == _get_expected_data_for_line(1, chan) + _validate_waveform_signals(sim_6363_device, waveform, [chan]) def test___digital_multi_channel___read_waveform_many_samples___returns_waveforms_with_many_samples( di_multi_channel_timing_task: nidaqmx.Task, + sim_6363_device: nidaqmx.system.Device, ) -> None: num_channels = di_multi_channel_timing_task.number_of_channels samples_to_read = 10 @@ -120,12 +132,14 @@ def test___digital_multi_channel___read_waveform_many_samples___returns_waveform assert isinstance(waveforms, list) assert len(waveforms) == num_channels assert all(isinstance(waveform, DigitalWaveform) for waveform in waveforms) - for chan_index, waveform in enumerate(waveforms): - assert _get_waveform_data(waveform) == _get_expected_data_for_line(10, chan_index) + for chan, waveform in enumerate(waveforms): + assert _get_waveform_data(waveform) == _get_expected_data_for_line(10, chan) + _validate_waveform_signals(sim_6363_device, waveform, [chan]) def test___digital_multi_channel___read_waveform_too_many_samples___returns_waveforms_with_correct_number_of_samples( di_multi_channel_timing_task: nidaqmx.Task, + sim_6363_device: nidaqmx.system.Device, ) -> None: num_channels = di_multi_channel_timing_task.number_of_channels samples_to_read = 100 @@ -139,6 +153,7 @@ def test___digital_multi_channel___read_waveform_too_many_samples___returns_wave for chan, waveform in enumerate(waveforms): assert waveform.sample_count == samples_available assert _get_waveform_data(waveform) == _get_expected_data_for_line(samples_available, chan) + _validate_waveform_signals(sim_6363_device, waveform, [chan]) def test___digital_multi_channel___read_waveform_different_lines___returns_valid_waveforms( @@ -156,23 +171,13 @@ def test___digital_multi_channel___read_waveform_different_lines___returns_valid assert all(isinstance(waveform, DigitalWaveform) for waveform in waveforms) assert _get_waveform_data(waveforms[0]) == [0, 1, 0, 1, 0, 1, 0, 1, 0, 1] assert waveforms[0].channel_name == di_multi_chan_diff_lines_timing_task.di_channels[0].name - assert waveforms[0]._get_signal_names() == [ - sim_6363_device.di_lines[0].name, - ] + _validate_waveform_signals(sim_6363_device, waveforms[0], [0]) assert _get_waveform_data(waveforms[1]) == [0, 0, 1, 1, 2, 2, 3, 3, 0, 0] assert waveforms[1].channel_name == di_multi_chan_diff_lines_timing_task.di_channels[1].name - assert waveforms[1]._get_signal_names() == [ - sim_6363_device.di_lines[1].name, - sim_6363_device.di_lines[2].name, - ] + _validate_waveform_signals(sim_6363_device, waveforms[1], [2, 1]) assert _get_waveform_data(waveforms[2]) == [0, 0, 0, 0, 0, 0, 0, 0, 1, 1] assert waveforms[2].channel_name == di_multi_chan_diff_lines_timing_task.di_channels[2].name - assert waveforms[2]._get_signal_names() == [ - sim_6363_device.di_lines[3].name, - sim_6363_device.di_lines[4].name, - sim_6363_device.di_lines[5].name, - sim_6363_device.di_lines[6].name, - ] + _validate_waveform_signals(sim_6363_device, waveforms[2], [6, 5, 4, 3]) def test___digital_multi_channel___read_waveform_lines_and_port___returns_valid_waveforms( @@ -190,62 +195,45 @@ def test___digital_multi_channel___read_waveform_lines_and_port___returns_valid_ assert _get_waveform_data(waveforms[0]) == [0, 1, 0, 1, 0, 1, 0, 1, 0, 1] assert waveforms[0].sample_count == samples_to_read assert waveforms[0].channel_name == di_multi_chan_lines_and_port_task.di_channels[0].name - assert waveforms[0]._get_signal_names() == [ - sim_6363_device.di_lines[0].name, - ] + _validate_waveform_signals(sim_6363_device, waveforms[0], [0]) assert _get_waveform_data(waveforms[1]) == [0, 0, 1, 1, 2, 2, 3, 3, 0, 0] assert waveforms[1].sample_count == samples_to_read assert waveforms[1].channel_name == di_multi_chan_lines_and_port_task.di_channels[1].name - assert waveforms[1]._get_signal_names() == [ - sim_6363_device.di_lines[1].name, - sim_6363_device.di_lines[2].name, - ] + _validate_waveform_signals(sim_6363_device, waveforms[1], [2, 1]) assert _get_waveform_data(waveforms[2]) == [0, 0, 0, 0, 0, 0, 0, 0, 1, 1] assert waveforms[2].sample_count == samples_to_read assert waveforms[2].channel_name == di_multi_chan_lines_and_port_task.di_channels[2].name - assert waveforms[2]._get_signal_names() == [ - sim_6363_device.di_lines[3].name, - sim_6363_device.di_lines[4].name, - sim_6363_device.di_lines[5].name, - sim_6363_device.di_lines[6].name, - ] - # Note, the data on the port's waveform is MSB instead of LSB because of bug AB#3178052 - # When that bug is fixed, these asserts should be updated - assert _get_waveform_data(waveforms[3]) == [0, 128, 64, 192, 32, 160, 96, 224, 16, 144] + _validate_waveform_signals(sim_6363_device, waveforms[2], [6, 5, 4, 3]) + assert _get_waveform_port_data(waveforms[3]) == [0, 1, 2, 3, 4, 5, 6, 7, 8, 9] assert waveforms[3].sample_count == samples_to_read assert waveforms[3].channel_name == di_multi_chan_lines_and_port_task.di_channels[3].name - assert waveforms[3]._get_signal_names() == [ - sim_6363_device.di_lines[39].name, - sim_6363_device.di_lines[38].name, - sim_6363_device.di_lines[37].name, - sim_6363_device.di_lines[36].name, - sim_6363_device.di_lines[35].name, - sim_6363_device.di_lines[34].name, - sim_6363_device.di_lines[33].name, - sim_6363_device.di_lines[32].name, - ] + _validate_waveform_signals(sim_6363_device, waveforms[3], range(32, 40)) def test___digital_single_channel___read_waveform_read_all_available___returns_valid_waveform( di_single_channel_timing_task: nidaqmx.Task, + sim_6363_device: nidaqmx.system.Device, ) -> None: waveform = di_single_channel_timing_task.read_waveform(READ_ALL_AVAILABLE) assert isinstance(waveform, DigitalWaveform) assert waveform.sample_count == 50 assert _get_waveform_data(waveform) == _get_expected_data_for_line(waveform.sample_count, 0) + _validate_waveform_signals(sim_6363_device, waveform, [0]) def test___digital_multi_channel___read_waveform_read_all_available___returns_valid_waveforms( di_multi_channel_timing_task: nidaqmx.Task, + sim_6363_device: nidaqmx.system.Device, ) -> None: waveforms = di_multi_channel_timing_task.read_waveform(READ_ALL_AVAILABLE) assert isinstance(waveforms, list) assert len(waveforms) == di_multi_channel_timing_task.number_of_channels assert all(isinstance(waveform, DigitalWaveform) for waveform in waveforms) - for chan_index, waveform in enumerate(waveforms): + for chan, waveform in enumerate(waveforms): assert waveform.sample_count == 50 assert _get_waveform_data(waveform) == _get_expected_data_for_line( - waveform.sample_count, chan_index + waveform.sample_count, chan ) + _validate_waveform_signals(sim_6363_device, waveform, [chan]) diff --git a/tests/component/task/test_task_write_waveform_do.py b/tests/component/task/test_task_write_waveform_do.py index b07d991e..e1c5c5f3 100644 --- a/tests/component/task/test_task_write_waveform_do.py +++ b/tests/component/task/test_task_write_waveform_do.py @@ -14,7 +14,7 @@ _create_waveforms_for_mixed_lines, _get_digital_data, _get_waveform_data, - _get_waveform_data_msb, + _get_waveform_port_data, ) @@ -279,9 +279,7 @@ def test___task___write_waveform_port_uint8___outputs_match_final_values( assert samples_written == num_samples actual_value = di_port1_loopback_task.read() - assert ( - actual_value == _get_waveform_data_msb(waveform)[i - 1] - ) # TODO: AB#3178052 - change to _get_waveform_data() + assert actual_value == _get_waveform_port_data(waveform)[i - 1] def test___task___write_waveform_port_uint32___outputs_match_final_values( @@ -300,9 +298,7 @@ def test___task___write_waveform_port_uint32___outputs_match_final_values( assert samples_written == num_samples actual_value = di_port0_loopback_task.read() - assert ( - actual_value == _get_waveform_data_msb(waveform)[i - 1] - ) # TODO: AB#3178052 - change to _get_waveform_data() + assert actual_value == _get_waveform_port_data(waveform)[i - 1] @pytest.mark.disable_feature_toggle(WAVEFORM_SUPPORT) @@ -417,9 +413,9 @@ def test___task___write_waveforms_ports___outputs_match_final_values( actual_value = di_multi_channel_port_loopback_task.read() assert actual_value[0] != actual_value[1] assert actual_value == [ - _get_waveform_data_msb(waveforms[0])[-1], - _get_waveform_data_msb(waveforms[1])[-1], - ] # TODO: AB#3178052 - change to _get_waveform_data() + _get_waveform_port_data(waveforms[0])[-1], + _get_waveform_port_data(waveforms[1])[-1], + ] def test___task___write_waveforms_port_and_lines___outputs_match_final_values( @@ -442,9 +438,9 @@ def test___task___write_waveforms_port_and_lines___outputs_match_final_values( assert samples_written == num_samples actual_value = di_multi_channel_port_and_lines_loopback_task.read() assert actual_value == [ - _get_waveform_data_msb(waveforms[0])[-1], + _get_waveform_port_data(waveforms[0])[-1], _get_waveform_data(waveforms[1])[-1], - ] # TODO: AB#3178052 - change to _get_waveform_data() + ] def test___task___write_waveforms_with_non_contiguous_data___outputs_match_final_values(