Skip to content

Commit 6d9ffaa

Browse files
mikeprossernimikeprosserni
committed
task tests
1 parent 9a7883c commit 6d9ffaa

File tree

3 files changed

+252
-10
lines changed

3 files changed

+252
-10
lines changed

‎generated/nidaqmx/task/_task.py‎

Lines changed: 13 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -1392,11 +1392,7 @@ def write(self, data, auto_start=AUTO_START_UNSET, timeout=10.0):
13921392
Specifies the actual number of samples this method
13931393
successfully wrote.
13941394
"""
1395-
if isinstance(data, (AnalogWaveform, DigitalWaveform)) or (
1396-
isinstance(data, list)
1397-
and data
1398-
and all(isinstance(wf, (AnalogWaveform, DigitalWaveform)) for wf in data)
1399-
):
1395+
if self._is_waveform_data(data):
14001396
return self.write_waveform(data, auto_start, timeout)
14011397

14021398
channels_to_write = self.channels
@@ -1580,6 +1576,18 @@ def write(self, data, auto_start=AUTO_START_UNSET, timeout=10.0):
15801576
else:
15811577
self._raise_no_output_channels_error()
15821578

1579+
def _is_waveform_data(self, data):
1580+
"""Check if data is waveform data (single waveform or list of waveforms)."""
1581+
if isinstance(data, (AnalogWaveform, DigitalWaveform)):
1582+
return True
1583+
1584+
if not isinstance(data, list) or not data:
1585+
return False
1586+
1587+
return all(isinstance(wf, AnalogWaveform) for wf in data) or all(
1588+
isinstance(wf, DigitalWaveform) for wf in data
1589+
)
1590+
15831591
@requires_feature(WAVEFORM_SUPPORT)
15841592
def write_waveform(
15851593
self,

‎src/handwritten/task/_task.py‎

Lines changed: 13 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -1392,11 +1392,7 @@ def write(self, data, auto_start=AUTO_START_UNSET, timeout=10.0):
13921392
Specifies the actual number of samples this method
13931393
successfully wrote.
13941394
"""
1395-
if isinstance(data, (AnalogWaveform, DigitalWaveform)) or (
1396-
isinstance(data, list)
1397-
and data
1398-
and all(isinstance(wf, (AnalogWaveform, DigitalWaveform)) for wf in data)
1399-
):
1395+
if self._is_waveform_data(data):
14001396
return self.write_waveform(data, auto_start, timeout)
14011397

14021398
channels_to_write = self.channels
@@ -1580,6 +1576,18 @@ def write(self, data, auto_start=AUTO_START_UNSET, timeout=10.0):
15801576
else:
15811577
self._raise_no_output_channels_error()
15821578

1579+
def _is_waveform_data(self, data):
1580+
"""Check if data is waveform data (single waveform or list of waveforms)."""
1581+
if isinstance(data, (AnalogWaveform, DigitalWaveform)):
1582+
return True
1583+
1584+
if not isinstance(data, list) or not data:
1585+
return False
1586+
1587+
return all(isinstance(wf, AnalogWaveform) for wf in data) or all(
1588+
isinstance(wf, DigitalWaveform) for wf in data
1589+
)
1590+
15831591
@requires_feature(WAVEFORM_SUPPORT)
15841592
def write_waveform(
15851593
self,

‎tests/component/task/test_task_write_waveform_do.py‎

Lines changed: 226 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -10,6 +10,9 @@
1010
from tests.component._digital_utils import (
1111
_create_digital_waveform,
1212
_create_non_contiguous_digital_waveform,
13+
_create_waveform_for_line,
14+
_create_waveforms_for_mixed_lines,
15+
_get_digital_data,
1316
_get_waveform_data,
1417
_get_waveform_data_msb,
1518
)
@@ -315,3 +318,226 @@ def test___task___write_waveform_port_uint32___outputs_match_final_values(
315318
assert (
316319
actual_value == _get_waveform_data_msb(waveform)[i - 1]
317320
) # TODO: AB#3178052 - change to _get_waveform_data()
321+
322+
323+
@pytest.mark.disable_feature_toggle(WAVEFORM_SUPPORT)
324+
def test___task___write_waveforms_feature_disabled___raises_feature_not_supported_error(
325+
do_multi_channel_multi_line_task: nidaqmx.Task,
326+
) -> None:
327+
waveforms = [_create_digital_waveform(20), _create_digital_waveform(20)]
328+
329+
with pytest.raises(FeatureNotSupportedError) as exc_info:
330+
do_multi_channel_multi_line_task.write_waveform(waveforms)
331+
332+
error_message = exc_info.value.args[0]
333+
assert "WAVEFORM_SUPPORT feature is not supported" in error_message
334+
assert "NIDAQMX_ENABLE_WAVEFORM_SUPPORT" in error_message
335+
336+
337+
@pytest.mark.grpc_skip(reason="write_digital_waveform not implemented in GRPC")
338+
def test___task___write_waveforms_single_lines___outputs_match_final_values(
339+
do_multi_channel_multi_line_task: nidaqmx.Task,
340+
di_multi_line_loopback_task: nidaqmx.Task,
341+
) -> None:
342+
# Since digital outputs don't have built-in loopback channels like analog outputs,
343+
# we can only read back the last value. So to verify the whole signal, we must
344+
# write waveforms of increasing length and verify the final value each time.
345+
for i in range(1, 50):
346+
num_samples = i
347+
num_channels = 8
348+
waveforms = [_create_waveform_for_line(num_samples, chan) for chan in range(num_channels)]
349+
350+
samples_written = do_multi_channel_multi_line_task.write_waveform(waveforms)
351+
352+
assert samples_written == num_samples
353+
actual_value = di_multi_line_loopback_task.read()
354+
assert actual_value == i - 1
355+
assert actual_value == _get_digital_data(num_channels, num_samples)[i - 1]
356+
357+
358+
@pytest.mark.grpc_skip(reason="write_digital_waveform not implemented in GRPC")
359+
def test___task___write_waveforms_with_write___outputs_match_final_values(
360+
do_multi_channel_multi_line_task: nidaqmx.Task,
361+
di_multi_line_loopback_task: nidaqmx.Task,
362+
) -> None:
363+
# Since digital outputs don't have built-in loopback channels like analog outputs,
364+
# we can only read back the last value. So to verify the whole signal, we must
365+
# write waveforms of increasing length and verify the final value each time.
366+
for i in range(1, 50):
367+
num_samples = i
368+
num_channels = 8
369+
waveforms = [_create_waveform_for_line(num_samples, chan) for chan in range(num_channels)]
370+
371+
samples_written = do_multi_channel_multi_line_task.write(waveforms)
372+
373+
assert samples_written == num_samples
374+
actual_value = di_multi_line_loopback_task.read()
375+
assert actual_value == i - 1
376+
assert actual_value == _get_digital_data(num_channels, num_samples)[i - 1]
377+
378+
379+
@pytest.mark.grpc_skip(reason="write_digital_waveform not implemented in GRPC")
380+
def test___task___write_waveforms_with_auto_start___output_matches_final_value(
381+
do_multi_channel_multi_line_task_with_timing: nidaqmx.Task,
382+
di_multi_line_loopback_task: nidaqmx.Task,
383+
) -> None:
384+
num_samples = 5
385+
num_channels = 8
386+
waveforms = [_create_waveform_for_line(num_samples, chan) for chan in range(num_channels)]
387+
388+
samples_written = do_multi_channel_multi_line_task_with_timing.write_waveform(
389+
waveforms, auto_start=True
390+
)
391+
392+
assert samples_written == num_samples
393+
do_multi_channel_multi_line_task_with_timing.wait_until_done(timeout=2.0)
394+
actual_value = di_multi_line_loopback_task.read()
395+
assert actual_value == _get_digital_data(num_channels, num_samples)[-1]
396+
397+
398+
@pytest.mark.grpc_skip(reason="write_digital_waveform not implemented in GRPC")
399+
def test___task___write_waveforms_mixed_lines___outputs_match_final_values(
400+
do_multi_channel_mixed_line_task: nidaqmx.Task,
401+
di_multi_line_loopback_task: nidaqmx.Task,
402+
) -> None:
403+
# Since digital outputs don't have built-in loopback channels like analog outputs,
404+
# we can only read back the last value. So to verify the whole signal, we must
405+
# write waveforms of increasing length and verify the final value each time.
406+
for i in range(1, 10):
407+
num_samples = i
408+
num_channels = 8
409+
waveforms = _create_waveforms_for_mixed_lines(num_samples)
410+
411+
samples_written = do_multi_channel_mixed_line_task.write_waveform(waveforms)
412+
413+
assert samples_written == num_samples
414+
actual_value = di_multi_line_loopback_task.read()
415+
assert actual_value == _get_digital_data(num_channels, num_samples)[i - 1]
416+
417+
418+
@pytest.mark.grpc_skip(reason="write_digital_waveform not implemented in GRPC")
419+
def test___task___write_waveforms_ports___outputs_match_final_values(
420+
do_multi_channel_port_task: nidaqmx.Task,
421+
di_multi_channel_port_loopback_task: nidaqmx.Task,
422+
) -> None:
423+
# Since digital outputs don't have built-in loopback channels like analog outputs,
424+
# we can only read back the last value. So to verify the whole signal, we must
425+
# write waveforms of increasing length and verify the final value each time.
426+
for i in range(1, 50):
427+
num_samples = i
428+
num_lines = 8
429+
waveforms = [
430+
_create_digital_waveform(num_samples, num_lines),
431+
_create_digital_waveform(num_samples, num_lines, invert=True),
432+
]
433+
434+
samples_written = do_multi_channel_port_task.write_waveform(waveforms)
435+
436+
assert samples_written == num_samples
437+
actual_value = di_multi_channel_port_loopback_task.read()
438+
assert actual_value[0] != actual_value[1]
439+
assert actual_value == [
440+
_get_waveform_data_msb(waveforms[0])[-1],
441+
_get_waveform_data_msb(waveforms[1])[-1],
442+
] # TODO: AB#3178052 - change to _get_waveform_data()
443+
444+
445+
@pytest.mark.grpc_skip(reason="write_digital_waveform not implemented in GRPC")
446+
def test___task___write_waveforms_port_and_lines___outputs_match_final_values(
447+
do_multi_channel_port_and_lines_task: nidaqmx.Task,
448+
di_multi_channel_port_and_lines_loopback_task: nidaqmx.Task,
449+
) -> None:
450+
# Since digital outputs don't have built-in loopback channels like analog outputs,
451+
# we can only read back the last value. So to verify the whole signal, we must
452+
# write waveforms of increasing length and verify the final value each time.
453+
for i in range(1, 50):
454+
num_samples = i
455+
num_lines = 8
456+
waveforms = [
457+
_create_digital_waveform(num_samples, num_lines),
458+
_create_digital_waveform(num_samples, num_lines, invert=True),
459+
]
460+
461+
samples_written = do_multi_channel_port_and_lines_task.write_waveform(waveforms)
462+
463+
assert samples_written == num_samples
464+
actual_value = di_multi_channel_port_and_lines_loopback_task.read()
465+
assert actual_value == [
466+
_get_waveform_data_msb(waveforms[0])[-1],
467+
_get_waveform_data(waveforms[1])[-1],
468+
] # TODO: AB#3178052 - change to _get_waveform_data()
469+
470+
471+
@pytest.mark.grpc_skip(reason="write_digital_waveform not implemented in GRPC")
472+
def test___task___write_waveforms_with_non_contiguous_data___outputs_match_final_values(
473+
do_multi_channel_multi_line_task: nidaqmx.Task,
474+
di_multi_line_loopback_task: nidaqmx.Task,
475+
) -> None:
476+
# Since digital outputs don't have built-in loopback channels like analog outputs,
477+
# we can only read back the last value. So to verify the whole signal, we must
478+
# write waveforms of increasing length and verify the final value each time.
479+
for i in range(2, 50):
480+
num_samples = i
481+
num_lines = 8
482+
waveforms = [
483+
_create_non_contiguous_digital_waveform(num_samples, first_line=i, num_lines=1)
484+
for i in range(num_lines)
485+
]
486+
487+
samples_written = do_multi_channel_multi_line_task.write_waveform(waveforms)
488+
489+
assert samples_written == num_samples
490+
actual_value = di_multi_line_loopback_task.read()
491+
assert actual_value == _get_digital_data(num_lines, num_samples)[-1]
492+
493+
494+
@pytest.mark.grpc_skip(reason="write_digital_waveform not implemented in GRPC")
495+
def test___task___write_waveforms_with_different_sample_counts___raises_daq_error(
496+
do_multi_channel_port_task: nidaqmx.Task,
497+
) -> None:
498+
num_lines = 8
499+
waveforms = [
500+
_create_digital_waveform(10, num_lines),
501+
_create_digital_waveform(11, num_lines),
502+
]
503+
504+
with pytest.raises(DaqError) as exc_info:
505+
do_multi_channel_port_task.write_waveform(waveforms)
506+
507+
error_message = exc_info.value.args[0]
508+
assert "The waveforms must all have the same sample count." in error_message
509+
510+
511+
@pytest.mark.grpc_skip(reason="write_digital_waveform not implemented in GRPC")
512+
def test___task___write_waveforms_with_too_many___raises_daq_error(
513+
do_multi_channel_port_task: nidaqmx.Task,
514+
) -> None:
515+
num_lines = 8
516+
waveforms = [
517+
_create_digital_waveform(10, num_lines),
518+
_create_digital_waveform(10, num_lines),
519+
_create_digital_waveform(10, num_lines),
520+
]
521+
522+
with pytest.raises(DaqError) as exc_info:
523+
do_multi_channel_port_task.write_waveform(waveforms)
524+
525+
error_message = exc_info.value.args[0]
526+
assert "Write cannot be performed, because the number of channels" in error_message
527+
528+
529+
@pytest.mark.grpc_skip(reason="write_digital_waveform not implemented in GRPC")
530+
def test___task___write_waveforms_with_too_many_signals___raises_daq_error(
531+
do_multi_channel_port_task: nidaqmx.Task,
532+
) -> None:
533+
num_samples = 10
534+
waveforms = [
535+
_create_digital_waveform(num_samples, 8),
536+
_create_digital_waveform(num_samples, 10),
537+
]
538+
539+
with pytest.raises(DaqError) as exc_info:
540+
do_multi_channel_port_task.write_waveform(waveforms)
541+
542+
error_message = exc_info.value.args[0]
543+
assert "Specified read or write operation failed, because the number of lines" in error_message

0 commit comments

Comments
 (0)