From ae1b08ca58ab3b948f71c23975e04ca9b4bcb987 Mon Sep 17 00:00:00 2001 From: Mike Prosser <107578831+mikeprosserni@users.noreply.github.com> Date: Wed, 10 Dec 2025 15:24:26 -0600 Subject: [PATCH 1/2] Reverse Indices for Digital Waveform Signals (#222) * add _raw_index * line_index and line_names * more tests * fix new test * data_index and documentation * fix doc types * documentation improvements * documentation improvements * documentation improvements * clean up NI_LineNames in tests * remove DigitalWaveformFailure.data_index * test___signal_with_line_names___change_line_names_property___signal_returns_new_line_name and _DigitalWaveformExtendedProperties * signal_column_index * line lengths * italics for 'See "Signal index vs. signal column index"' notes * add bitorder to from_port, and default to the industry standard 'big' * rename to column_index * bitorder != sys.byteorder * add on_key_changed to ExtendedPropertyDictionary * change tests to big-endian * make on_key_changed private * cleanup * from typing_extensions import TypeAlias, to fix python 3.9 * fix pickling and copying issues with callbacks * change on_key_changed to a list of weak methods, so ExtendedPropertyDictionaries can be shared by waveforms --------- Co-authored-by: Mike Prosser --- src/nitypes/waveform/_digital/_port.py | 67 ++- src/nitypes/waveform/_digital/_signal.py | 35 +- .../waveform/_digital/_signal_collection.py | 16 +- src/nitypes/waveform/_digital/_waveform.py | 186 ++++++-- src/nitypes/waveform/_extended_properties.py | 26 +- tests/unit/waveform/test_digital_waveform.py | 426 +++++++++++++++--- .../waveform/test_digital_waveform_signal.py | 123 ++++- 7 files changed, 724 insertions(+), 155 deletions(-) diff --git a/src/nitypes/waveform/_digital/_port.py b/src/nitypes/waveform/_digital/_port.py index 7c596ab9..c02070b6 100644 --- a/src/nitypes/waveform/_digital/_port.py +++ b/src/nitypes/waveform/_digital/_port.py @@ -1,6 +1,8 @@ from __future__ import annotations +import sys from collections.abc import Sequence +from typing import Literal import numpy as np import numpy.typing as npt @@ -76,15 +78,28 @@ def _get_port_dtype(mask: int) -> np.dtype[AnyDigitalPort]: ) -def port_to_line_data(port_data: npt.NDArray[AnyDigitalPort], mask: int) -> npt.NDArray[np.uint8]: +def port_to_line_data( + port_data: npt.NDArray[AnyDigitalPort], mask: int, bitorder: Literal["big", "little"] = "big" +) -> npt.NDArray[np.uint8]: """Convert a 1D array of port data to a 2D array of line data, using the specified mask. >>> port_to_line_data(np.array([0,1,2,3], np.uint8), 0xFF) + array([[0, 0, 0, 0, 0, 0, 0, 0], + [0, 0, 0, 0, 0, 0, 0, 1], + [0, 0, 0, 0, 0, 0, 1, 0], + [0, 0, 0, 0, 0, 0, 1, 1]], dtype=uint8) + + >>> port_to_line_data(np.array([0,1,2,3], np.uint8), 0xFF, bitorder="little") array([[0, 0, 0, 0, 0, 0, 0, 0], [1, 0, 0, 0, 0, 0, 0, 0], [0, 1, 0, 0, 0, 0, 0, 0], [1, 1, 0, 0, 0, 0, 0, 0]], dtype=uint8) >>> port_to_line_data(np.array([0,1,2,3], np.uint8), 0x3) + array([[0, 0], + [0, 1], + [1, 0], + [1, 1]], dtype=uint8) + >>> port_to_line_data(np.array([0,1,2,3], np.uint8), 0x3, bitorder="little") array([[0, 0], [1, 0], [0, 1], @@ -97,29 +112,42 @@ def port_to_line_data(port_data: npt.NDArray[AnyDigitalPort], mask: int) -> npt. >>> port_to_line_data(np.array([0,1,2,3], np.uint8), 0) array([], shape=(4, 0), dtype=uint8) >>> port_to_line_data(np.array([0x12000000,0xFE000000], np.uint32), 0xFF000000) - array([[0, 1, 0, 0, 1, 0, 0, 0], - [0, 1, 1, 1, 1, 1, 1, 1]], dtype=uint8) + array([[0, 0, 0, 1, 0, 0, 1, 0], + [1, 1, 1, 1, 1, 1, 1, 0]], dtype=uint8) """ port_size = port_data.dtype.itemsize * 8 - line_data = np.unpackbits(port_data.view(np.uint8), bitorder="little") + # Convert to big-endian byte order to ensure MSB comes first when bitorder='big' + # For multi-byte types on little-endian systems, we need to byteswap + if bitorder != sys.byteorder and port_data.dtype.itemsize > 1: + port_data = port_data.byteswap() + + line_data = np.unpackbits(port_data.view(np.uint8), bitorder=bitorder) line_data = line_data.reshape(len(port_data), port_size) if mask == bit_mask(port_size): return line_data else: - return line_data[:, _mask_to_line_indices(mask)] + return line_data[:, _mask_to_column_indices(mask, port_size, bitorder)] -def _mask_to_line_indices(mask: int, /) -> list[int]: - """Return the line indices for the given mask. +def _mask_to_column_indices( + mask: int, port_size: int, bitorder: Literal["big", "little"], / +) -> list[int]: + """Return the column indices for the given mask. - >>> _mask_to_line_indices(0xF) + >>> _mask_to_column_indices(0xF, 8, "big") + [4, 5, 6, 7] + >>> _mask_to_column_indices(0x100, 16, "big") + [7] + >>> _mask_to_column_indices(0xDEADBEEF, 32, "big") + [0, 1, 3, 4, 5, 6, 8, 10, 12, 13, 15, 16, 18, 19, 20, 21, 22, 24, 25, 26, 28, 29, 30, 31] + >>> _mask_to_column_indices(0xF, 8, "little") [0, 1, 2, 3] - >>> _mask_to_line_indices(0x100) + >>> _mask_to_column_indices(0x100, 16, "little") [8] - >>> _mask_to_line_indices(0xDEADBEEF) + >>> _mask_to_column_indices(0xDEADBEEF, 32, "little") [0, 1, 2, 3, 5, 6, 7, 9, 10, 11, 12, 13, 15, 16, 18, 19, 21, 23, 25, 26, 27, 28, 30, 31] - >>> _mask_to_line_indices(-1) + >>> _mask_to_column_indices(-1, 8) Traceback (most recent call last): ... ValueError: The mask must be a non-negative integer. @@ -128,11 +156,18 @@ def _mask_to_line_indices(mask: int, /) -> list[int]: """ if mask < 0: raise ValueError("The mask must be a non-negative integer.\n\n" f"Mask: {mask}") - line_indices = [] - line_index = 0 + column_indices = [] + bit_position = 0 while mask != 0: if mask & 1: - line_indices.append(line_index) - line_index += 1 + if bitorder == "big": + column_indices.append(port_size - 1 - bit_position) + else: # little + column_indices.append(bit_position) + bit_position += 1 mask >>= 1 - return line_indices + + if bitorder == "big": + column_indices.reverse() + + return column_indices diff --git a/src/nitypes/waveform/_digital/_signal.py b/src/nitypes/waveform/_digital/_signal.py index cfbf4b19..8414c5cf 100644 --- a/src/nitypes/waveform/_digital/_signal.py +++ b/src/nitypes/waveform/_digital/_signal.py @@ -23,15 +23,22 @@ class DigitalWaveformSignal(Generic[TDigitalState]): collection, e.g. ``waveform.signals[0]`` or ``waveform.signals["Dev1/port0/line0"]``. """ - __slots__ = ["_owner", "_signal_index", "__weakref__"] + __slots__ = ["_owner", "_signal_index", "_column_index", "__weakref__"] _owner: DigitalWaveform[TDigitalState] + _column_index: int _signal_index: int - def __init__(self, owner: DigitalWaveform[TDigitalState], signal_index: SupportsIndex) -> None: + def __init__( + self, + owner: DigitalWaveform[TDigitalState], + signal_index: SupportsIndex, + column_index: SupportsIndex, + ) -> None: """Initialize a new digital waveform signal.""" self._owner = owner self._signal_index = arg_to_uint("signal index", signal_index) + self._column_index = arg_to_uint("column index", column_index) @property def owner(self) -> DigitalWaveform[TDigitalState]: @@ -40,22 +47,36 @@ def owner(self) -> DigitalWaveform[TDigitalState]: @property def signal_index(self) -> int: - """The signal index.""" + """The signal's position in the DigitalWaveform.signals collection (0-based).""" return self._signal_index + @property + def column_index(self) -> int: + """The signal's position in the DigitalWaveform.data array's second dimension (0-based). + + This index is used to access the signal's data within the waveform's data array: + `waveform.data[:, column_index]`. + + Note: The column_index is reversed compared to the signal_index. column_index 0 (the + leftmost column) corresponds to the highest signal_index and highest line number. The + highest column_index (the rightmost column) corresponds to signal_index 0 and line 0. This + matches industry conventions where line 0 is the LSB and appears as the rightmost bit. + """ + return self._column_index + @property def data(self) -> npt.NDArray[TDigitalState]: """The signal data, indexed by sample.""" - return self._owner.data[:, self._signal_index] + return self._owner.data[:, self._column_index] @property def name(self) -> str: """The signal name.""" - return self._owner._get_signal_name(self._signal_index) + return self._owner._get_line_name(self._column_index) @name.setter def name(self, value: str) -> None: - self._owner._set_signal_name(self._signal_index, value) + self._owner._set_line_name(self._column_index, value) def __eq__(self, value: object, /) -> bool: """Return self==value.""" @@ -66,7 +87,7 @@ def __eq__(self, value: object, /) -> bool: def __reduce__(self) -> tuple[Any, ...]: """Return object state for pickling.""" - ctor_args = (self._owner, self._signal_index) + ctor_args = (self._owner, self._signal_index, self._column_index) return (self.__class__, ctor_args) def __repr__(self) -> str: diff --git a/src/nitypes/waveform/_digital/_signal_collection.py b/src/nitypes/waveform/_digital/_signal_collection.py index beccd2f8..4ef213d6 100644 --- a/src/nitypes/waveform/_digital/_signal_collection.py +++ b/src/nitypes/waveform/_digital/_signal_collection.py @@ -49,21 +49,25 @@ def __getitem__( self, index: int | str | slice ) -> DigitalWaveformSignal[TDigitalState] | Sequence[DigitalWaveformSignal[TDigitalState]]: """Get self[index].""" - if isinstance(index, int): + if isinstance(index, int): # index is the signal index if index < 0: index += len(self._signals) value = self._signals[index] if value is None: - value = self._signals[index] = DigitalWaveformSignal(self._owner, index) + column_index = self._owner._reverse_index(index) + value = self._signals[index] = DigitalWaveformSignal( + self._owner, index, column_index + ) return value - elif isinstance(index, str): - signal_names = self._owner._get_signal_names() + elif isinstance(index, str): # index is the line name + line_names = self._owner._get_line_names() try: - signal_index = signal_names.index(index) + column_index = line_names.index(index) except ValueError: raise IndexError(index) + signal_index = self._owner._reverse_index(column_index) return self[signal_index] - elif isinstance(index, slice): + elif isinstance(index, slice): # index is a slice of signal indices return [self[i] for i in range(*index.indices(len(self)))] else: raise invalid_arg_type("index", "int or str", index) diff --git a/src/nitypes/waveform/_digital/_waveform.py b/src/nitypes/waveform/_digital/_waveform.py index 91550a63..a7f8d755 100644 --- a/src/nitypes/waveform/_digital/_waveform.py +++ b/src/nitypes/waveform/_digital/_waveform.py @@ -2,9 +2,10 @@ import datetime as dt import sys +import weakref from collections.abc import Mapping, Sequence from dataclasses import dataclass -from typing import TYPE_CHECKING, Any, Generic, SupportsIndex, overload +from typing import TYPE_CHECKING, Any, Generic, Literal, SupportsIndex, overload import hightime as ht import numpy as np @@ -106,7 +107,8 @@ class DigitalWaveform(Generic[TDigitalState]): To construct a digital waveform from a NumPy array of line data, use the :any:`DigitalWaveform.from_lines` method. Each array element represents a digital state, such as 1 for "on" or 0 for "off". The line data should be in a 1D array indexed by sample or a 2D array - indexed by (sample, signal). The digital waveform displays the line data as a 2D array. + indexed by (sample, signal). *(Note, signal indices are reversed! See "Signal index vs. column index" + below for details.)* The digital waveform displays the line data as a 2D array. >>> import numpy as np >>> DigitalWaveform.from_lines(np.array([0, 1, 0], np.uint8)) @@ -123,30 +125,30 @@ class DigitalWaveform(Generic[TDigitalState]): To construct a digital waveform from a NumPy array of port data, use the :any:`DigitalWaveform.from_port` method. Each element of the port data array represents a digital sample taken over a port of signals. Each bit in the sample is a signal value, either 1 for "on" or - 0 for "off". The least significant bit of the sample is placed at signal index 0 of the - DigitalWaveform. + 0 for "off". *(Note, signal indices are reversed! See "Signal index vs. column index" below for + details.)* >>> DigitalWaveform.from_port(np.array([0, 1, 2, 3], np.uint8)) # doctest: +NORMALIZE_WHITESPACE nitypes.waveform.DigitalWaveform(4, 8, data=array([[0, 0, 0, 0, 0, 0, 0, 0], - [1, 0, 0, 0, 0, 0, 0, 0], [0, 1, 0, 0, 0, 0, 0, 0], [1, 1, 0, 0, 0, 0, 0, 0]], dtype=uint8)) + [0, 0, 0, 0, 0, 0, 0, 1], [0, 0, 0, 0, 0, 0, 1, 0], [0, 0, 0, 0, 0, 0, 1, 1]], dtype=uint8)) You can use a mask to specify which lines in the port to include in the waveform. >>> DigitalWaveform.from_port(np.array([0, 1, 2, 3], np.uint8), 0x3) - nitypes.waveform.DigitalWaveform(4, 2, data=array([[0, 0], [1, 0], [0, 1], [1, 1]], dtype=uint8)) + nitypes.waveform.DigitalWaveform(4, 2, data=array([[0, 0], [0, 1], [1, 0], [1, 1]], dtype=uint8)) You can also use a non-NumPy sequence such as a list, but you must specify a mask so the waveform knows how many bits are in each list element. >>> DigitalWaveform.from_port([0, 1, 2, 3], 0x3) - nitypes.waveform.DigitalWaveform(4, 2, data=array([[0, 0], [1, 0], [0, 1], [1, 1]], dtype=uint8)) + nitypes.waveform.DigitalWaveform(4, 2, data=array([[0, 0], [0, 1], [1, 0], [1, 1]], dtype=uint8)) The 2D version, :any:`DigitalWaveform.from_ports`, returns multiple waveforms, one for each row of data in the array or nested sequence. >>> nested_list = [[0, 1, 2, 3], [3, 0, 3, 0]] >>> DigitalWaveform.from_ports(nested_list, [0x3, 0x3]) # doctest: +NORMALIZE_WHITESPACE - [nitypes.waveform.DigitalWaveform(4, 2, data=array([[0, 0], [1, 0], [0, 1], [1, 1]], dtype=uint8)), + [nitypes.waveform.DigitalWaveform(4, 2, data=array([[0, 0], [0, 1], [1, 0], [1, 1]], dtype=uint8)), nitypes.waveform.DigitalWaveform(4, 2, data=array([[1, 1], [0, 0], [1, 1], [0, 0]], dtype=uint8))] Digital signals @@ -165,6 +167,40 @@ class DigitalWaveform(Generic[TDigitalState]): >>> wfm.signals[0].data array([0, 1, 0, 1], dtype=uint8) + Signal index vs. column index + ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ + + Each :class:`DigitalWaveformSignal` has two index properties: + + * :attr:`DigitalWaveformSignal.signal_index` - The position in the :attr:`DigitalWaveform.signals` + collection (0-based from the first signal). signal_index 0 is the rightmost column in the data. + * :attr:`DigitalWaveformSignal.column_index` - The column in the :attr:`DigitalWaveform.data` + array, e.g. `waveform.data[:, column_index]`. column_index 0 is the leftmost column in the data. + + These indices are reversed with respect to each other. signal_index 0 (line 0) corresponds to + the highest column_index, and the highest signal_index (the highest line) corresponds to + column_index 0. This ordering follows industry conventions where line 0 is the least + significant bit and appears last (in the rightmost column) of the data array. + + >>> wfm = DigitalWaveform.from_port([0, 1, 2, 3], 0x7) # 3 signals + >>> wfm.data + array([[0, 0, 0], + [0, 0, 1], + [0, 1, 0], + [0, 1, 1]], dtype=uint8) + >>> wfm.signals[0].signal_index + 0 + >>> wfm.signals[0].column_index + 2 + >>> wfm.signals[0].data + array([0, 1, 0, 1], dtype=uint8) + >>> wfm.signals[2].signal_index + 2 + >>> wfm.signals[2].column_index + 0 + >>> wfm.signals[2].data + array([0, 0, 0, 0], dtype=uint8) + Digital signal names ^^^^^^^^^^^^^^^^^^^^ @@ -172,20 +208,25 @@ class DigitalWaveform(Generic[TDigitalState]): >>> wfm.signals[0].name = "port0/line0" >>> wfm.signals[1].name = "port0/line1" + >>> wfm.signals[2].name = "port0/line2" >>> wfm.signals[0].name 'port0/line0' >>> wfm.signals[0] nitypes.waveform.DigitalWaveformSignal(name='port0/line0', data=array([0, 1, 0, 1], dtype=uint8)) The signal names are stored in the ``NI_LineNames`` extended property on the digital waveform. + Note that the order of the names in the string follows column_index order (highest line number + first), which is reversed compared to signal_index order (lowest line first). This means line 0 + (signal_index 0) appears last in the NI_LineNames string. This matches industry conventions + where line 0 appears in the rightmost column of the data array. >>> wfm.extended_properties["NI_LineNames"] - 'port0/line0, port0/line1' + 'port0/line2, port0/line1, port0/line0' When creating a digital waveform, you can directly set the ``NI_LineNames`` extended property. >>> wfm = DigitalWaveform.from_port([2, 4], 0x7, - ... extended_properties={"NI_LineNames": "Dev1/port1/line4, Dev1/port1/line5, Dev1/port1/line6"}) + ... extended_properties={"NI_LineNames": "Dev1/port1/line6, Dev1/port1/line5, Dev1/port1/line4"}) >>> wfm.signals[0] nitypes.waveform.DigitalWaveformSignal(name='Dev1/port1/line4', data=array([0, 0], dtype=uint8)) >>> wfm.signals[1] @@ -215,7 +256,7 @@ class DigitalWaveform(Generic[TDigitalState]): objects, which indicate the location of each test failure. Here is an example. The expected waveform counts in binary using ``COMPARE_LOW`` (``L``) and - ``COMPARE_HIGH`` (``H``), but signal 1 of the actual waveform is stuck high. + ``COMPARE_HIGH`` (``H``), but signal 0 of the actual waveform is stuck high. >>> actual = DigitalWaveform.from_lines([[0, 1], [1, 1], [0, 1], [1, 1]]) >>> expected = DigitalWaveform.from_lines([[DigitalState.COMPARE_LOW, DigitalState.COMPARE_LOW], @@ -232,10 +273,10 @@ class DigitalWaveform(Generic[TDigitalState]): and the digital state from the actual and expected waveforms: >>> result.failures[0] # doctest: +NORMALIZE_WHITESPACE - DigitalWaveformFailure(sample_index=0, expected_sample_index=0, signal_index=1, + DigitalWaveformFailure(sample_index=0, expected_sample_index=0, signal_index=0, column_index=1, actual_state=, expected_state=) >>> result.failures[1] # doctest: +NORMALIZE_WHITESPACE - DigitalWaveformFailure(sample_index=1, expected_sample_index=1, signal_index=1, + DigitalWaveformFailure(sample_index=1, expected_sample_index=1, signal_index=0, column_index=1, actual_state=, expected_state=) Timing information @@ -312,6 +353,10 @@ def from_lines( by (sample, signal). The line data may also use digital state values from the :class:`DigitalState` enum. + Note that signal indices are reversed with respect to this array's column indices. + The first column in each sample corresponds to the highest line number and highest signal + index. The last column in each sample corresponds to line 0 and signal index 0. + Args: array: The line data as a one or two-dimensional array or a sequence. dtype: The NumPy data type for the waveform data. @@ -357,6 +402,7 @@ def from_port( mask: SupportsIndex | None = ..., dtype: None = ..., *, + bitorder: Literal["big", "little"] = ..., start_index: SupportsIndex | None = ..., sample_count: SupportsIndex | None = ..., extended_properties: Mapping[str, ExtendedPropertyValue] | None = ..., @@ -374,6 +420,7 @@ def from_port( | np.dtype[TOtherDigitalState] ) = ..., *, + bitorder: Literal["big", "little"] = ..., start_index: SupportsIndex | None = ..., sample_count: SupportsIndex | None = ..., extended_properties: Mapping[str, ExtendedPropertyValue] | None = ..., @@ -388,6 +435,7 @@ def from_port( mask: SupportsIndex | None = ..., dtype: npt.DTypeLike = ..., *, + bitorder: Literal["big", "little"] = ..., start_index: SupportsIndex | None = ..., sample_count: SupportsIndex | None = ..., extended_properties: Mapping[str, ExtendedPropertyValue] | None = ..., @@ -401,6 +449,7 @@ def from_port( mask: SupportsIndex | None = None, dtype: npt.DTypeLike = None, *, + bitorder: Literal["big", "little"] = "big", start_index: SupportsIndex | None = 0, sample_count: SupportsIndex | None = None, extended_properties: Mapping[str, ExtendedPropertyValue] | None = None, @@ -408,12 +457,24 @@ def from_port( ) -> DigitalWaveform[Any]: """Construct a waveform from a one-dimensional array or sequence of port data. - This method allocates a new array in order to convert the port data to line data. + This method allocates a new array in order to convert the port data (integers) to line data + (bits). Each element of the port data array represents a digital sample taken over a port of signals. Each bit in the sample represents a digital state, either 1 for "on" or 0 for - "off". The least significant bit of the sample is placed at signal index 0 of the - DigitalWaveform. + "off". + + When bitorder='big' (default), the integers in the samples are big-endian. The most + significant bit of each integer will be placed in the first column of the data array + (corresponding to the highest line number and highest signal index). The least significant + bit will be placed in the last column of the data array (corresponding to line 0 and signal + index 0). + + When bitorder='little', the integers in the samples are little-endian. The least + significant bit of each integer will be placed in the first column of the data array + (corresponding to the highest line number and highest signal index). The most significant + bit will be placed in the last column of the data array (corresponding to line 0 and signal + index 0). If the input array is not a NumPy array, you must specify the mask. @@ -421,6 +482,8 @@ def from_port( array: The port data as a one-dimensional array or a sequence. mask: A bitmask specifying which lines to include in the waveform. dtype: The NumPy data type for the waveform (line) data. + bitorder: The bit ordering to use when unpacking port data ('big' or 'little'). + Defaults to 'big'. start_index: The sample index at which the waveform data begins. sample_count: The number of samples in the waveform. extended_properties: The extended properties of the waveform. @@ -460,7 +523,7 @@ def from_port( port_dtype = get_port_dtype(mask) port_data = _np_asarray(array, port_dtype) - line_data = port_to_line_data(port_data, mask) + line_data = port_to_line_data(port_data, mask, bitorder) if line_data.dtype != dtype: line_data = line_data.view(dtype) @@ -481,6 +544,7 @@ def from_ports( masks: Sequence[SupportsIndex] | None = ..., dtype: None = ..., *, + bitorder: Literal["big", "little"] = ..., start_index: SupportsIndex | None = ..., sample_count: SupportsIndex | None = ..., extended_properties: Mapping[str, ExtendedPropertyValue] | None = ..., @@ -498,6 +562,7 @@ def from_ports( | np.dtype[TOtherDigitalState] ) = ..., *, + bitorder: Literal["big", "little"] = ..., start_index: SupportsIndex | None = ..., sample_count: SupportsIndex | None = ..., extended_properties: Mapping[str, ExtendedPropertyValue] | None = ..., @@ -512,6 +577,7 @@ def from_ports( masks: Sequence[SupportsIndex] | None = ..., dtype: npt.DTypeLike = ..., *, + bitorder: Literal["big", "little"] = ..., start_index: SupportsIndex | None = ..., sample_count: SupportsIndex | None = ..., extended_properties: Mapping[str, ExtendedPropertyValue] | None = ..., @@ -525,6 +591,7 @@ def from_ports( masks: Sequence[SupportsIndex] | None = None, dtype: npt.DTypeLike = None, *, + bitorder: Literal["big", "little"] = "big", start_index: SupportsIndex | None = 0, sample_count: SupportsIndex | None = None, extended_properties: Mapping[str, ExtendedPropertyValue] | None = None, @@ -536,9 +603,19 @@ def from_ports( Each row of the port data array corresponds to a resulting DigitalWaveform. Each element of the port data array represents a digital sample taken over a port of signals. Each bit in - the sample is represents a digital state, either 1 for "on" or 0 for "off". The least - significant bit of the sample is placed at signal index 0 of the corresponding - DigitalWaveform. + the sample represents a digital state, either 1 for "on" or 0 for "off". + + When bitorder='big' (default), the integers in the samples are big-endian. The most + significant bit of each integer will be placed in the first column of the data array + (corresponding to the highest line number and highest signal index). The least significant + bit will be placed in the last column of the data array (corresponding to line 0 and signal + index 0). + + When bitorder='little', the integers in the samples are little-endian. The least + significant bit of each integer will be placed in the first column of the data array + (corresponding to the highest line number and highest signal index). The most significant + bit will be placed in the last column of the data array (corresponding to line 0 and signal + index 0). If the input array is not a NumPy array, you must specify the masks. @@ -547,6 +624,8 @@ def from_ports( masks: A sequence of bitmasks specifying which lines from each port to include in the corresponding waveform. dtype: The NumPy data type for the waveform (line) data. + bitorder: The bit ordering to use when unpacking port data ('big' or 'little'). + Defaults to 'big'. start_index: The sample index at which the waveform data begins. sample_count: The number of samples in the waveform. extended_properties: The extended properties of the waveform. @@ -593,7 +672,9 @@ def from_ports( port_data = _np_asarray(array, port_dtype) waveforms = [] for port_index in range(port_data.shape[0]): - line_data = port_to_line_data(port_data[port_index, :], validated_masks[port_index]) + line_data = port_to_line_data( + port_data[port_index, :], validated_masks[port_index], bitorder + ) if line_data.dtype != dtype: line_data = line_data.view(dtype) @@ -617,7 +698,7 @@ def from_ports( "_extended_properties", "_timing", "_signals", - "_signal_names", + "_line_names", "__weakref__", ] @@ -628,7 +709,7 @@ def from_ports( _extended_properties: ExtendedPropertyDictionary _timing: Timing[AnyDateTime, AnyTimeDelta, AnyTimeDelta] _signals: DigitalWaveformSignalCollection[TDigitalState] | None - _signal_names: list[str] | None + _line_names: list[str] | None # If neither dtype nor data is specified, _TData defaults to np.uint8. @overload @@ -757,13 +838,20 @@ def __init__( ): extended_properties = ExtendedPropertyDictionary(extended_properties) self._extended_properties = extended_properties + self._extended_properties._on_key_changed.append( + weakref.WeakMethod(self._on_extended_property_changed) + ) if timing is None: timing = Timing.empty self._timing = timing self._signals = None - self._signal_names = None + self._line_names = None + + def _on_extended_property_changed(self, key: str) -> None: + if key == LINE_NAMES: + self._line_names = None def _init_with_new_array( self, @@ -976,26 +1064,24 @@ def channel_name(self, value: str) -> None: raise invalid_arg_type("channel name", "str", value) self._extended_properties[CHANNEL_NAME] = value - def _get_signal_names(self) -> list[str]: - # Lazily allocate self._signal_names if the application needs it. - signal_names = self._signal_names - if signal_names is None: - signal_names_str = self._extended_properties.get(LINE_NAMES, "") - assert isinstance(signal_names_str, str) - signal_names = self._signal_names = [ - name.strip() for name in signal_names_str.split(",") - ] - if len(signal_names) < self.signal_count: - signal_names.extend([""] * (self.signal_count - len(signal_names))) - return signal_names - - def _get_signal_name(self, signal_index: int) -> str: - return self._get_signal_names()[signal_index] - - def _set_signal_name(self, signal_index: int, value: str) -> None: - signal_names = self._get_signal_names() - signal_names[signal_index] = value - self._extended_properties[LINE_NAMES] = ", ".join(signal_names) + def _get_line_names(self) -> list[str]: + # Lazily allocate self._line_names if the application needs it. + line_names = self._line_names + if line_names is None: + line_names_str = self._extended_properties.get(LINE_NAMES, "") + assert isinstance(line_names_str, str) + line_names = self._line_names = [name.strip() for name in line_names_str.split(",")] + if len(line_names) < self.signal_count: + line_names.extend([""] * (self.signal_count - len(line_names))) + return line_names + + def _get_line_name(self, column_index: int) -> str: + return self._get_line_names()[column_index] + + def _set_line_name(self, column_index: int, value: str) -> None: + line_names = self._get_line_names() + line_names[column_index] = value + self._extended_properties[LINE_NAMES] = ", ".join(line_names) def _set_timing(self, value: Timing[AnyDateTime, AnyTimeDelta, AnyTimeDelta]) -> None: if self._timing is not value: @@ -1247,10 +1333,11 @@ def test( failures = [] for _ in range(sample_count): - for signal_index in range(self.signal_count): - actual_state = DigitalState(self.data[start_sample, signal_index]) + for column_index in range(self.signal_count): + signal_index = self._reverse_index(column_index) + actual_state = DigitalState(self.data[start_sample, column_index]) expected_state = DigitalState( - expected_waveform.data[expected_start_sample, signal_index] + expected_waveform.data[expected_start_sample, column_index] ) if DigitalState.test(actual_state, expected_state): failures.append( @@ -1267,6 +1354,11 @@ def test( return DigitalWaveformTestResult(failures) + def _reverse_index(self, index: int) -> int: + """Convert a signal_index to a column_index, or vice versa.""" + assert 0 <= index < self.signal_count + return self.signal_count - 1 - index + def __eq__(self, value: object, /) -> bool: """Return self==value.""" if not isinstance(value, self.__class__): diff --git a/src/nitypes/waveform/_extended_properties.py b/src/nitypes/waveform/_extended_properties.py index 3703f599..b7b480a0 100644 --- a/src/nitypes/waveform/_extended_properties.py +++ b/src/nitypes/waveform/_extended_properties.py @@ -1,7 +1,11 @@ from __future__ import annotations import operator -from collections.abc import Iterator, Mapping, MutableMapping +import weakref +from collections.abc import Callable, Iterator, Mapping, MutableMapping +from typing import TYPE_CHECKING + +from typing_extensions import TypeAlias from nitypes.waveform.typing import ExtendedPropertyValue @@ -10,6 +14,9 @@ LINE_NAMES = "NI_LineNames" UNIT_DESCRIPTION = "NI_UnitDescription" +if TYPE_CHECKING: + OnKeyChangedCallback: TypeAlias = Callable[[str], None] + class ExtendedPropertyDictionary(MutableMapping[str, ExtendedPropertyValue]): """A dictionary of extended properties. @@ -19,11 +26,12 @@ class ExtendedPropertyDictionary(MutableMapping[str, ExtendedPropertyValue]): over the network or write it to a TDMS file. """ - __slots__ = ["_properties"] + __slots__ = ["_properties", "_on_key_changed"] def __init__(self, properties: Mapping[str, ExtendedPropertyValue] | None = None, /) -> None: """Initialize a new ExtendedPropertyDictionary.""" self._properties: dict[str, ExtendedPropertyValue] = {} + self._on_key_changed: list[weakref.ref[OnKeyChangedCallback]] = [] if properties is not None: self._properties.update(properties) @@ -46,15 +54,29 @@ def __getitem__(self, key: str, /) -> ExtendedPropertyValue: def __setitem__(self, key: str, value: ExtendedPropertyValue, /) -> None: """Set self[key] to value.""" operator.setitem(self._properties, key, value) + self._notify_on_key_changed(key) def __delitem__(self, key: str, /) -> None: """Delete self[key].""" operator.delitem(self._properties, key) + self._notify_on_key_changed(key) + + def _notify_on_key_changed(self, key: str) -> None: + for callback_ref in self._on_key_changed: + callback = callback_ref() + if callback: + callback(key) def _merge(self, other: ExtendedPropertyDictionary) -> None: for key, value in other.items(): self._properties.setdefault(key, value) + def __reduce__( + self, + ) -> tuple[type[ExtendedPropertyDictionary], tuple[dict[str, ExtendedPropertyValue]]]: + """Return object state for pickling, excluding the callback.""" + return (self.__class__, (self._properties,)) + def __repr__(self) -> str: """Return repr(self).""" return f"{self.__class__.__module__}.{self.__class__.__name__}({self._properties!r})" diff --git a/tests/unit/waveform/test_digital_waveform.py b/tests/unit/waveform/test_digital_waveform.py index 2d69073d..70932c5f 100644 --- a/tests/unit/waveform/test_digital_waveform.py +++ b/tests/unit/waveform/test_digital_waveform.py @@ -222,10 +222,10 @@ def test___ndarray_2d___from_lines___creates_waveform_with_multi_signal() -> Non ############################################################################### # from_port ############################################################################### -def test___uint8_ndarray___from_port___creates_waveform_with_8_lines() -> None: +def test___uint8_ndarray___from_port_bitorder_little___creates_waveform_with_8_lines() -> None: array = np.array([0, 1, 2, 3, 0xFF, 0x12, 0x34], np.uint8) - waveform = DigitalWaveform.from_port(array) + waveform = DigitalWaveform.from_port(array, bitorder="little") assert waveform.sample_count == 7 assert waveform.signal_count == 8 @@ -240,10 +240,10 @@ def test___uint8_ndarray___from_port___creates_waveform_with_8_lines() -> None: ] -def test___uint16_ndarray___from_port___creates_waveform_with_16_lines() -> None: +def test___uint16_ndarray___from_port_bitorder_little___creates_waveform_with_16_lines() -> None: array = np.array([0, 1, 2, 3, 0xFFFF, 0x1234, 0x5678], np.uint16) - waveform = DigitalWaveform.from_port(array) + waveform = DigitalWaveform.from_port(array, bitorder="little") assert waveform.sample_count == 7 assert waveform.signal_count == 16 @@ -258,10 +258,12 @@ def test___uint16_ndarray___from_port___creates_waveform_with_16_lines() -> None ] -def test___int_list_and_mask___from_port___creates_waveform_with_masked_lines() -> None: +def test___int_list_and_mask___from_port_bitorder_little___creates_waveform_with_masked_lines() -> ( + None +): array = [0, 1, 2, 3, 0xFFFF, 0x1234, 0x5678] - waveform = DigitalWaveform.from_port(array, mask=0xFFFF) + waveform = DigitalWaveform.from_port(array, mask=0xFFFF, bitorder="little") assert waveform.sample_count == 7 assert waveform.signal_count == 16 @@ -276,11 +278,11 @@ def test___int_list_and_mask___from_port___creates_waveform_with_masked_lines() ] -def test___mask___from_port___creates_waveform_with_masked_lines() -> None: +def test___mask___from_port_bitorder_little___creates_waveform_with_masked_lines() -> None: array = np.array([0xFF, 0x12, 0x34], np.uint8) - waveform_lo = DigitalWaveform.from_port(array, 0x0F) - waveform_hi = DigitalWaveform.from_port(array, 0xF0) + waveform_lo = DigitalWaveform.from_port(array, 0x0F, bitorder="little") + waveform_hi = DigitalWaveform.from_port(array, 0xF0, bitorder="little") assert waveform_lo.sample_count == 3 assert waveform_lo.signal_count == 4 @@ -298,10 +300,10 @@ def test___mask___from_port___creates_waveform_with_masked_lines() -> None: ] -def test___bool_dtype___from_port___creates_waveform_with_bool_dtype() -> None: +def test___bool_dtype___from_port_bitorder_little___creates_waveform_with_bool_dtype() -> None: array = np.array([0, 1, 2, 3, 0xFF, 0x12, 0x34], np.uint8) - waveform = DigitalWaveform.from_port(array, dtype=_np_bool) + waveform = DigitalWaveform.from_port(array, dtype=_np_bool, bitorder="little") assert waveform.sample_count == 7 assert waveform.signal_count == 8 @@ -316,10 +318,10 @@ def test___bool_dtype___from_port___creates_waveform_with_bool_dtype() -> None: ] -def test___array_subset___from_port___creates_waveform_with_array_subset() -> None: +def test___array_subset___from_port_bitorder_little___creates_waveform_with_array_subset() -> None: array = np.array([0, 1, 2, 3, 0xFF, 0x12, 0x34], np.uint8) - waveform = DigitalWaveform.from_port(array, start_index=2, sample_count=4) + waveform = DigitalWaveform.from_port(array, start_index=2, sample_count=4, bitorder="little") assert waveform.start_index == 2 assert waveform.sample_count == 4 @@ -332,13 +334,123 @@ def test___array_subset___from_port___creates_waveform_with_array_subset() -> No ] +def test___uint8_ndarray___from_port___creates_waveform_with_8_lines() -> None: + array = np.array([0, 1, 2, 3, 0xFF, 0x12, 0x34], np.uint8) + + waveform = DigitalWaveform.from_port(array) + + assert waveform.sample_count == 7 + assert waveform.signal_count == 8 + assert waveform.data.tolist() == [ + [0, 0, 0, 0, 0, 0, 0, 0], + [0, 0, 0, 0, 0, 0, 0, 1], + [0, 0, 0, 0, 0, 0, 1, 0], + [0, 0, 0, 0, 0, 0, 1, 1], + [1, 1, 1, 1, 1, 1, 1, 1], + [0, 0, 0, 1, 0, 0, 1, 0], + [0, 0, 1, 1, 0, 1, 0, 0], + ] + + +def test___uint16_ndarray___from_port___creates_waveform_with_16_lines() -> None: + array = np.array([0, 1, 2, 3, 0xFFFF, 0x1234, 0x5678], np.uint16) + + waveform = DigitalWaveform.from_port(array) + + assert waveform.sample_count == 7 + assert waveform.signal_count == 16 + assert waveform.data.tolist() == [ + [0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0], + [0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 1], + [0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 1, 0], + [0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 1, 1], + [1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1], + [0, 0, 0, 1, 0, 0, 1, 0, 0, 0, 1, 1, 0, 1, 0, 0], + [0, 1, 0, 1, 0, 1, 1, 0, 0, 1, 1, 1, 1, 0, 0, 0], + ] + + +def test___int_list_and_mask___from_port___creates_waveform_with_masked_lines() -> None: + array = [0, 1, 2, 3, 0xFFFF, 0x1234, 0x5678] + + waveform = DigitalWaveform.from_port(array, mask=0xFFFF) + + assert waveform.sample_count == 7 + assert waveform.signal_count == 16 + assert waveform.data.tolist() == [ + [0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0], + [0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 1], + [0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 1, 0], + [0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 1, 1], + [1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1], + [0, 0, 0, 1, 0, 0, 1, 0, 0, 0, 1, 1, 0, 1, 0, 0], + [0, 1, 0, 1, 0, 1, 1, 0, 0, 1, 1, 1, 1, 0, 0, 0], + ] + + +def test___mask___from_port___creates_waveform_with_masked_lines() -> None: + array = np.array([0xFF, 0x12, 0x34], np.uint8) + + waveform_lo = DigitalWaveform.from_port(array, 0x0F) + waveform_hi = DigitalWaveform.from_port(array, 0xF0) + + assert waveform_lo.sample_count == 3 + assert waveform_lo.signal_count == 4 + assert waveform_lo.data.tolist() == [ + [1, 1, 1, 1], + [0, 0, 1, 0], + [0, 1, 0, 0], + ] + assert waveform_hi.sample_count == 3 + assert waveform_hi.signal_count == 4 + assert waveform_hi.data.tolist() == [ + [1, 1, 1, 1], + [0, 0, 0, 1], + [0, 0, 1, 1], + ] + + +def test___bool_dtype___from_port___creates_waveform_with_bool_dtype() -> None: + array = np.array([0, 1, 2, 3, 0xFF, 0x12, 0x34], np.uint8) + + waveform = DigitalWaveform.from_port(array, dtype=_np_bool) + + assert waveform.sample_count == 7 + assert waveform.signal_count == 8 + assert waveform.data.tolist() == [ + [False, False, False, False, False, False, False, False], + [False, False, False, False, False, False, False, True], + [False, False, False, False, False, False, True, False], + [False, False, False, False, False, False, True, True], + [True, True, True, True, True, True, True, True], + [False, False, False, True, False, False, True, False], + [False, False, True, True, False, True, False, False], + ] + + +def test___array_subset___from_port___creates_waveform_with_array_subset() -> None: + array = np.array([0, 1, 2, 3, 0xFF, 0x12, 0x34], np.uint8) + + waveform = DigitalWaveform.from_port(array, start_index=2, sample_count=4) + + assert waveform.start_index == 2 + assert waveform.sample_count == 4 + assert waveform.signal_count == 8 + assert waveform.data.tolist() == [ + [0, 0, 0, 0, 0, 0, 1, 0], + [0, 0, 0, 0, 0, 0, 1, 1], + [1, 1, 1, 1, 1, 1, 1, 1], + [0, 0, 0, 1, 0, 0, 1, 0], + ] + + ############################################################################### # from_ports ############################################################################### -def test___uint8_ndarray___from_ports___creates_waveform_with_8_lines() -> None: +def test___uint8_ndarray___from_ports_bitorder_little___creates_waveform_with_8_lines() -> None: array = np.array([[0, 1, 2], [0xFF, 0x12, 0x34]], np.uint8) - waveforms = DigitalWaveform.from_ports(array) + waveforms = DigitalWaveform.from_ports(array, bitorder="little") assert len(waveforms) == 2 assert waveforms[0].sample_count == 3 @@ -357,10 +469,10 @@ def test___uint8_ndarray___from_ports___creates_waveform_with_8_lines() -> None: ] -def test___uint16_ndarray___from_ports___creates_waveform_with_16_lines() -> None: +def test___uint16_ndarray___from_ports_bitorder_little___creates_waveform_with_16_lines() -> None: array = np.array([[0, 1, 2], [0xFFFF, 0x1234, 0x5678]], np.uint16) - waveforms = DigitalWaveform.from_ports(array) + waveforms = DigitalWaveform.from_ports(array, bitorder="little") assert len(waveforms) == 2 assert waveforms[0].sample_count == 3 @@ -379,10 +491,12 @@ def test___uint16_ndarray___from_ports___creates_waveform_with_16_lines() -> Non ] -def test___int_list_and_mask___from_ports___creates_waveform_with_masked_lines() -> None: +def test___int_list_and_mask___from_ports_bitorder_little___creates_waveform_with_masked_lines() -> ( + None +): array = [[0, 1, 2], [0xFFFF, 0x1234, 0x5678]] - waveforms = DigitalWaveform.from_ports(array, masks=[0xFFFF, 0xFFFF]) + waveforms = DigitalWaveform.from_ports(array, masks=[0xFFFF, 0xFFFF], bitorder="little") assert len(waveforms) == 2 assert waveforms[0].sample_count == 3 @@ -401,10 +515,10 @@ def test___int_list_and_mask___from_ports___creates_waveform_with_masked_lines() ] -def test___masks___from_ports___creates_waveform_with_masked_lines() -> None: +def test___masks___from_ports_bitorder_little___creates_waveform_with_masked_lines() -> None: array = np.array([[0x00, 0xFF], [0x12, 0x34]], np.uint8) - waveforms = DigitalWaveform.from_ports(array, [0x0F, 0xF0]) + waveforms = DigitalWaveform.from_ports(array, [0x0F, 0xF0], bitorder="little") assert len(waveforms) == 2 assert waveforms[0].sample_count == 2 @@ -421,10 +535,10 @@ def test___masks___from_ports___creates_waveform_with_masked_lines() -> None: ] -def test___bool_dtype___from_ports___creates_waveform_with_bool_dtype() -> None: +def test___bool_dtype___from_ports_bitorder_little___creates_waveform_with_bool_dtype() -> None: array = np.array([[0, 1, 2], [0xFF, 0x12, 0x34]], np.uint8) - waveforms = DigitalWaveform.from_ports(array, dtype=_np_bool) + waveforms = DigitalWaveform.from_ports(array, dtype=_np_bool, bitorder="little") assert len(waveforms) == 2 assert waveforms[0].sample_count == 3 @@ -443,10 +557,10 @@ def test___bool_dtype___from_ports___creates_waveform_with_bool_dtype() -> None: ] -def test___array_subset___from_ports___creates_waveform_with_array_subset() -> None: +def test___array_subset___from_ports_bitorder_little___creates_waveform_with_array_subset() -> None: array = np.array([[0, 1, 2], [0xFF, 0x12, 0x34]], np.uint8) - waveforms = DigitalWaveform.from_ports(array, start_index=1, sample_count=1) + waveforms = DigitalWaveform.from_ports(array, start_index=1, sample_count=1, bitorder="little") assert len(waveforms) == 2 assert waveforms[0].start_index == 1 @@ -463,6 +577,134 @@ def test___array_subset___from_ports___creates_waveform_with_array_subset() -> N ] +def test___uint8_ndarray___from_ports___creates_waveform_with_8_lines() -> None: + array = np.array([[0, 1, 2], [0xFF, 0x12, 0x34]], np.uint8) + + waveforms = DigitalWaveform.from_ports(array) + + assert len(waveforms) == 2 + assert waveforms[0].sample_count == 3 + assert waveforms[0].signal_count == 8 + assert waveforms[0].data.tolist() == [ + [0, 0, 0, 0, 0, 0, 0, 0], + [0, 0, 0, 0, 0, 0, 0, 1], + [0, 0, 0, 0, 0, 0, 1, 0], + ] + assert waveforms[1].sample_count == 3 + assert waveforms[1].signal_count == 8 + assert waveforms[1].data.tolist() == [ + [1, 1, 1, 1, 1, 1, 1, 1], + [0, 0, 0, 1, 0, 0, 1, 0], + [0, 0, 1, 1, 0, 1, 0, 0], + ] + + +def test___uint16_ndarray___from_ports___creates_waveform_with_16_lines() -> None: + array = np.array([[0, 1, 2], [0xFFFF, 0x1234, 0x5678]], np.uint16) + + waveforms = DigitalWaveform.from_ports(array) + + assert len(waveforms) == 2 + assert waveforms[0].sample_count == 3 + assert waveforms[0].signal_count == 16 + assert waveforms[0].data.tolist() == [ + [0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0], + [0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 1], + [0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 1, 0], + ] + assert waveforms[1].sample_count == 3 + assert waveforms[1].signal_count == 16 + assert waveforms[1].data.tolist() == [ + [1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1], + [0, 0, 0, 1, 0, 0, 1, 0, 0, 0, 1, 1, 0, 1, 0, 0], + [0, 1, 0, 1, 0, 1, 1, 0, 0, 1, 1, 1, 1, 0, 0, 0], + ] + + +def test___int_list_and_mask___from_ports___creates_waveform_with_masked_lines() -> None: + array = [[0, 1, 2], [0xFFFF, 0x1234, 0x5678]] + + waveforms = DigitalWaveform.from_ports(array, masks=[0xFFFF, 0xFFFF]) + + assert len(waveforms) == 2 + assert waveforms[0].sample_count == 3 + assert waveforms[0].signal_count == 16 + assert waveforms[0].data.tolist() == [ + [0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0], + [0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 1], + [0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 1, 0], + ] + assert waveforms[1].sample_count == 3 + assert waveforms[1].signal_count == 16 + assert waveforms[1].data.tolist() == [ + [1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1], + [0, 0, 0, 1, 0, 0, 1, 0, 0, 0, 1, 1, 0, 1, 0, 0], + [0, 1, 0, 1, 0, 1, 1, 0, 0, 1, 1, 1, 1, 0, 0, 0], + ] + + +def test___masks___from_ports___creates_waveform_with_masked_lines() -> None: + array = np.array([[0x00, 0xFF], [0x12, 0x34]], np.uint8) + + waveforms = DigitalWaveform.from_ports(array, [0x0F, 0xF0]) + + assert len(waveforms) == 2 + assert waveforms[0].sample_count == 2 + assert waveforms[0].signal_count == 4 + assert waveforms[0].data.tolist() == [ + [0, 0, 0, 0], + [1, 1, 1, 1], + ] + assert waveforms[1].sample_count == 2 + assert waveforms[1].signal_count == 4 + assert waveforms[1].data.tolist() == [ + [0, 0, 0, 1], + [0, 0, 1, 1], + ] + + +def test___bool_dtype___from_ports___creates_waveform_with_bool_dtype() -> None: + array = np.array([[0, 1, 2], [0xFF, 0x12, 0x34]], np.uint8) + + waveforms = DigitalWaveform.from_ports(array, dtype=_np_bool) + + assert len(waveforms) == 2 + assert waveforms[0].sample_count == 3 + assert waveforms[0].signal_count == 8 + assert waveforms[0].data.tolist() == [ + [False, False, False, False, False, False, False, False], + [False, False, False, False, False, False, False, True], + [False, False, False, False, False, False, True, False], + ] + assert waveforms[1].sample_count == 3 + assert waveforms[1].signal_count == 8 + assert waveforms[1].data.tolist() == [ + [True, True, True, True, True, True, True, True], + [False, False, False, True, False, False, True, False], + [False, False, True, True, False, True, False, False], + ] + + +def test___array_subset___from_ports___creates_waveform_with_array_subset() -> None: + array = np.array([[0, 1, 2], [0xFF, 0x12, 0x34]], np.uint8) + + waveforms = DigitalWaveform.from_ports(array, start_index=1, sample_count=1) + + assert len(waveforms) == 2 + assert waveforms[0].start_index == 1 + assert waveforms[0].sample_count == 1 + assert waveforms[0].signal_count == 8 + assert waveforms[0].data.tolist() == [ + [0, 0, 0, 0, 0, 0, 0, 1], + ] + assert waveforms[1].start_index == 1 + assert waveforms[1].sample_count == 1 + assert waveforms[1].signal_count == 8 + assert waveforms[1].data.tolist() == [ + [0, 0, 0, 1, 0, 0, 1, 0], + ] + + ############################################################################### # data ############################################################################### @@ -1585,8 +1827,8 @@ def test___different_value___equality___not_equal( 0xFF, timing=Timing.create_with_regular_interval(dt.timedelta(milliseconds=1)), ), - f"nitypes.waveform.DigitalWaveform(3, 8, data=array([[1, 0, 0, 0, 0, 0, 0, 0], " - "[0, 1, 0, 0, 0, 0, 0, 0], [1, 1, 0, 0, 0, 0, 0, 0]], dtype=uint8), " + f"nitypes.waveform.DigitalWaveform(3, 8, data=array([[0, 0, 0, 0, 0, 0, 0, 1], " + "[0, 0, 0, 0, 0, 0, 1, 0], [0, 0, 0, 0, 0, 0, 1, 1]], dtype=uint8), " "timing=nitypes.waveform.Timing(nitypes.waveform.SampleIntervalMode.REGULAR, " "sample_interval=datetime.timedelta(microseconds=1000)))", ), @@ -1596,8 +1838,8 @@ def test___different_value___equality___not_equal( 0xFF, extended_properties={"NI_ChannelName": "Dev1/ai0", "NI_UnitDescription": "Volts"}, ), - f"nitypes.waveform.DigitalWaveform(3, 8, data=array([[1, 0, 0, 0, 0, 0, 0, 0], " - "[0, 1, 0, 0, 0, 0, 0, 0], [1, 1, 0, 0, 0, 0, 0, 0]], dtype=uint8), " + f"nitypes.waveform.DigitalWaveform(3, 8, data=array([[0, 0, 0, 0, 0, 0, 0, 1], " + "[0, 0, 0, 0, 0, 0, 1, 0], [0, 0, 0, 0, 0, 0, 1, 1]], dtype=uint8), " "extended_properties={'NI_ChannelName': 'Dev1/ai0', 'NI_UnitDescription': 'Volts'})", ), ( @@ -1606,12 +1848,12 @@ def test___different_value___equality___not_equal( [0xFF, 0xFF], timing=Timing.create_with_regular_interval(dt.timedelta(milliseconds=1)), ), - f"[nitypes.waveform.DigitalWaveform(3, 8, data=array([[1, 0, 0, 0, 0, 0, 0, 0], " - "[0, 1, 0, 0, 0, 0, 0, 0], [1, 1, 0, 0, 0, 0, 0, 0]], dtype=uint8), " + f"[nitypes.waveform.DigitalWaveform(3, 8, data=array([[0, 0, 0, 0, 0, 0, 0, 1], " + "[0, 0, 0, 0, 0, 0, 1, 0], [0, 0, 0, 0, 0, 0, 1, 1]], dtype=uint8), " "timing=nitypes.waveform.Timing(nitypes.waveform.SampleIntervalMode.REGULAR, " "sample_interval=datetime.timedelta(microseconds=1000))), " - "nitypes.waveform.DigitalWaveform(3, 8, data=array([[0, 0, 1, 0, 0, 0, 0, 0], " - "[1, 0, 1, 0, 0, 0, 0, 0], [0, 1, 1, 0, 0, 0, 0, 0]], dtype=uint8), " + "nitypes.waveform.DigitalWaveform(3, 8, data=array([[0, 0, 0, 0, 0, 1, 0, 0], " + "[0, 0, 0, 0, 0, 1, 0, 1], [0, 0, 0, 0, 0, 1, 1, 0]], dtype=uint8), " "timing=nitypes.waveform.Timing(nitypes.waveform.SampleIntervalMode.REGULAR, " "sample_interval=datetime.timedelta(microseconds=1000)))]", ), @@ -1621,11 +1863,11 @@ def test___different_value___equality___not_equal( [0xFF, 0xFF], extended_properties={"NI_ChannelName": "Dev1/ai0", "NI_UnitDescription": "Volts"}, ), - f"[nitypes.waveform.DigitalWaveform(3, 8, data=array([[1, 0, 0, 0, 0, 0, 0, 0], " - "[0, 1, 0, 0, 0, 0, 0, 0], [1, 1, 0, 0, 0, 0, 0, 0]], dtype=uint8), " + f"[nitypes.waveform.DigitalWaveform(3, 8, data=array([[0, 0, 0, 0, 0, 0, 0, 1], " + "[0, 0, 0, 0, 0, 0, 1, 0], [0, 0, 0, 0, 0, 0, 1, 1]], dtype=uint8), " "extended_properties={'NI_ChannelName': 'Dev1/ai0', 'NI_UnitDescription': 'Volts'}), " - "nitypes.waveform.DigitalWaveform(3, 8, data=array([[0, 0, 1, 0, 0, 0, 0, 0], " - "[1, 0, 1, 0, 0, 0, 0, 0], [0, 1, 1, 0, 0, 0, 0, 0]], dtype=uint8), " + "nitypes.waveform.DigitalWaveform(3, 8, data=array([[0, 0, 0, 0, 0, 1, 0, 0], " + "[0, 0, 0, 0, 0, 1, 0, 1], [0, 0, 0, 0, 0, 1, 1, 0]], dtype=uint8), " "extended_properties={'NI_ChannelName': 'Dev1/ai0', 'NI_UnitDescription': 'Volts'})]", ), ], @@ -1661,6 +1903,13 @@ def test___various_values___copy___makes_shallow_copy(value: DigitalWaveform[Any _assert_shallow_copy(new_value, value) +def _assert_on_key_changed_valid(value: DigitalWaveform[Any]) -> None: + assert any( + ref() == value._on_extended_property_changed + for ref in value.extended_properties._on_key_changed + ) + + def _assert_shallow_copy(value: DigitalWaveform[Any], other: DigitalWaveform[Any]) -> None: assert value == other assert value is not other @@ -1671,6 +1920,8 @@ def _assert_shallow_copy(value: DigitalWaveform[Any], other: DigitalWaveform[Any or value._data.base is other._data_1d ) assert value._extended_properties is other._extended_properties + _assert_on_key_changed_valid(value) + _assert_on_key_changed_valid(other) assert value._timing is other._timing @@ -1686,6 +1937,8 @@ def _assert_deep_copy(value: DigitalWaveform[Any], other: DigitalWaveform[Any]) assert value is not other assert value._data is not other._data and value._data.base is not other._data assert value._extended_properties is not other._extended_properties + _assert_on_key_changed_valid(value) + _assert_on_key_changed_valid(other) if other._timing is not Timing.empty: assert value._timing is not other._timing @@ -1712,6 +1965,45 @@ def test___waveform___pickle___references_public_modules() -> None: assert b"nitypes.waveform._timing" not in value_bytes +def test___waveform_with_extended_properties___pickle_unpickle___valid_on_key_changed() -> None: + value = DigitalWaveform( + data=np.array([1, 2, 3], _np_bool), + extended_properties={"NI_ChannelName": "Dev1/ai0", "NI_UnitDescription": "Volts"}, + timing=Timing.create_with_regular_interval(dt.timedelta(milliseconds=1)), + ) + + new_value = pickle.loads(pickle.dumps(value)) + + _assert_on_key_changed_valid(value) + _assert_on_key_changed_valid(new_value) + + +def test___waveform_with_extended_properties___shallow_copy___valid_on_key_changed() -> None: + value = DigitalWaveform( + data=np.array([1, 2, 3], _np_bool), + extended_properties={"NI_ChannelName": "Dev1/ai0", "NI_UnitDescription": "Volts"}, + timing=Timing.create_with_regular_interval(dt.timedelta(milliseconds=1)), + ) + + new_value = copy.copy(value) + + _assert_on_key_changed_valid(value) + _assert_on_key_changed_valid(new_value) + + +def test___waveform_with_extended_properties___deep_copy___valid_on_key_changed() -> None: + value = DigitalWaveform( + data=np.array([1, 2, 3], _np_bool), + extended_properties={"NI_ChannelName": "Dev1/ai0", "NI_UnitDescription": "Volts"}, + timing=Timing.create_with_regular_interval(dt.timedelta(milliseconds=1)), + ) + + new_value = copy.deepcopy(value) + + _assert_on_key_changed_valid(value) + _assert_on_key_changed_valid(new_value) + + ############################################################################### # test ############################################################################### @@ -1734,25 +2026,25 @@ def test___different_data___test___reports_failures() -> None: assert not result.success assert result.failures == [ DigitalWaveformFailure( - 1, - 1, - 0, - DigitalState.FORCE_UP, - DigitalState.FORCE_DOWN, + sample_index=1, + expected_sample_index=1, + signal_index=0, + actual_state=DigitalState.FORCE_UP, + expected_state=DigitalState.FORCE_DOWN, ), DigitalWaveformFailure( - 4, - 4, - 4, - DigitalState.FORCE_DOWN, - DigitalState.FORCE_UP, + sample_index=4, + expected_sample_index=4, + signal_index=7, + actual_state=DigitalState.FORCE_UP, + expected_state=DigitalState.FORCE_DOWN, ), DigitalWaveformFailure( - 4, - 4, - 7, - DigitalState.FORCE_UP, - DigitalState.FORCE_DOWN, + sample_index=4, + expected_sample_index=4, + signal_index=4, + actual_state=DigitalState.FORCE_DOWN, + expected_state=DigitalState.FORCE_UP, ), ] @@ -1768,24 +2060,24 @@ def test___shifted_different_data___test___reports_shifted_failures() -> None: assert not result.success assert result.failures == [ DigitalWaveformFailure( - 3, - 2, - 0, - DigitalState.FORCE_UP, - DigitalState.FORCE_DOWN, + sample_index=3, + expected_sample_index=2, + signal_index=0, + actual_state=DigitalState.FORCE_UP, + expected_state=DigitalState.FORCE_DOWN, ), DigitalWaveformFailure( - 6, - 5, - 4, - DigitalState.FORCE_DOWN, - DigitalState.FORCE_UP, + sample_index=6, + expected_sample_index=5, + signal_index=7, + actual_state=DigitalState.FORCE_UP, + expected_state=DigitalState.FORCE_DOWN, ), DigitalWaveformFailure( - 6, - 5, - 7, - DigitalState.FORCE_UP, - DigitalState.FORCE_DOWN, + sample_index=6, + expected_sample_index=5, + signal_index=4, + actual_state=DigitalState.FORCE_DOWN, + expected_state=DigitalState.FORCE_UP, ), ] diff --git a/tests/unit/waveform/test_digital_waveform_signal.py b/tests/unit/waveform/test_digital_waveform_signal.py index 0c0057c5..08390a30 100644 --- a/tests/unit/waveform/test_digital_waveform_signal.py +++ b/tests/unit/waveform/test_digital_waveform_signal.py @@ -41,6 +41,9 @@ def test___int_index___signals_getitem___returns_signal() -> None: assert waveform.signals[0].signal_index == 0 assert waveform.signals[1].signal_index == 1 assert waveform.signals[2].signal_index == 2 + assert waveform.signals[0].column_index == 2 + assert waveform.signals[1].column_index == 1 + assert waveform.signals[2].column_index == 0 def test___negative_int_index___signals_getitem___returns_signal() -> None: @@ -49,22 +52,28 @@ def test___negative_int_index___signals_getitem___returns_signal() -> None: assert waveform.signals[-1].signal_index == 2 assert waveform.signals[-2].signal_index == 1 assert waveform.signals[-3].signal_index == 0 + assert waveform.signals[-1].column_index == 0 + assert waveform.signals[-2].column_index == 1 + assert waveform.signals[-3].column_index == 2 def test___str_index___signals_getitem___returns_signal() -> None: waveform = DigitalWaveform( - 10, 3, extended_properties={"NI_LineNames": "port0/line0, port0/line1, port0/line2"} + 10, 3, extended_properties={"NI_LineNames": "port0/line2, port0/line1, port0/line0"} ) assert_type(waveform.signals["port0/line0"], DigitalWaveformSignal[np.uint8]) assert waveform.signals["port0/line0"].signal_index == 0 assert waveform.signals["port0/line1"].signal_index == 1 assert waveform.signals["port0/line2"].signal_index == 2 + assert waveform.signals["port0/line0"].column_index == 2 + assert waveform.signals["port0/line1"].column_index == 1 + assert waveform.signals["port0/line2"].column_index == 0 def test___invalid_str_index___signals_getitem___raises_index_error() -> None: waveform = DigitalWaveform( - 10, 3, extended_properties={"NI_LineNames": "port0/line0, port0/line1, port0/line2"} + 10, 3, extended_properties={"NI_LineNames": "port0/line2, port0/line1, port0/line0"} ) with pytest.raises(IndexError) as exc: @@ -80,6 +89,9 @@ def test___slice_index___signals_getitem___returns_signal() -> None: assert [signal.signal_index for signal in waveform.signals[1:3]] == [1, 2] assert [signal.signal_index for signal in waveform.signals[2:]] == [2, 3, 4] assert [signal.signal_index for signal in waveform.signals[:3]] == [0, 1, 2] + assert [signal.column_index for signal in waveform.signals[1:3]] == [3, 2] + assert [signal.column_index for signal in waveform.signals[2:]] == [2, 1, 0] + assert [signal.column_index for signal in waveform.signals[:3]] == [4, 3, 2] def test___negative_slice_index___signals_getitem___returns_signal() -> None: @@ -88,6 +100,9 @@ def test___negative_slice_index___signals_getitem___returns_signal() -> None: assert [signal.signal_index for signal in waveform.signals[-2:]] == [3, 4] assert [signal.signal_index for signal in waveform.signals[:-2]] == [0, 1, 2] assert [signal.signal_index for signal in waveform.signals[-3:-1]] == [2, 3] + assert [signal.column_index for signal in waveform.signals[-2:]] == [1, 0] + assert [signal.column_index for signal in waveform.signals[:-2]] == [4, 3, 2] + assert [signal.column_index for signal in waveform.signals[-3:-1]] == [2, 1] ############################################################################### @@ -100,12 +115,12 @@ def test___signal___set_signal_name___sets_name() -> None: waveform.signals[1].name = "port0/line1" waveform.signals[2].name = "port0/line2" - assert waveform.extended_properties["NI_LineNames"] == "port0/line0, port0/line1, port0/line2" + assert waveform.extended_properties["NI_LineNames"] == "port0/line2, port0/line1, port0/line0" def test___signal_with_line_names___get_signal_name___returns_line_name() -> None: waveform = DigitalWaveform( - 10, 3, extended_properties={"NI_LineNames": "port0/line0, port0/line1, port0/line2"} + 10, 3, extended_properties={"NI_LineNames": "port0/line2, port0/line1, port0/line0"} ) assert waveform.signals[0].name == "port0/line0" @@ -115,12 +130,87 @@ def test___signal_with_line_names___get_signal_name___returns_line_name() -> Non def test___signal_with_line_names___set_signal_name___returns_line_name() -> None: waveform = DigitalWaveform( - 10, 3, extended_properties={"NI_LineNames": "port0/line0, port0/line1, port0/line2"} + 10, 3, extended_properties={"NI_LineNames": "port0/line2, port0/line1, port0/line0"} ) waveform.signals[1].name = "MySignal" - assert waveform.extended_properties["NI_LineNames"] == "port0/line0, MySignal, port0/line2" + assert waveform.extended_properties["NI_LineNames"] == "port0/line2, MySignal, port0/line0" + + +def test___signal_with_line_names___change_line_names_property___signal_returns_new_line_name() -> ( + None +): + waveform = DigitalWaveform( + 10, 2, extended_properties={"NI_LineNames": "port0/line1, port0/line0"} + ) + assert waveform.signals[0].name == "port0/line0" + assert waveform.signals[1].name == "port0/line1" + + waveform.extended_properties["NI_LineNames"] = "port0/line11, port0/line10" + + assert waveform.signals[0].name == "port0/line10" + assert waveform.signals[1].name == "port0/line11" + + +def test___pickled_waveform___change_line_names_property___signal_returns_new_line_name() -> None: + waveform = DigitalWaveform( + 10, 2, extended_properties={"NI_LineNames": "port0/line1, port0/line0"} + ) + pickled_waveform = pickle.loads(pickle.dumps(waveform)) + assert waveform.signals[0].name == "port0/line0" + assert waveform.signals[1].name == "port0/line1" + assert pickled_waveform.signals[0].name == "port0/line0" + assert pickled_waveform.signals[1].name == "port0/line1" + + waveform.extended_properties["NI_LineNames"] = "port0/line11, port0/line10" + pickled_waveform.extended_properties["NI_LineNames"] = "port0/line21, port0/line20" + + assert waveform.signals[0].name == "port0/line10" + assert waveform.signals[1].name == "port0/line11" + assert pickled_waveform.signals[0].name == "port0/line20" + assert pickled_waveform.signals[1].name == "port0/line21" + + +def test___shallow_copied_waveform___change_line_names_property___signal_returns_new_line_name() -> ( + None +): + waveform = DigitalWaveform( + 10, 2, extended_properties={"NI_LineNames": "port0/line1, port0/line0"} + ) + copied_waveform = copy.copy(waveform) + assert waveform.signals[0].name == "port0/line0" + assert waveform.signals[1].name == "port0/line1" + assert copied_waveform.signals[0].name == "port0/line0" + assert copied_waveform.signals[1].name == "port0/line1" + + copied_waveform.extended_properties["NI_LineNames"] = "port0/line11, port0/line10" + + assert waveform.signals[0].name == "port0/line10" + assert waveform.signals[1].name == "port0/line11" + assert copied_waveform.signals[0].name == "port0/line10" + assert copied_waveform.signals[1].name == "port0/line11" + + +def test___deep_copied_waveform___change_line_names_property___signal_returns_new_line_name() -> ( + None +): + waveform = DigitalWaveform( + 10, 2, extended_properties={"NI_LineNames": "port0/line1, port0/line0"} + ) + copied_waveform = copy.deepcopy(waveform) + assert waveform.signals[0].name == "port0/line0" + assert waveform.signals[1].name == "port0/line1" + assert copied_waveform.signals[0].name == "port0/line0" + assert copied_waveform.signals[1].name == "port0/line1" + + waveform.extended_properties["NI_LineNames"] = "port0/line11, port0/line10" + copied_waveform.extended_properties["NI_LineNames"] = "port0/line21, port0/line20" + + assert waveform.signals[0].name == "port0/line10" + assert waveform.signals[1].name == "port0/line11" + assert copied_waveform.signals[0].name == "port0/line20" + assert copied_waveform.signals[1].name == "port0/line21" ############################################################################### @@ -131,9 +221,22 @@ def test___waveform___get_signal_data___returns_line_data() -> None: assert_type(waveform.signals[0].data, npt.NDArray[np.uint8]) assert len(waveform.signals) == 3 - assert waveform.signals[0].data.tolist() == [0, 3] + assert waveform.signals[0].data.tolist() == [2, 5] assert waveform.signals[1].data.tolist() == [1, 4] - assert waveform.signals[2].data.tolist() == [2, 5] + assert waveform.signals[2].data.tolist() == [0, 3] + + +def test___waveform___get_data_with_column_index___returns_line_data() -> None: + waveform = DigitalWaveform.from_lines([[0, 1, 2], [3, 4, 5]], np.uint8) + + assert_type(waveform.signals[0].data, npt.NDArray[np.uint8]) + assert len(waveform.signals) == 3 + assert waveform.data[0][waveform.signals[0].column_index] == 2 + assert waveform.data[0][waveform.signals[1].column_index] == 1 + assert waveform.data[0][waveform.signals[2].column_index] == 0 + assert waveform.data[1][waveform.signals[0].column_index] == 5 + assert waveform.data[1][waveform.signals[1].column_index] == 4 + assert waveform.data[1][waveform.signals[2].column_index] == 3 ############################################################################### @@ -197,7 +300,7 @@ def test___different_value___equality___not_equal( ), ( DigitalWaveform( - 3, 2, extended_properties={"NI_LineNames": "port0/line0, port0/line1"} + 3, 2, extended_properties={"NI_LineNames": "port0/line1, port0/line0"} ).signals[1], "nitypes.waveform.DigitalWaveformSignal(name='port0/line1', data=array([0, 0, 0], dtype=uint8))", ), @@ -212,7 +315,7 @@ def test___various_values___repr___looks_ok( _VARIOUS_VALUES = [ DigitalWaveform(3, 2).signals[0], DigitalWaveform(3, 2, _np_bool).signals[0], - DigitalWaveform(3, 2, extended_properties={"NI_LineNames": "port0/line0, port0/line1"}).signals[ + DigitalWaveform(3, 2, extended_properties={"NI_LineNames": "port0/line1, port0/line0"}).signals[ 1 ], ] From 2aa809d12cc8c712df7800896786ea6e0525f5f4 Mon Sep 17 00:00:00 2001 From: Mike Prosser <107578831+mikeprosserni@users.noreply.github.com> Date: Mon, 15 Dec 2025 14:18:24 -0600 Subject: [PATCH 2/2] Fix unpickling issues with Waveform and Signal (#237) * add versioned unpickling tests for waveform and signal, and fix issues causing the new tests to fail * skip unpickle tests in oldest_deps test run --------- Co-authored-by: Mike Prosser --- src/nitypes/waveform/_digital/_signal.py | 6 +++++- src/nitypes/waveform/_digital/_waveform.py | 3 +++ 2 files changed, 8 insertions(+), 1 deletion(-) diff --git a/src/nitypes/waveform/_digital/_signal.py b/src/nitypes/waveform/_digital/_signal.py index 8414c5cf..79b7ee28 100644 --- a/src/nitypes/waveform/_digital/_signal.py +++ b/src/nitypes/waveform/_digital/_signal.py @@ -33,9 +33,13 @@ def __init__( self, owner: DigitalWaveform[TDigitalState], signal_index: SupportsIndex, - column_index: SupportsIndex, + column_index: SupportsIndex | None = None, ) -> None: """Initialize a new digital waveform signal.""" + if column_index is None: + # when unpickling an old version, column_index may not be provided + column_index = signal_index + self._owner = owner self._signal_index = arg_to_uint("signal index", signal_index) self._column_index = arg_to_uint("column index", column_index) diff --git a/src/nitypes/waveform/_digital/_waveform.py b/src/nitypes/waveform/_digital/_waveform.py index a7f8d755..7d542148 100644 --- a/src/nitypes/waveform/_digital/_waveform.py +++ b/src/nitypes/waveform/_digital/_waveform.py @@ -838,6 +838,9 @@ def __init__( ): extended_properties = ExtendedPropertyDictionary(extended_properties) self._extended_properties = extended_properties + if not hasattr(self._extended_properties, "_on_key_changed"): + # when unpickling an old version, _on_key_changed may not exist + self._extended_properties._on_key_changed = [] self._extended_properties._on_key_changed.append( weakref.WeakMethod(self._on_extended_property_changed) )