Reverse Indices for Digital Waveform Signals

src/nitypes/waveform/_digital/_port.py

@@ -118,7 +118,7 @@ def port_to_line_data(
     port_size = port_data.dtype.itemsize * 8
     # 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 == "big" and port_data.dtype.itemsize > 1 and sys.byteorder == "little":
+    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)

src/nitypes/waveform/_digital/_signal.py

@@ -23,22 +23,22 @@ class DigitalWaveformSignal(Generic[TDigitalState]):
     collection, e.g. ``waveform.signals[0]`` or ``waveform.signals["Dev1/port0/line0"]``.
     """
 
-    __slots__ = ["_owner", "_signal_index", "_signal_column_index", "__weakref__"]
+    __slots__ = ["_owner", "_signal_index", "_column_index", "__weakref__"]
 
     _owner: DigitalWaveform[TDigitalState]
-    _signal_column_index: int
+    _column_index: int
     _signal_index: int
 
     def __init__(
         self,
         owner: DigitalWaveform[TDigitalState],
         signal_index: SupportsIndex,
-        signal_column_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._signal_column_index = arg_to_uint("signal column index", signal_column_index)
+        self._column_index = arg_to_uint("column index", column_index)
 
     @property
     def owner(self) -> DigitalWaveform[TDigitalState]:
@@ -51,33 +51,32 @@ def signal_index(self) -> int:
         return self._signal_index
 
     @property
-    def signal_column_index(self) -> int:
+    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[:, signal_column_index]`.
+        `waveform.data[:, column_index]`.
 
-        Note: The signal_column_index is reversed compared to the signal_index. signal_column_index
-        0 (the leftmost column) corresponds to the highest signal_index and highest line number.
-        The highest signal_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.
+        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._signal_column_index
+        return self._column_index
 
     @property
     def data(self) -> npt.NDArray[TDigitalState]:
         """The signal data, indexed by sample."""
-        return self._owner.data[:, self._signal_column_index]
+        return self._owner.data[:, self._column_index]
 
     @property
     def name(self) -> str:
         """The signal name."""
-        return self._owner._get_line_name(self._signal_column_index)
+        return self._owner._get_line_name(self._column_index)
 
     @name.setter
     def name(self, value: str) -> None:
-        self._owner._set_line_name(self._signal_column_index, value)
+        self._owner._set_line_name(self._column_index, value)
 
     def __eq__(self, value: object, /) -> bool:
         """Return self==value."""
@@ -88,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, self._signal_column_index)
+        ctor_args = (self._owner, self._signal_index, self._column_index)
         return (self.__class__, ctor_args)
 
     def __repr__(self) -> str:

src/nitypes/waveform/_digital/_signal_collection.py

@@ -54,18 +54,18 @@ def __getitem__(
                 index += len(self._signals)
             value = self._signals[index]
             if value is None:
-                signal_column_index = self._owner._reverse_index(index)
+                column_index = self._owner._reverse_index(index)
                 value = self._signals[index] = DigitalWaveformSignal(
-                    self._owner, index, signal_column_index
+                    self._owner, index, column_index
                 )
             return value
         elif isinstance(index, str):  # index is the line name
             line_names = self._owner._get_line_names()
             try:
-                signal_column_index = line_names.index(index)
+                column_index = line_names.index(index)
             except ValueError:
                 raise IndexError(index)
-            signal_index = self._owner._reverse_index(signal_column_index)
+            signal_index = self._owner._reverse_index(column_index)
             return self[signal_index]
         elif isinstance(index, slice):  # index is a slice of signal indices
             return [self[i] for i in range(*index.indices(len(self)))]

src/nitypes/waveform/_digital/_waveform.py

@@ -86,30 +86,6 @@ def success(self) -> bool:
     """A collection of test failure information."""
 
 
-class _DigitalWaveformExtendedProperties(ExtendedPropertyDictionary):
-    """Extended properties that invalidate cached line names when NI_LineNames is modified."""
-
-    __slots__ = ["_waveform"]
-
-    def __init__(
-        self,
-        waveform: DigitalWaveform[Any],
-        properties: Mapping[str, ExtendedPropertyValue] | None = None,
-    ) -> None:
-        super().__init__(properties)
-        self._waveform = waveform
-
-    def __setitem__(self, key: str, value: ExtendedPropertyValue, /) -> None:
-        super().__setitem__(key, value)
-        if key == LINE_NAMES:
-            self._waveform._line_names = None
-
-    def __delitem__(self, key: str, /) -> None:
-        super().__delitem__(key)
-        if key == LINE_NAMES:
-            self._waveform._line_names = None
-
-
 class DigitalWaveform(Generic[TDigitalState]):
     """A digital waveform, which encapsulates digital data and timing information.
 
@@ -130,7 +106,7 @@ 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). *(Note, signal indices are reversed! See "Signal index vs. signal column index"
+    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
@@ -148,7 +124,7 @@ 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". *(Note, signal indices are reversed! See "Signal index vs. signal column index" below for
+    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
@@ -190,20 +166,20 @@ class DigitalWaveform(Generic[TDigitalState]):
     >>> wfm.signals[0].data
     array([0, 1, 0, 1], dtype=uint8)
 
-    Signal index vs. signal column index
-    ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+    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.signal_column_index` - The position in the :attr:`DigitalWaveform.data`
-      array's second dimension (0-based from the first column). signal_column_index 0 is the leftmost
-      column in the data.
+    * :attr:`DigitalWaveformSignal.column_index` - The position in the :attr:`DigitalWaveform.data`
+      array's second dimension (0-based from the first column). 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 signal_column_index, and the highest signal_index (the highest line) corresponds to
-    signal_column_index 0. This ordering follows industry conventions where line 0 is the least
+    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
@@ -214,13 +190,13 @@ class DigitalWaveform(Generic[TDigitalState]):
            [0, 1, 1]], dtype=uint8)
     >>> wfm.signals[0].signal_index
     0
-    >>> wfm.signals[0].signal_column_index
+    >>> 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].signal_column_index
+    >>> wfm.signals[2].column_index
     0
     >>> wfm.signals[2].data
     array([0, 0, 0, 0], dtype=uint8)
@@ -239,10 +215,10 @@ class DigitalWaveform(Generic[TDigitalState]):
     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 signal_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.
+    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/line2, port0/line1, port0/line0'
@@ -297,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=0, signal_column_index=1,
+    DigitalWaveformFailure(sample_index=0, expected_sample_index=0, signal_index=0, column_index=1,
     actual_state=<DigitalState.FORCE_UP: 1>, expected_state=<DigitalState.COMPARE_LOW: 3>)
     >>> result.failures[1]  # doctest: +NORMALIZE_WHITESPACE
-    DigitalWaveformFailure(sample_index=1, expected_sample_index=1, signal_index=0, signal_column_index=1,
+    DigitalWaveformFailure(sample_index=1, expected_sample_index=1, signal_index=0, column_index=1,
     actual_state=<DigitalState.FORCE_UP: 1>, expected_state=<DigitalState.COMPARE_LOW: 3>)
 
     Timing information
@@ -858,10 +834,11 @@ def __init__(
             raise invalid_arg_type("raw data", "NumPy ndarray", data)
 
         if copy_extended_properties or not isinstance(
-            extended_properties, _DigitalWaveformExtendedProperties
+            extended_properties, ExtendedPropertyDictionary
         ):
-            extended_properties = _DigitalWaveformExtendedProperties(self, extended_properties)
+            extended_properties = ExtendedPropertyDictionary(extended_properties)
         self._extended_properties = extended_properties
+        self._extended_properties.on_key_changed = self._on_extended_property_changed
 
         if timing is None:
             timing = Timing.empty
@@ -870,6 +847,10 @@ def __init__(
         self._signals = 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,
         sample_count: SupportsIndex | None = None,
@@ -1092,12 +1073,12 @@ def _get_line_names(self) -> list[str]:
                 line_names.extend([""] * (self.signal_count - len(line_names)))
         return line_names
 
-    def _get_line_name(self, signal_column_index: int) -> str:
-        return self._get_line_names()[signal_column_index]
+    def _get_line_name(self, column_index: int) -> str:
+        return self._get_line_names()[column_index]
 
-    def _set_line_name(self, signal_column_index: int, value: str) -> None:
+    def _set_line_name(self, column_index: int, value: str) -> None:
         line_names = self._get_line_names()
-        line_names[signal_column_index] = value
+        line_names[column_index] = value
         self._extended_properties[LINE_NAMES] = ", ".join(line_names)
 
     def _set_timing(self, value: Timing[AnyDateTime, AnyTimeDelta, AnyTimeDelta]) -> None:
@@ -1350,11 +1331,11 @@ def test(
 
         failures = []
         for _ in range(sample_count):
-            for signal_column_index in range(self.signal_count):
-                signal_index = self._reverse_index(signal_column_index)
-                actual_state = DigitalState(self.data[start_sample, signal_column_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_column_index]
+                    expected_waveform.data[expected_start_sample, column_index]
                 )
                 if DigitalState.test(actual_state, expected_state):
                     failures.append(
@@ -1372,7 +1353,7 @@ def test(
         return DigitalWaveformTestResult(failures)
 
     def _reverse_index(self, index: int) -> int:
-        """Convert a signal_index to a signal_column_index, or vice versa."""
+        """Convert a signal_index to a column_index, or vice versa."""
         assert 0 <= index < self.signal_count
         return self.signal_count - 1 - index
 
@@ -1387,24 +1368,12 @@ def __eq__(self, value: object, /) -> bool:
             and self._timing == value._timing
         )
 
-    def __copy__(self) -> Self:
-        """Return a shallow copy of self."""
-        return self.__class__(
-            self._sample_count,
-            self.signal_count,
-            self.dtype,
-            data=self.data,
-            extended_properties=self._extended_properties,
-            copy_extended_properties=False,
-            timing=self._timing,
-        )
-
     def __reduce__(self) -> tuple[Any, ...]:
         """Return object state for pickling."""
         ctor_args = (self._sample_count, self.signal_count, self.dtype)
         ctor_kwargs: dict[str, Any] = {
             "data": self.data,
-            "extended_properties": ExtendedPropertyDictionary(self._extended_properties),
+            "extended_properties": self._extended_properties,
             "copy_extended_properties": False,
             "timing": self._timing,
         }

src/nitypes/waveform/_extended_properties.py

@@ -1,7 +1,8 @@
 from __future__ import annotations
 
 import operator
-from collections.abc import Iterator, Mapping, MutableMapping
+from collections.abc import Callable, Iterator, Mapping, MutableMapping
+from typing import TYPE_CHECKING
 
 from nitypes.waveform.typing import ExtendedPropertyValue
 
@@ -10,6 +11,9 @@
 LINE_NAMES = "NI_LineNames"
 UNIT_DESCRIPTION = "NI_UnitDescription"
 
+if TYPE_CHECKING:
+    OnKeyChangedCallback = Callable[[str], None]
+
 
 class ExtendedPropertyDictionary(MutableMapping[str, ExtendedPropertyValue]):
     """A dictionary of extended properties.
@@ -19,14 +23,24 @@ 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: OnKeyChangedCallback | None = None
         if properties is not None:
             self._properties.update(properties)
 
+    @property
+    def on_key_changed(self) -> OnKeyChangedCallback | None:
+        """Callback invoked when a key is set or deleted."""
+        return self._on_key_changed
+
+    @on_key_changed.setter
+    def on_key_changed(self, value: OnKeyChangedCallback | None) -> None:
+        self._on_key_changed = value
+
     def __len__(self) -> int:
         """Return len(self)."""
         return len(self._properties)
@@ -46,10 +60,14 @@ 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)
+        if self._on_key_changed is not None:
+            self._on_key_changed(key)
 
     def __delitem__(self, key: str, /) -> None:
         """Delete self[key]."""
         operator.delitem(self._properties, key)
+        if self._on_key_changed is not None:
+            self._on_key_changed(key)
 
     def _merge(self, other: ExtendedPropertyDictionary) -> None:
         for key, value in other.items():