Skip to content
Merged
Show file tree
Hide file tree
Changes from 10 commits
Commits
Show all changes
27 commits
Select commit Hold shift + click to select a range
1f7d1fa
add _raw_index
Nov 19, 2025
7a92534
line_index and line_names
Nov 20, 2025
9819c31
more tests
Nov 20, 2025
0c7e0e0
fix new test
Nov 20, 2025
101e9c1
data_index and documentation
Nov 20, 2025
ab0894d
fix doc types
Nov 21, 2025
acd1a8c
documentation improvements
Nov 21, 2025
1e60206
documentation improvements
Nov 21, 2025
61d2fc2
documentation improvements
Nov 21, 2025
897b3bd
clean up NI_LineNames in tests
Nov 21, 2025
1d558c0
remove DigitalWaveformFailure.data_index
Nov 24, 2025
b5ea682
Merge remote-tracking branch 'origin/main' into users/mprosser/bug-31…
Nov 24, 2025
992cac1
test___signal_with_line_names___change_line_names_property___signal_r…
Nov 25, 2025
329abd5
Merge remote-tracking branch 'origin/main' into users/mprosser/bug-31…
Dec 3, 2025
9d0eb7d
signal_column_index
Dec 3, 2025
957e82c
line lengths
Dec 3, 2025
9f5f60d
italics for 'See "Signal index vs. signal column index"' notes
Dec 3, 2025
63e4e70
add bitorder to from_port, and default to the industry standard 'big'
Dec 4, 2025
4968e62
rename to column_index
Dec 5, 2025
7ac5a23
bitorder != sys.byteorder
Dec 5, 2025
72a5b27
add on_key_changed to ExtendedPropertyDictionary
Dec 5, 2025
91f2810
change tests to big-endian
Dec 5, 2025
62dbd68
make on_key_changed private
Dec 8, 2025
a14f5c3
cleanup
Dec 8, 2025
2e774ac
from typing_extensions import TypeAlias, to fix python 3.9
Dec 8, 2025
a408007
fix pickling and copying issues with callbacks
Dec 10, 2025
7adf82c
change on_key_changed to a list of weak methods, so ExtendedPropertyD…
Dec 10, 2025
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
35 changes: 28 additions & 7 deletions src/nitypes/waveform/_digital/_signal.py
Original file line number Diff line number Diff line change
Expand Up @@ -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", "_data_index", "__weakref__"]

_owner: DigitalWaveform[TDigitalState]
_data_index: int
_signal_index: int

def __init__(self, owner: DigitalWaveform[TDigitalState], signal_index: SupportsIndex) -> None:
def __init__(
self,
owner: DigitalWaveform[TDigitalState],
signal_index: SupportsIndex,
data_index: SupportsIndex,
) -> None:
"""Initialize a new digital waveform signal."""
self._owner = owner
self._signal_index = arg_to_uint("signal index", signal_index)
self._data_index = arg_to_uint("data index", data_index)

@property
def owner(self) -> DigitalWaveform[TDigitalState]:
Expand All @@ -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 data_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[:, data_index]`.

Note: The data_index is reversed compared to the signal_index. Data index 0 (the leftmost
column) corresponds to the highest signal_index and highest line number. The highest
data_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._data_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._data_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._data_index)

@name.setter
def name(self, value: str) -> None:
self._owner._set_signal_name(self._signal_index, value)
self._owner._set_line_name(self._data_index, value)

def __eq__(self, value: object, /) -> bool:
"""Return self==value."""
Expand All @@ -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._data_index)
return (self.__class__, ctor_args)

def __repr__(self) -> str:
Expand Down
14 changes: 8 additions & 6 deletions src/nitypes/waveform/_digital/_signal_collection.py
Original file line number Diff line number Diff line change
Expand Up @@ -49,21 +49,23 @@ 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)
data_index = self._owner._reverse_index(index)
value = self._signals[index] = DigitalWaveformSignal(self._owner, index, data_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)
data_index = line_names.index(index)
except ValueError:
raise IndexError(index)
signal_index = self._owner._reverse_index(data_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)
142 changes: 101 additions & 41 deletions src/nitypes/waveform/_digital/_waveform.py
Original file line number Diff line number Diff line change
Expand Up @@ -66,6 +66,9 @@ class DigitalWaveformFailure:
signal_index: int
"""The signal index where the test failure occurred."""

data_index: int
"""The data index where the test failure occurred."""

actual_state: DigitalState
"""The state from the compared waveform where the test failure occurred."""

Expand Down Expand Up @@ -106,7 +109,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. data 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))
Expand All @@ -123,8 +127,8 @@ 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. data 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],
Expand Down Expand Up @@ -156,14 +160,48 @@ class DigitalWaveform(Generic[TDigitalState]):

>>> wfm = DigitalWaveform.from_port([0, 1, 2, 3], 0x3)
>>> wfm.signals[0]
nitypes.waveform.DigitalWaveformSignal(data=array([0, 1, 0, 1], dtype=uint8))
>>> wfm.signals[1]
nitypes.waveform.DigitalWaveformSignal(data=array([0, 0, 1, 1], dtype=uint8))
>>> wfm.signals[1]
nitypes.waveform.DigitalWaveformSignal(data=array([0, 1, 0, 1], dtype=uint8))

The :any:`DigitalWaveformSignal.data` property returns a view of the data for that signal.

>>> wfm.signals[0].data
array([0, 0, 1, 1], dtype=uint8)

Signal index vs. data 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.data_index` - The position in the :attr:`DigitalWaveform.data` array's
second dimension (0-based from the first column). Data 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 data_index, and the highest signal index (the highest line) corresponds to data_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, 4, 2, 6], 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].data_index
2
>>> wfm.signals[0].data
array([0, 1, 0, 1], dtype=uint8)
>>> wfm.signals[2].signal_index
2
>>> wfm.signals[2].data_index
0
>>> wfm.signals[2].data
array([0, 0, 0, 0], dtype=uint8)

Digital signal names
^^^^^^^^^^^^^^^^^^^^
Expand All @@ -172,20 +210,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 data_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]
Expand Down Expand Up @@ -215,7 +258,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],
Expand All @@ -232,10 +275,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, data_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=1,
DigitalWaveformFailure(sample_index=1, expected_sample_index=1, signal_index=0, data_index=1,
actual_state=<DigitalState.FORCE_UP: 1>, expected_state=<DigitalState.COMPARE_LOW: 3>)

Timing information
Expand Down Expand Up @@ -312,6 +355,11 @@ def from_lines(
by (sample, signal). The line data may also use digital state values from the
:class:`DigitalState` enum.

The rightmost bit (the last column) in the sample will be the least significant bit (line
0) and will be placed at signal index 0 in the DigitalWaveform. The leftmost bit (the first
column) in the sample will be the most significant bit (highest line number) and will be
placed at the highest signal index in the DigitalWaveform.

Args:
array: The line data as a one or two-dimensional array or a sequence.
dtype: The NumPy data type for the waveform data.
Expand Down Expand Up @@ -412,8 +460,12 @@ def from_port(

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".

The rightmost bit (the last column) in the sample will be the least significant bit (line
0) and will be placed at signal index 0 in the DigitalWaveform. The leftmost bit (the first
column) in the sample will be the most significant bit (highest line number) and will be
placed at the highest signal index in the DigitalWaveform.

If the input array is not a NumPy array, you must specify the mask.

Expand Down Expand Up @@ -536,9 +588,12 @@ 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".

The rightmost bit (the last column) in the sample will be the least significant bit (line
0) and will be placed at signal index 0 in the DigitalWaveform. The leftmost bit (the first
column) in the sample will be the most significant bit (highest line number) and will be
placed at the highest signal index in the DigitalWaveform.

If the input array is not a NumPy array, you must specify the masks.

Expand Down Expand Up @@ -617,7 +672,7 @@ def from_ports(
"_extended_properties",
"_timing",
"_signals",
"_signal_names",
"_line_names",
"__weakref__",
]

Expand All @@ -628,7 +683,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
Expand Down Expand Up @@ -763,7 +818,7 @@ def __init__(
self._timing = timing

self._signals = None
self._signal_names = None
self._line_names = None

def _init_with_new_array(
self,
Expand Down Expand Up @@ -976,26 +1031,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, data_index: int) -> str:
return self._get_line_names()[data_index]

def _set_line_name(self, data_index: int, value: str) -> None:
line_names = self._get_line_names()
line_names[data_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:
Expand Down Expand Up @@ -1247,17 +1300,19 @@ 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 data_index in range(self.signal_count):
signal_index = self._reverse_index(data_index)
actual_state = DigitalState(self.data[start_sample, data_index])
expected_state = DigitalState(
expected_waveform.data[expected_start_sample, signal_index]
expected_waveform.data[expected_start_sample, data_index]
)
if DigitalState.test(actual_state, expected_state):
failures.append(
DigitalWaveformFailure(
start_sample,
expected_start_sample,
signal_index,
data_index,
actual_state,
expected_state,
)
Expand All @@ -1267,6 +1322,11 @@ def test(

return DigitalWaveformTestResult(failures)

def _reverse_index(self, index: int) -> int:
"""Convert a signal index to a data 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__):
Expand Down
Loading