diff --git a/src/nitypes/_exceptions.py b/src/nitypes/_exceptions.py index d49cc2fd..890bf6e2 100644 --- a/src/nitypes/_exceptions.py +++ b/src/nitypes/_exceptions.py @@ -44,7 +44,7 @@ def invalid_arg_type(arg_description: str, type_description: str, value: object) def invalid_array_ndim(arg_description: str, valid_value_description: str, ndim: int) -> ValueError: """Create a ValueError for an array with an invalid number of dimensions.""" - raise ValueError( + return ValueError( f"The {arg_description} must be {_a(valid_value_description)}.\n\n" f"Number of dimensions: {ndim}" ) @@ -52,7 +52,7 @@ def invalid_array_ndim(arg_description: str, valid_value_description: str, ndim: def invalid_requested_type(type_description: str, requested_type: type) -> TypeError: """Create a TypeError for an invalid requested type.""" - raise TypeError( + return TypeError( f"The requested type must be {_a(type_description)} type.\n\n" f"Requested type: {requested_type}" ) @@ -60,7 +60,7 @@ def invalid_requested_type(type_description: str, requested_type: type) -> TypeE def unsupported_arg(arg_description: str, value: object) -> ValueError: """Create a ValueError for an unsupported argument.""" - raise ValueError( + return ValueError( f"The {arg_description} argument is not supported.\n\n" f"Provided value: {reprlib.repr(value)}" ) diff --git a/src/nitypes/waveform/__init__.py b/src/nitypes/waveform/__init__.py index 5efb2fe5..d627e723 100644 --- a/src/nitypes/waveform/__init__.py +++ b/src/nitypes/waveform/__init__.py @@ -1,6 +1,7 @@ """Waveform data types for NI Python APIs.""" from nitypes.waveform._analog_waveform import AnalogWaveform +from nitypes.waveform._exceptions import TimingMismatchError from nitypes.waveform._extended_properties import ( ExtendedPropertyDictionary, ExtendedPropertyValue, @@ -11,7 +12,13 @@ NoneScaleMode, ScaleMode, ) -from nitypes.waveform._timing import BaseTiming, PrecisionTiming, SampleIntervalMode, Timing +from nitypes.waveform._timing import ( + BaseTiming, + PrecisionTiming, + SampleIntervalMode, + Timing, +) +from nitypes.waveform._warnings import ScalingMismatchWarning, TimingMismatchWarning __all__ = [ "AnalogWaveform", @@ -24,7 +31,10 @@ "PrecisionTiming", "SampleIntervalMode", "ScaleMode", + "ScalingMismatchWarning", "Timing", + "TimingMismatchError", + "TimingMismatchWarning", ] # Hide that it was defined in a helper file @@ -33,8 +43,12 @@ ExtendedPropertyDictionary.__module__ = __name__ # ExtendedPropertyValue is a TypeAlias LinearScaleMode.__module__ = __name__ +# NO_SCALING is a constant NoneScaleMode.__module__ = __name__ PrecisionTiming.__module__ = __name__ SampleIntervalMode.__module__ = __name__ ScaleMode.__module__ = __name__ +ScalingMismatchWarning.__module__ = __name__ Timing.__module__ = __name__ +TimingMismatchError.__module__ = __name__ +TimingMismatchWarning.__module__ = __name__ diff --git a/src/nitypes/waveform/_analog_waveform.py b/src/nitypes/waveform/_analog_waveform.py index 91f4ff9c..f947b0d5 100644 --- a/src/nitypes/waveform/_analog_waveform.py +++ b/src/nitypes/waveform/_analog_waveform.py @@ -1,34 +1,37 @@ from __future__ import annotations +import datetime as dt import sys +import warnings from collections.abc import Sequence -from typing import ( - Any, - Generic, - SupportsIndex, - TypeVar, - overload, -) +from typing import Any, Generic, SupportsIndex, TypeVar, Union, cast, overload +import hightime as ht import numpy as np import numpy.typing as npt -from nitypes._arguments import arg_to_uint, validate_dtype +from nitypes._arguments import arg_to_uint, validate_dtype, validate_unsupported_arg from nitypes._exceptions import invalid_arg_type, invalid_array_ndim +from nitypes._typing import TypeAlias from nitypes.waveform._extended_properties import ( CHANNEL_NAME, UNIT_DESCRIPTION, ExtendedPropertyDictionary, ) from nitypes.waveform._scaling import NO_SCALING, ScaleMode -from nitypes.waveform._timing import PrecisionTiming, Timing, convert_timing +from nitypes.waveform._timing import BaseTiming, PrecisionTiming, Timing, convert_timing +from nitypes.waveform._warnings import scale_mode_mismatch if sys.version_info < (3, 10): import array as std_array + _ScalarType = TypeVar("_ScalarType", bound=np.generic) _ScalarType_co = TypeVar("_ScalarType_co", bound=np.generic, covariant=True) +_AnyTiming: TypeAlias = Union[BaseTiming[Any, Any], Timing, PrecisionTiming] +_TTiming = TypeVar("_TTiming", bound=BaseTiming[Any, Any]) + # Use the C types here because np.isdtype() considers some of them to be distinct types, even when # they have the same size (e.g. np.intc vs. np.int_ vs. np.long). _ANALOG_DTYPES = ( @@ -57,7 +60,6 @@ np.double, ) - # Note about NumPy type hints: # - At time of writing (April 2025), shape typing is still under development, so we do not # distinguish between 1D and 2D arrays in type hints. @@ -229,7 +231,7 @@ def from_array_2d( "_sample_count", "_extended_properties", "_timing", - "_precision_timing", + "_converted_timing_cache", "_scale_mode", "__weakref__", ] @@ -238,8 +240,8 @@ def from_array_2d( _start_index: int _sample_count: int _extended_properties: ExtendedPropertyDictionary - _timing: Timing | None - _precision_timing: PrecisionTiming | None + _timing: BaseTiming[Any, Any] + _converted_timing_cache: dict[type[_AnyTiming], _AnyTiming] _scale_mode: ScaleMode # If neither dtype nor _data is specified, the type parameter defaults to np.float64. @@ -357,7 +359,7 @@ def _init_with_new_array( self._sample_count = sample_count self._extended_properties = ExtendedPropertyDictionary() self._timing = Timing.empty - self._precision_timing = None + self._converted_timing_cache = {} self._scale_mode = NO_SCALING def _init_with_provided_array( @@ -414,7 +416,7 @@ def _init_with_provided_array( self._sample_count = sample_count self._extended_properties = ExtendedPropertyDictionary() self._timing = Timing.empty - self._precision_timing = None + self._converted_timing_cache = {} self._scale_mode = NO_SCALING @property @@ -579,32 +581,47 @@ def unit_description(self, value: str) -> None: raise invalid_arg_type("unit description", "str", value) self._extended_properties[UNIT_DESCRIPTION] = value + def _get_timing(self, requested_type: type[_TTiming]) -> _TTiming: + if isinstance(self._timing, requested_type): + return self._timing + value = cast(_TTiming, self._converted_timing_cache.get(requested_type)) + if value is None: + value = convert_timing(requested_type, self._timing) + self._converted_timing_cache[requested_type] = value + return value + + def _set_timing(self, value: _TTiming) -> None: + if self._timing is not value: + self._timing = value + self._converted_timing_cache.clear() + + def _validate_timing(self, value: _TTiming) -> None: + if value._timestamps is not None and len(value._timestamps) != self._sample_count: + raise ValueError( + "The number of irregular timestamps is not equal to the number of samples in the waveform.\n\n" + f"Number of timestamps: {len(value._timestamps)}\n" + f"Number of samples in the waveform: {self._sample_count}" + ) + @property def timing(self) -> Timing: """The timing information of the analog waveform. The default value is Timing.empty. """ - if self._timing is None: - if self._precision_timing is PrecisionTiming.empty: - self._timing = Timing.empty - elif self._precision_timing is not None: - self._timing = convert_timing(Timing, self._precision_timing) - else: - raise RuntimeError("The waveform has no timing information.") - return self._timing + return self._get_timing(Timing) @timing.setter def timing(self, value: Timing) -> None: if not isinstance(value, Timing): raise invalid_arg_type("timing information", "Timing object", value) - self._timing = value - self._precision_timing = None + self._validate_timing(value) + self._set_timing(value) @property def is_precision_timing_initialized(self) -> bool: - """Indicates whether the waveform's precision timing information is initialized.""" - return self._precision_timing is not None + """Indicates whether the waveform's timing information was set using precision timing.""" + return isinstance(self._timing, PrecisionTiming) @property def precision_timing(self) -> PrecisionTiming: @@ -622,21 +639,14 @@ def precision_timing(self) -> PrecisionTiming: set using AnalogWaveform.timing. Use AnalogWaveform.is_precision_timing_initialized to determine if AnalogWaveform.precision_timing has been initialized. """ - if self._precision_timing is None: - if self._timing is Timing.empty: - self._precision_timing = PrecisionTiming.empty - elif self._timing is not None: - self._precision_timing = convert_timing(PrecisionTiming, self._timing) - else: - raise RuntimeError("The waveform has no timing information.") - return self._precision_timing + return self._get_timing(PrecisionTiming) @precision_timing.setter def precision_timing(self, value: PrecisionTiming) -> None: if not isinstance(value, PrecisionTiming): raise invalid_arg_type("precision timing information", "PrecisionTiming object", value) - self._precision_timing = value - self._timing = None + self._validate_timing(value) + self._set_timing(value) @property def scale_mode(self) -> ScaleMode: @@ -648,3 +658,129 @@ def scale_mode(self, value: ScaleMode) -> None: if not isinstance(value, ScaleMode): raise invalid_arg_type("scale mode", "ScaleMode object", value) self._scale_mode = value + + def append( + self, + other: ( + npt.NDArray[_ScalarType_co] + | AnalogWaveform[_ScalarType_co] + | Sequence[AnalogWaveform[_ScalarType_co]] + ), + /, + timestamps: Sequence[dt.datetime] | Sequence[ht.datetime] | None = None, + ) -> None: + """Append data to the analog waveform. + + Args: + other: The array or waveform(s) to append. + timestamps: A sequence of timestamps. When the current waveform has + SampleIntervalMode.IRREGULAR, you must provide a sequence of timestamps with the + same length as the array. + + Raises: + TimingMismatchError: The current and other waveforms have incompatible timing. + ValueError: The other array has the wrong number of dimensions or the length of the + timestamps argument does not match the length of the other array. + TypeError: The data types of the current waveform and other array or waveform(s) do not + match, or an argument has the wrong data type. + + Warnings: + TimingMismatchWarning: The sample intervals of the waveform(s) do not match. + ScalingMismatchWarning: The scale modes of the waveform(s) do not match. + + When appending waveforms: + + * Timing information is merged based on the sample interval mode of the current + waveform: + + * SampleIntervalMode.NONE or SampleIntervalMode.REGULAR: The other waveform(s) must also + have SampleIntervalMode.NONE or SampleIntervalMode.REGULAR. If the sample interval does + not match, a TimingMismatchWarning is generated. Otherwise, the timing information of + the other waveform(s) is discarded. + + * SampleIntervalMode.IRREGULAR: The other waveforms(s) must also have + SampleIntervalMode.IRREGULAR. The timestamps of the other waveforms(s) are appended to + the current waveform's timing information. + + * Extended properties of the other waveform(s) are merged into the current waveform if they + are not already set in the current waveform. + + * If the scale mode of other waveform(s) does not match the scale mode of the current + waveform, a ScalingMismatchWarning is generated. Otherwise, the scaling information of the + other waveform(s) is discarded. + """ + if isinstance(other, np.ndarray): + self._append_array(other, timestamps) + elif isinstance(other, AnalogWaveform): + validate_unsupported_arg("timestamps", timestamps) + self._append_waveform(other) + elif isinstance(other, Sequence) and all(isinstance(x, AnalogWaveform) for x in other): + validate_unsupported_arg("timestamps", timestamps) + self._append_waveforms(other) + else: + raise invalid_arg_type("input", "array or waveform(s)", other) + + def _append_array( + self, + array: npt.NDArray[_ScalarType_co], + timestamps: Sequence[dt.datetime] | Sequence[ht.datetime] | None = None, + ) -> None: + if array.dtype != self.dtype: + raise TypeError( + "The data type of the input array must match the waveform data type.\n\n" + f"Input array data type: {array.dtype}\n" + f"Waveform data type: {self.dtype}" + ) + if array.ndim != 1: + raise ValueError( + "The input array must be a one-dimensional array.\n\n" + f"Number of dimensions: {array.ndim}" + ) + if timestamps is not None and len(array) != len(timestamps): + raise ValueError( + "The number of irregular timestamps must be equal to the input array length.\n\n" + f"Number of timestamps: {len(timestamps)}\n" + f"Array length: {len(array)}" + ) + + new_timing = self._timing._append_timestamps(timestamps) + + self._increase_capacity(len(array)) + self._set_timing(new_timing) + + offset = self._start_index + self._sample_count + self._data[offset : offset + len(array)] = array + self._sample_count += len(array) + + def _append_waveform(self, waveform: AnalogWaveform[_ScalarType_co]) -> None: + self._append_waveforms([waveform]) + + def _append_waveforms(self, waveforms: Sequence[AnalogWaveform[_ScalarType_co]]) -> None: + for waveform in waveforms: + if waveform.dtype != self.dtype: + raise TypeError( + "The data type of the input waveform must match the waveform data type.\n\n" + f"Input waveform data type: {waveform.dtype}\n" + f"Waveform data type: {self.dtype}" + ) + if waveform._scale_mode != self._scale_mode: + warnings.warn(scale_mode_mismatch()) + + new_timing = self._timing + for waveform in waveforms: + new_timing = new_timing._append_timing(waveform._timing) + + self._increase_capacity(sum(waveform.sample_count for waveform in waveforms)) + self._set_timing(new_timing) + + offset = self._start_index + self._sample_count + for waveform in waveforms: + self._data[offset : offset + waveform.sample_count] = waveform.raw_data + offset += waveform.sample_count + self._sample_count += waveform.sample_count + self._extended_properties._merge(waveform._extended_properties) + + def _increase_capacity(self, amount: int) -> None: + new_capacity = self._start_index + self._sample_count + amount + if new_capacity > self.capacity: + self.capacity = new_capacity diff --git a/src/nitypes/waveform/_exceptions.py b/src/nitypes/waveform/_exceptions.py new file mode 100644 index 00000000..f52fb788 --- /dev/null +++ b/src/nitypes/waveform/_exceptions.py @@ -0,0 +1,23 @@ +from __future__ import annotations + + +class TimingMismatchError(RuntimeError): + """Exception used when appending waveforms with mismatched timing.""" + + pass + + +def no_timestamp_information() -> RuntimeError: + """Create a RuntimeError for waveform timing with no timestamp information.""" + return RuntimeError( + "The waveform timing does not have valid timestamp information. " + "To obtain timestamps, the waveform must be irregular or must be initialized " + "with a valid time stamp and sample interval." + ) + + +def sample_interval_mode_mismatch() -> TimingMismatchError: + """Create a TimingMismatchError about mixing none/regular with irregular timing.""" + return TimingMismatchError( + "The timing of one or more waveforms does not match the timing of the current waveform." + ) diff --git a/src/nitypes/waveform/_extended_properties.py b/src/nitypes/waveform/_extended_properties.py index a598591c..987d023b 100644 --- a/src/nitypes/waveform/_extended_properties.py +++ b/src/nitypes/waveform/_extended_properties.py @@ -51,3 +51,7 @@ def __delitem__( # noqa: D105 - Missing docstring in magic method (auto-generat self, key: str, / ) -> None: operator.delitem(self._properties, key) + + def _merge(self, other: ExtendedPropertyDictionary) -> None: + for key, value in other.items(): + self._properties.setdefault(key, value) diff --git a/src/nitypes/waveform/_timing/__init__.py b/src/nitypes/waveform/_timing/__init__.py index 9695898e..0629fe70 100644 --- a/src/nitypes/waveform/_timing/__init__.py +++ b/src/nitypes/waveform/_timing/__init__.py @@ -1,8 +1,15 @@ """Waveform timing data types for NI Python APIs.""" -from nitypes.waveform._timing._base import BaseTiming, SampleIntervalMode +from nitypes.waveform._timing._base import BaseTiming from nitypes.waveform._timing._conversion import convert_timing from nitypes.waveform._timing._precision import PrecisionTiming +from nitypes.waveform._timing._sample_interval import SampleIntervalMode from nitypes.waveform._timing._standard import Timing -__all__ = ["BaseTiming", "convert_timing", "PrecisionTiming", "SampleIntervalMode", "Timing"] +__all__ = [ + "BaseTiming", + "convert_timing", + "PrecisionTiming", + "SampleIntervalMode", + "Timing", +] diff --git a/src/nitypes/waveform/_timing/_base.py b/src/nitypes/waveform/_timing/_base.py index e3b97939..7980c1b9 100644 --- a/src/nitypes/waveform/_timing/_base.py +++ b/src/nitypes/waveform/_timing/_base.py @@ -3,51 +3,99 @@ import datetime as dt import operator from abc import ABC, abstractmethod -from collections.abc import Generator, Iterable, Sequence -from enum import Enum +from collections.abc import Iterable, Sequence from typing import Generic, SupportsIndex, TypeVar -from nitypes._arguments import validate_unsupported_arg -from nitypes._exceptions import add_note, invalid_arg_type +from nitypes._exceptions import add_note +from nitypes._typing import Self +from nitypes.waveform._timing._sample_interval import ( + SampleIntervalMode, + SampleIntervalStrategy, + create_sample_interval_strategy, +) -class SampleIntervalMode(Enum): - """The sample interval mode that specifies how the waveform is sampled.""" +_TDateTime = TypeVar("_TDateTime", bound=dt.datetime) +_TTimeDelta = TypeVar("_TTimeDelta", bound=dt.timedelta) - NONE = 0 - """No sample interval.""" - REGULAR = 1 - """Regular sample interval.""" +class BaseTiming(ABC, Generic[_TDateTime, _TTimeDelta]): + """Base class for waveform timing information.""" + + @classmethod + @abstractmethod + def create_with_no_interval( + cls, timestamp: _TDateTime | None = None, time_offset: _TTimeDelta | None = None + ) -> Self: + """Create a waveform timing object with no sample interval. - IRREGULAR = 2 - """Irregular sample interval.""" + Args: + timestamp: A timestamp representing the start of an acquisition or a related + occurrence. + time_offset: The time difference between the timestamp and the time that the first + sample was acquired. + Returns: + A waveform timing object. + """ + raise NotImplementedError -# TODO: should these be constrained types? I guess we'll find out when we add NI-BTF types. -_TDateTime_co = TypeVar("_TDateTime_co", bound=dt.datetime) -_TTimeDelta_co = TypeVar("_TTimeDelta_co", bound=dt.timedelta) + @classmethod + @abstractmethod + def create_with_regular_interval( + cls, + sample_interval: _TTimeDelta, + timestamp: _TDateTime | None = None, + time_offset: _TTimeDelta | None = None, + ) -> Self: + """Create a waveform timing object with a regular sample interval. + Args: + sample_interval: The time difference between samples. + timestamp: A timestamp representing the start of an acquisition or a related + occurrence. + time_offset: The time difference between the timestamp and the time that the first + sample was acquired. -class BaseTiming(ABC, Generic[_TDateTime_co, _TTimeDelta_co]): - """Base class for waveform timing information.""" + Returns: + A waveform timing object. + """ + raise NotImplementedError + + @classmethod + @abstractmethod + def create_with_irregular_interval( + cls, + timestamps: Sequence[_TDateTime], + ) -> Self: + """Create a waveform timing object with an irregular sample interval. + + Args: + timestamps: A sequence containing a timestamp for each sample in the waveform, + specifying the time that the sample was acquired. + + Returns: + A waveform timing object. + """ + raise NotImplementedError @staticmethod @abstractmethod - def _get_datetime_type() -> type[_TDateTime_co]: + def _get_datetime_type() -> type[_TDateTime]: raise NotImplementedError() @staticmethod @abstractmethod - def _get_timedelta_type() -> type[_TTimeDelta_co]: + def _get_timedelta_type() -> type[_TTimeDelta]: raise NotImplementedError() @staticmethod @abstractmethod - def _get_default_time_offset() -> _TTimeDelta_co: + def _get_default_time_offset() -> _TTimeDelta: raise NotImplementedError() __slots__ = [ + "_sample_interval_strategy", "_sample_interval_mode", "_timestamp", "_time_offset", @@ -56,114 +104,48 @@ def _get_default_time_offset() -> _TTimeDelta_co: "__weakref__", ] + _sample_interval_strategy: SampleIntervalStrategy[_TDateTime, _TTimeDelta] _sample_interval_mode: SampleIntervalMode - _timestamp: _TDateTime_co | None - _time_offset: _TTimeDelta_co | None - _sample_interval: _TTimeDelta_co | None - _timestamps: list[_TDateTime_co] | None + _timestamp: _TDateTime | None + _time_offset: _TTimeDelta | None + _sample_interval: _TTimeDelta | None + _timestamps: list[_TDateTime] | None def __init__( self, sample_interval_mode: SampleIntervalMode, - timestamp: _TDateTime_co | None, - time_offset: _TTimeDelta_co | None, - sample_interval: _TTimeDelta_co | None, - timestamps: Sequence[_TDateTime_co] | None, + timestamp: _TDateTime | None, + time_offset: _TTimeDelta | None, + sample_interval: _TTimeDelta | None, + timestamps: Sequence[_TDateTime] | None, ) -> None: """Construct a base waveform timing object.""" - self._validate_init_args( - sample_interval_mode, timestamp, time_offset, sample_interval, timestamps - ) + sample_interval_strategy = create_sample_interval_strategy(sample_interval_mode) + try: + sample_interval_strategy.validate_init_args( + self, sample_interval_mode, timestamp, time_offset, sample_interval, timestamps + ) + except (TypeError, ValueError) as e: + add_note(e, f"Sample interval mode: {sample_interval_mode}") + raise if timestamps is not None and not isinstance(timestamps, list): timestamps = list(timestamps) + self._sample_interval_strategy = sample_interval_strategy self._sample_interval_mode = sample_interval_mode self._timestamp = timestamp self._time_offset = time_offset self._sample_interval = sample_interval self._timestamps = timestamps - def _validate_init_args( - self, - sample_interval_mode: SampleIntervalMode, - timestamp: _TDateTime_co | None, - time_offset: _TTimeDelta_co | None, - sample_interval: _TTimeDelta_co | None, - timestamps: Sequence[_TDateTime_co] | None, - ) -> None: - validate_func = self.__class__._VALIDATE_INIT_ARGS_FOR_MODE.get(sample_interval_mode) - if validate_func is None: - raise ValueError(f"Unsupported sample interval mode {sample_interval_mode}.") - - try: - validate_func(self, timestamp, time_offset, sample_interval, timestamps) - except (TypeError, ValueError) as e: - add_note(e, f"Sample interval mode: {sample_interval_mode}") - raise - - def _validate_init_args_none( - self, - timestamp: _TDateTime_co | None, - time_offset: _TTimeDelta_co | None, - sample_interval: _TTimeDelta_co | None, - timestamps: Sequence[_TDateTime_co] | None, - ) -> None: - datetime_type = self.__class__._get_datetime_type() - timedelta_type = self.__class__._get_timedelta_type() - if not isinstance(timestamp, (datetime_type, type(None))): - raise invalid_arg_type("timestamp", "datetime or None", timestamp) - if not isinstance(time_offset, (timedelta_type, type(None))): - raise invalid_arg_type("time offset", "timedelta or None", time_offset) - validate_unsupported_arg("sample interval", sample_interval) - validate_unsupported_arg("timestamps", timestamps) - - def _validate_init_args_regular( - self, - timestamp: _TDateTime_co | None, - time_offset: _TTimeDelta_co | None, - sample_interval: _TTimeDelta_co | None, - timestamps: Sequence[_TDateTime_co] | None, - ) -> None: - datetime_type = self.__class__._get_datetime_type() - timedelta_type = self.__class__._get_timedelta_type() - if not isinstance(timestamp, (datetime_type, type(None))): - raise invalid_arg_type("timestamp", "datetime or None", timestamp) - if not isinstance(time_offset, (timedelta_type, type(None))): - raise invalid_arg_type("time offset", "timedelta or None", time_offset) - if not isinstance(sample_interval, timedelta_type): - raise invalid_arg_type("sample interval", "timedelta", sample_interval) - validate_unsupported_arg("timestamps", timestamps) - - def _validate_init_args_irregular( - self, - timestamp: _TDateTime_co | None, - time_offset: _TTimeDelta_co | None, - sample_interval: _TTimeDelta_co | None, - timestamps: Sequence[_TDateTime_co] | None, - ) -> None: - datetime_type = self.__class__._get_datetime_type() - validate_unsupported_arg("timestamp", timestamp) - validate_unsupported_arg("time offset", time_offset) - validate_unsupported_arg("sample interval", sample_interval) - if not isinstance(timestamps, Sequence) or not all( - isinstance(ts, datetime_type) for ts in timestamps - ): - raise invalid_arg_type("timestamps", "sequence of datetime objects", timestamps) - - _VALIDATE_INIT_ARGS_FOR_MODE = { - SampleIntervalMode.NONE: _validate_init_args_none, - SampleIntervalMode.REGULAR: _validate_init_args_regular, - SampleIntervalMode.IRREGULAR: _validate_init_args_irregular, - } - @property def has_timestamp(self) -> bool: """Indicates whether the waveform timing has a timestamp.""" return self._timestamp is not None @property - def timestamp(self) -> _TDateTime_co: + def timestamp(self) -> _TDateTime: """A timestamp representing the start of an acquisition or a related occurrence.""" value = self._timestamp if value is None: @@ -171,12 +153,12 @@ def timestamp(self) -> _TDateTime_co: return value @property - def start_time(self) -> _TDateTime_co: + def start_time(self) -> _TDateTime: """The time that the first sample in the waveform was acquired.""" return self.timestamp + self.time_offset @property - def time_offset(self) -> _TTimeDelta_co: + def time_offset(self) -> _TTimeDelta: """The time difference between the timestamp and the first sample.""" value = self._time_offset if value is None: @@ -184,7 +166,7 @@ def time_offset(self) -> _TTimeDelta_co: return value @property - def sample_interval(self) -> _TTimeDelta_co: + def sample_interval(self) -> _TTimeDelta: """The time interval between samples.""" value = self._sample_interval if value is None: @@ -198,7 +180,7 @@ def sample_interval_mode(self) -> SampleIntervalMode: def get_timestamps( self, start_index: SupportsIndex, count: SupportsIndex - ) -> Iterable[_TDateTime_co]: + ) -> Iterable[_TDateTime]: """Retrieve the timestamps of the waveform samples. Args: @@ -216,31 +198,7 @@ def get_timestamps( if count < 0: raise ValueError("The count must be a non-negative integer.") - if self._sample_interval_mode == SampleIntervalMode.REGULAR and self.has_timestamp: - return self._generate_regular_timestamps(start_index, count) - elif self._sample_interval_mode == SampleIntervalMode.IRREGULAR: - assert self._timestamps is not None - if count > len(self._timestamps): - raise ValueError( - "The count must be less than or equal to the number of timestamps." - ) - return self._timestamps[start_index : start_index + count] - else: - raise RuntimeError( - "The waveform timing does not have valid timestamp information. " - "To obtain timestamps, the waveform must be irregular or must be initialized " - "with a valid time stamp and sample interval." - ) - - def _generate_regular_timestamps( - self, start_index: int, count: int - ) -> Generator[_TDateTime_co]: - sample_interval = self.sample_interval - timestamp = self.start_time + start_index * sample_interval - for i in range(count): - if i != 0: - timestamp += sample_interval - yield timestamp + return self._sample_interval_strategy.get_timestamps(self, start_index, count) def __eq__(self, value: object) -> bool: # noqa: D105 - Missing docstring in magic method if not isinstance(value, self.__class__): @@ -265,3 +223,18 @@ def __repr__(self) -> str: # noqa: D105 - Missing docstring in magic method if self._timestamps is not None: args.append(f"timestamps={self._timestamps!r}") return f"{self.__class__.__module__}.{self.__class__.__name__}({', '.join(args)})" + + def _append_timestamps(self, timestamps: Sequence[_TDateTime] | None) -> Self: + new_timing = self._sample_interval_strategy.append_timestamps(self, timestamps) + assert isinstance(new_timing, self.__class__) + return new_timing + + def _append_timing(self, other: Self) -> Self: + if not isinstance(other, self.__class__): + raise TypeError( + "The input waveform(s) must have the same waveform timing type as the current waveform." + ) + + new_timing = self._sample_interval_strategy.append_timing(self, other) + assert isinstance(new_timing, self.__class__) + return new_timing diff --git a/src/nitypes/waveform/_timing/_conversion.py b/src/nitypes/waveform/_timing/_conversion.py index 5d26dfb4..8debd2e7 100644 --- a/src/nitypes/waveform/_timing/_conversion.py +++ b/src/nitypes/waveform/_timing/_conversion.py @@ -10,11 +10,13 @@ from nitypes._exceptions import invalid_arg_type, invalid_requested_type from nitypes._typing import TypeAlias from nitypes.time._conversion import convert_datetime, convert_timedelta +from nitypes.waveform._timing._base import BaseTiming from nitypes.waveform._timing._precision import PrecisionTiming from nitypes.waveform._timing._standard import Timing -_AnyTiming: TypeAlias = Union[Timing, PrecisionTiming] -_TTiming = TypeVar("_TTiming", Timing, PrecisionTiming) + +_AnyTiming: TypeAlias = Union[BaseTiming[Any, Any], Timing, PrecisionTiming] +_TTiming = TypeVar("_TTiming", bound=BaseTiming[Any, Any]) def convert_timing(requested_type: type[_TTiming], value: _AnyTiming, /) -> _TTiming: @@ -37,6 +39,8 @@ def _(value: Timing, /) -> Timing: @_convert_to_standard_timing.register def _(value: PrecisionTiming, /) -> Timing: + if value is PrecisionTiming.empty: + return Timing.empty return Timing( value._sample_interval_mode, None if value._timestamp is None else convert_datetime(dt.datetime, value._timestamp), @@ -65,6 +69,8 @@ def _convert_to_precision_timing(value: object, /) -> PrecisionTiming: @_convert_to_precision_timing.register def _(value: Timing, /) -> PrecisionTiming: + if value is Timing.empty: + return PrecisionTiming.empty return PrecisionTiming( value._sample_interval_mode, None if value._timestamp is None else convert_datetime(ht.datetime, value._timestamp), diff --git a/src/nitypes/waveform/_timing/_precision.py b/src/nitypes/waveform/_timing/_precision.py index dd8a5f64..d39f3bd6 100644 --- a/src/nitypes/waveform/_timing/_precision.py +++ b/src/nitypes/waveform/_timing/_precision.py @@ -5,7 +5,9 @@ import hightime as ht -from nitypes.waveform._timing._base import BaseTiming, SampleIntervalMode +from nitypes._typing import override +from nitypes.waveform._timing._base import BaseTiming +from nitypes.waveform._timing._sample_interval import SampleIntervalMode class PrecisionTiming(BaseTiming[ht.datetime, ht.timedelta]): @@ -17,67 +19,42 @@ class PrecisionTiming(BaseTiming[ht.datetime, ht.timedelta]): _DEFAULT_TIME_OFFSET = ht.timedelta() empty: ClassVar[PrecisionTiming] + """A waveform timing object with no timestamp, time offset, or sample interval.""" + @override @staticmethod - def create_with_no_interval( + def create_with_no_interval( # noqa: D102 - Missing docstring in public method - override timestamp: ht.datetime | None = None, time_offset: ht.timedelta | None = None ) -> PrecisionTiming: - """Create a waveform timing object with no sample interval. - - Args: - timestamp: A timestamp representing the start of an acquisition or a related - occurrence. - time_offset: The time difference between the timestamp and the time that the first - sample was acquired. - - Returns: - A waveform timing object. - """ return PrecisionTiming(SampleIntervalMode.NONE, timestamp, time_offset) + @override @staticmethod - def create_with_regular_interval( + def create_with_regular_interval( # noqa: D102 - Missing docstring in public method - override sample_interval: ht.timedelta, timestamp: ht.datetime | None = None, time_offset: ht.timedelta | None = None, ) -> PrecisionTiming: - """Create a waveform timing object with a regular sample interval. - - Args: - sample_interval: The time difference between samples. - timestamp: A timestamp representing the start of an acquisition or a related - occurrence. - time_offset: The time difference between the timestamp and the time that the first - sample was acquired. - - Returns: - A waveform timing object. - """ return PrecisionTiming(SampleIntervalMode.REGULAR, timestamp, time_offset, sample_interval) + @override @staticmethod - def create_with_irregular_interval( + def create_with_irregular_interval( # noqa: D102 - Missing docstring in public method - override timestamps: Sequence[ht.datetime], ) -> PrecisionTiming: - """Create a waveform timing object with an irregular sample interval. - - Args: - timestamps: A sequence containing a timestamp for each sample in the waveform, - specifying the time that the sample was acquired. - - Returns: - A waveform timing object. - """ return PrecisionTiming(SampleIntervalMode.IRREGULAR, timestamps=timestamps) + @override @staticmethod def _get_datetime_type() -> type[ht.datetime]: return ht.datetime + @override @staticmethod def _get_timedelta_type() -> type[ht.timedelta]: return ht.timedelta + @override @staticmethod def _get_default_time_offset() -> ht.timedelta: return PrecisionTiming._DEFAULT_TIME_OFFSET @@ -101,4 +78,3 @@ def __init__( PrecisionTiming.empty = PrecisionTiming.create_with_no_interval() -"""A waveform timing object with no timestamp, time offset, or sample interval.""" diff --git a/src/nitypes/waveform/_timing/_sample_interval/__init__.py b/src/nitypes/waveform/_timing/_sample_interval/__init__.py new file mode 100644 index 00000000..b9db8076 --- /dev/null +++ b/src/nitypes/waveform/_timing/_sample_interval/__init__.py @@ -0,0 +1,44 @@ +"""Sample interval strategies for waveform timing.""" + +from typing import Any + +from nitypes._exceptions import invalid_arg_value +from nitypes.waveform._timing._sample_interval._base import SampleIntervalStrategy +from nitypes.waveform._timing._sample_interval._irregular import ( + IrregularSampleIntervalStrategy, +) +from nitypes.waveform._timing._sample_interval._mode import SampleIntervalMode +from nitypes.waveform._timing._sample_interval._none import NoneSampleIntervalStrategy +from nitypes.waveform._timing._sample_interval._regular import ( + RegularSampleIntervalStrategy, +) + +__all__ = [ + "create_sample_interval_strategy", + "IrregularSampleIntervalStrategy", + "NoneSampleIntervalStrategy", + "RegularSampleIntervalStrategy", + "SampleIntervalMode", + "SampleIntervalStrategy", +] + + +def create_sample_interval_strategy( + sample_interval_mode: SampleIntervalMode, +) -> SampleIntervalStrategy[Any, Any]: + """Create a sample interval strategy for the specified mode.""" + strategy_type = _SAMPLE_INTERVAL_STRATEGY_TYPE_FOR_MODE.get(sample_interval_mode) + if strategy_type is None: + raise invalid_arg_value( + "sample interval mode", "SampleIntervalMode object", sample_interval_mode + ) + return strategy_type() + + +_SAMPLE_INTERVAL_STRATEGY_TYPE_FOR_MODE: dict[ + SampleIntervalMode, type[SampleIntervalStrategy[Any, Any]] +] = { + SampleIntervalMode.NONE: NoneSampleIntervalStrategy, + SampleIntervalMode.REGULAR: RegularSampleIntervalStrategy, + SampleIntervalMode.IRREGULAR: IrregularSampleIntervalStrategy, +} diff --git a/src/nitypes/waveform/_timing/_sample_interval/_base.py b/src/nitypes/waveform/_timing/_sample_interval/_base.py new file mode 100644 index 00000000..b3e25791 --- /dev/null +++ b/src/nitypes/waveform/_timing/_sample_interval/_base.py @@ -0,0 +1,61 @@ +from __future__ import annotations + +import datetime as dt +from abc import ABC, abstractmethod +from collections.abc import Iterable, Sequence +from typing import Generic, TypeVar, TYPE_CHECKING + +from nitypes.waveform._timing._sample_interval._mode import SampleIntervalMode + + +if TYPE_CHECKING: + from nitypes.waveform._timing._base import BaseTiming # circular import + +_TDateTime = TypeVar("_TDateTime", bound=dt.datetime) +_TTimeDelta = TypeVar("_TTimeDelta", bound=dt.timedelta) + + +class SampleIntervalStrategy(ABC, Generic[_TDateTime, _TTimeDelta]): + """Implements SampleIntervalMode specific behavior.""" + + # Note that timing is always passed as a parameter. The timing object has a reference to the + # strategy, so saving a reference to the timing object would introduce a reference cycle. + __slots__ = () + + @abstractmethod + def validate_init_args( + self, + timing: BaseTiming[_TDateTime, _TTimeDelta], + sample_interval_mode: SampleIntervalMode, + timestamp: _TDateTime | None, + time_offset: _TTimeDelta | None, + sample_interval: _TTimeDelta | None, + timestamps: Sequence[_TDateTime] | None, + ) -> None: + """Validate the BaseTiming.__init__ arguments for this mode.""" + raise NotImplementedError + + @abstractmethod + def get_timestamps( + self, timing: BaseTiming[_TDateTime, _TTimeDelta], start_index: int, count: int + ) -> Iterable[_TDateTime]: + """Get or generate timestamps for the specified samples.""" + raise NotImplementedError + + @abstractmethod + def append_timestamps( + self, + timing: BaseTiming[_TDateTime, _TTimeDelta], + timestamps: Sequence[_TDateTime] | None, + ) -> BaseTiming[_TDateTime, _TTimeDelta]: + """Append timestamps and return a new waveform timing if needed.""" + raise NotImplementedError + + @abstractmethod + def append_timing( + self, + timing: BaseTiming[_TDateTime, _TTimeDelta], + other: BaseTiming[_TDateTime, _TTimeDelta], + ) -> BaseTiming[_TDateTime, _TTimeDelta]: + """Append timing and return a new waveform timing if needed.""" + raise NotImplementedError diff --git a/src/nitypes/waveform/_timing/_sample_interval/_irregular.py b/src/nitypes/waveform/_timing/_sample_interval/_irregular.py new file mode 100644 index 00000000..7ac3a2f4 --- /dev/null +++ b/src/nitypes/waveform/_timing/_sample_interval/_irregular.py @@ -0,0 +1,128 @@ +from __future__ import annotations + +import datetime as dt +from collections.abc import Iterable, Sequence +from enum import Enum +from typing import TYPE_CHECKING, TypeVar + +from nitypes._arguments import validate_unsupported_arg +from nitypes._exceptions import invalid_arg_type +from nitypes.waveform._exceptions import ( + TimingMismatchError, + sample_interval_mode_mismatch, +) +from nitypes.waveform._timing._sample_interval._base import SampleIntervalStrategy +from nitypes.waveform._timing._sample_interval._mode import SampleIntervalMode + +if TYPE_CHECKING: + from nitypes.waveform._timing._base import BaseTiming # circular import + +_TDateTime = TypeVar("_TDateTime", bound=dt.datetime) +_TTimeDelta = TypeVar("_TTimeDelta", bound=dt.timedelta) + + +class _Direction(Enum): + INCREASING = -1 + UNKNOWN = 0 + DECREASING = 1 + + +def _are_timestamps_monotonic(timestamps: Sequence[_TDateTime]) -> bool: + direction = _Direction.UNKNOWN + for i in range(1, len(timestamps)): + comparison = _get_direction(timestamps[i - 1], timestamps[i]) + if comparison == _Direction.UNKNOWN: + continue + + if direction == _Direction.UNKNOWN: + direction = comparison + elif comparison != direction: + return False + return True + + +def _get_direction(left: _TDateTime, right: _TDateTime) -> _Direction: + if left < right: + return _Direction.INCREASING + if right < left: + return _Direction.DECREASING + return _Direction.UNKNOWN + + +class IrregularSampleIntervalStrategy(SampleIntervalStrategy[_TDateTime, _TTimeDelta]): + """Implements SampleIntervalMode.IRREGULAR specific behavior.""" + + def validate_init_args( # noqa: D102 - Missing docstring in public method - override + self, + timing: BaseTiming[_TDateTime, _TTimeDelta], + sample_interval_mode: SampleIntervalMode, + timestamp: _TDateTime | None, + time_offset: _TTimeDelta | None, + sample_interval: _TTimeDelta | None, + timestamps: Sequence[_TDateTime] | None, + ) -> None: + datetime_type = timing.__class__._get_datetime_type() + validate_unsupported_arg("timestamp", timestamp) + validate_unsupported_arg("time offset", time_offset) + validate_unsupported_arg("sample interval", sample_interval) + if not isinstance(timestamps, Sequence) or not all( + isinstance(ts, datetime_type) for ts in timestamps + ): + raise invalid_arg_type("timestamps", "sequence of datetime objects", timestamps) + if not _are_timestamps_monotonic(timestamps): + raise ValueError("The timestamps must be in ascending or descending order.") + + def get_timestamps( # noqa: D102 - Missing docstring in public method - override + self, timing: BaseTiming[_TDateTime, _TTimeDelta], start_index: int, count: int + ) -> Iterable[_TDateTime]: + assert timing._timestamps is not None + if count > len(timing._timestamps): + raise ValueError("The count must be less than or equal to the number of timestamps.") + return timing._timestamps[start_index : start_index + count] + + def append_timestamps( # noqa: D102 - Missing docstring in public method - override + self, + timing: BaseTiming[_TDateTime, _TTimeDelta], + timestamps: Sequence[_TDateTime] | None, + ) -> BaseTiming[_TDateTime, _TTimeDelta]: + assert timing._timestamps is not None + + if timestamps is None: + raise TimingMismatchError( + "The timestamps argument is required when appending to a waveform with irregular timing." + ) + + datetime_type = timing.__class__._get_datetime_type() + if not all(isinstance(ts, datetime_type) for ts in timestamps): + raise TypeError( + "The timestamp data type must match the timing information of the current waveform." + ) + + if len(timestamps) == 0: + return timing + else: + if not isinstance(timestamps, list): + timestamps = list(timestamps) + + return timing.__class__.create_with_irregular_interval(timing._timestamps + timestamps) + + def append_timing( # noqa: D102 - Missing docstring in public method - override + self, + timing: BaseTiming[_TDateTime, _TTimeDelta], + other: BaseTiming[_TDateTime, _TTimeDelta], + ) -> BaseTiming[_TDateTime, _TTimeDelta]: + if other._sample_interval_mode != SampleIntervalMode.IRREGULAR: + raise sample_interval_mode_mismatch() + + assert timing._timestamps is not None and other._timestamps is not None + + if len(timing._timestamps) == 0: + return other + elif len(other._timestamps) == 0: + return timing + else: + # The constructor will verify that the combined list of timestamps is monotonic. This is + # not optimal for a large number of appends. + return timing.__class__.create_with_irregular_interval( + timing._timestamps + other._timestamps + ) diff --git a/src/nitypes/waveform/_timing/_sample_interval/_mode.py b/src/nitypes/waveform/_timing/_sample_interval/_mode.py new file mode 100644 index 00000000..22858916 --- /dev/null +++ b/src/nitypes/waveform/_timing/_sample_interval/_mode.py @@ -0,0 +1,16 @@ +from __future__ import annotations + +from enum import Enum + + +class SampleIntervalMode(Enum): + """The sample interval mode that specifies how the waveform is sampled.""" + + NONE = 0 + """No sample interval.""" + + REGULAR = 1 + """Regular sample interval.""" + + IRREGULAR = 2 + """Irregular sample interval.""" diff --git a/src/nitypes/waveform/_timing/_sample_interval/_none.py b/src/nitypes/waveform/_timing/_sample_interval/_none.py new file mode 100644 index 00000000..9577c0d9 --- /dev/null +++ b/src/nitypes/waveform/_timing/_sample_interval/_none.py @@ -0,0 +1,72 @@ +from __future__ import annotations + +import datetime as dt +import warnings +from collections.abc import Iterable, Sequence +from typing import TYPE_CHECKING, TypeVar + +from nitypes._arguments import validate_unsupported_arg +from nitypes._exceptions import add_note, invalid_arg_type +from nitypes.waveform._exceptions import ( + no_timestamp_information, + sample_interval_mode_mismatch, +) +from nitypes.waveform._timing._sample_interval._base import SampleIntervalStrategy +from nitypes.waveform._timing._sample_interval._mode import SampleIntervalMode +from nitypes.waveform._warnings import sample_interval_mismatch + +if TYPE_CHECKING: + from nitypes.waveform._timing._base import BaseTiming # circular import + +_TDateTime = TypeVar("_TDateTime", bound=dt.datetime) +_TTimeDelta = TypeVar("_TTimeDelta", bound=dt.timedelta) + + +class NoneSampleIntervalStrategy(SampleIntervalStrategy[_TDateTime, _TTimeDelta]): + """Implements SampleIntervalMode.NONE specific behavior.""" + + def validate_init_args( # noqa: D102 - Missing docstring in public method - override + self, + timing: BaseTiming[_TDateTime, _TTimeDelta], + sample_interval_mode: SampleIntervalMode, + timestamp: _TDateTime | None, + time_offset: _TTimeDelta | None, + sample_interval: _TTimeDelta | None, + timestamps: Sequence[_TDateTime] | None, + ) -> None: + datetime_type = timing.__class__._get_datetime_type() + timedelta_type = timing.__class__._get_timedelta_type() + if not isinstance(timestamp, (datetime_type, type(None))): + raise invalid_arg_type("timestamp", "datetime or None", timestamp) + if not isinstance(time_offset, (timedelta_type, type(None))): + raise invalid_arg_type("time offset", "timedelta or None", time_offset) + validate_unsupported_arg("sample interval", sample_interval) + validate_unsupported_arg("timestamps", timestamps) + + def get_timestamps( # noqa: D102 - Missing docstring in public method - override + self, timing: BaseTiming[_TDateTime, _TTimeDelta], start_index: int, count: int + ) -> Iterable[_TDateTime]: + raise no_timestamp_information() + + def append_timestamps( # noqa: D102 - Missing docstring in public method - override + self, + timing: BaseTiming[_TDateTime, _TTimeDelta], + timestamps: Sequence[_TDateTime] | None, + ) -> BaseTiming[_TDateTime, _TTimeDelta]: + try: + validate_unsupported_arg("timestamps", timestamps) + except (TypeError, ValueError) as e: + add_note(e, f"Sample interval mode: {timing.sample_interval_mode}") + raise + return timing + + def append_timing( # noqa: D102 - Missing docstring in public method - override + self, + timing: BaseTiming[_TDateTime, _TTimeDelta], + other: BaseTiming[_TDateTime, _TTimeDelta], + ) -> BaseTiming[_TDateTime, _TTimeDelta]: + if other._sample_interval_mode not in (SampleIntervalMode.NONE, SampleIntervalMode.REGULAR): + raise sample_interval_mode_mismatch() + if timing._sample_interval != other._sample_interval: + warnings.warn(sample_interval_mismatch()) + return timing diff --git a/src/nitypes/waveform/_timing/_sample_interval/_regular.py b/src/nitypes/waveform/_timing/_sample_interval/_regular.py new file mode 100644 index 00000000..44a40235 --- /dev/null +++ b/src/nitypes/waveform/_timing/_sample_interval/_regular.py @@ -0,0 +1,85 @@ +from __future__ import annotations + +import datetime as dt +import warnings +from collections.abc import Generator, Iterable, Sequence +from typing import TYPE_CHECKING, TypeVar + +from nitypes._arguments import validate_unsupported_arg +from nitypes._exceptions import add_note, invalid_arg_type +from nitypes.waveform._exceptions import ( + no_timestamp_information, + sample_interval_mode_mismatch, +) +from nitypes.waveform._timing._sample_interval._base import SampleIntervalStrategy +from nitypes.waveform._timing._sample_interval._mode import SampleIntervalMode +from nitypes.waveform._warnings import sample_interval_mismatch + +if TYPE_CHECKING: + from nitypes.waveform._timing._base import BaseTiming # circular import + +_TDateTime = TypeVar("_TDateTime", bound=dt.datetime) +_TTimeDelta = TypeVar("_TTimeDelta", bound=dt.timedelta) + + +class RegularSampleIntervalStrategy(SampleIntervalStrategy[_TDateTime, _TTimeDelta]): + """Implements SampleIntervalMode.REGULAR specific behavior.""" + + def validate_init_args( # noqa: D102 - Missing docstring in public method - override + self, + timing: BaseTiming[_TDateTime, _TTimeDelta], + sample_interval_mode: SampleIntervalMode, + timestamp: _TDateTime | None, + time_offset: _TTimeDelta | None, + sample_interval: _TTimeDelta | None, + timestamps: Sequence[_TDateTime] | None, + ) -> None: + datetime_type = timing.__class__._get_datetime_type() + timedelta_type = timing.__class__._get_timedelta_type() + if not isinstance(timestamp, (datetime_type, type(None))): + raise invalid_arg_type("timestamp", "datetime or None", timestamp) + if not isinstance(time_offset, (timedelta_type, type(None))): + raise invalid_arg_type("time offset", "timedelta or None", time_offset) + if not isinstance(sample_interval, timedelta_type): + raise invalid_arg_type("sample interval", "timedelta", sample_interval) + validate_unsupported_arg("timestamps", timestamps) + + def get_timestamps( # noqa: D102 - Missing docstring in public method - override + self, timing: BaseTiming[_TDateTime, _TTimeDelta], start_index: int, count: int + ) -> Iterable[_TDateTime]: + if timing.has_timestamp: + return self._generate_regular_timestamps(timing, start_index, count) + raise no_timestamp_information() + + def _generate_regular_timestamps( + self, timing: BaseTiming[_TDateTime, _TTimeDelta], start_index: int, count: int + ) -> Generator[_TDateTime]: + sample_interval = timing.sample_interval + timestamp = timing.start_time + start_index * sample_interval + for i in range(count): + if i != 0: + timestamp += sample_interval + yield timestamp + + def append_timestamps( # noqa: D102 - Missing docstring in public method - override + self, + timing: BaseTiming[_TDateTime, _TTimeDelta], + timestamps: Sequence[_TDateTime] | None, + ) -> BaseTiming[_TDateTime, _TTimeDelta]: + try: + validate_unsupported_arg("timestamps", timestamps) + except (TypeError, ValueError) as e: + add_note(e, f"Sample interval mode: {timing.sample_interval_mode}") + raise + return timing + + def append_timing( # noqa: D102 - Missing docstring in public method - override + self, + timing: BaseTiming[_TDateTime, _TTimeDelta], + other: BaseTiming[_TDateTime, _TTimeDelta], + ) -> BaseTiming[_TDateTime, _TTimeDelta]: + if other._sample_interval_mode not in (SampleIntervalMode.NONE, SampleIntervalMode.REGULAR): + raise sample_interval_mode_mismatch() + if timing._sample_interval != other._sample_interval: + warnings.warn(sample_interval_mismatch()) + return timing diff --git a/src/nitypes/waveform/_timing/_standard.py b/src/nitypes/waveform/_timing/_standard.py index 8f7e918f..2534ba3e 100644 --- a/src/nitypes/waveform/_timing/_standard.py +++ b/src/nitypes/waveform/_timing/_standard.py @@ -4,7 +4,9 @@ from collections.abc import Sequence from typing import ClassVar -from nitypes.waveform._timing._base import BaseTiming, SampleIntervalMode +from nitypes._typing import override +from nitypes.waveform._timing._base import BaseTiming +from nitypes.waveform._timing._sample_interval import SampleIntervalMode class Timing(BaseTiming[dt.datetime, dt.timedelta]): @@ -17,68 +19,42 @@ class Timing(BaseTiming[dt.datetime, dt.timedelta]): _DEFAULT_TIME_OFFSET = dt.timedelta() empty: ClassVar[Timing] + """A waveform timing object with no timestamp, time offset, or sample interval.""" - # TODO: can these be classmethods in BaseTiming? + @override @staticmethod - def create_with_no_interval( + def create_with_no_interval( # noqa: D102 - Missing docstring in public method - override timestamp: dt.datetime | None = None, time_offset: dt.timedelta | None = None ) -> Timing: - """Create a waveform timing object with no sample interval. - - Args: - timestamp: A timestamp representing the start of an acquisition or a related - occurrence. - time_offset: The time difference between the timestamp and the time that the first - sample was acquired. - - Returns: - A waveform timing object. - """ return Timing(SampleIntervalMode.NONE, timestamp, time_offset) + @override @staticmethod - def create_with_regular_interval( + def create_with_regular_interval( # noqa: D102 - Missing docstring in public method - override sample_interval: dt.timedelta, timestamp: dt.datetime | None = None, time_offset: dt.timedelta | None = None, ) -> Timing: - """Create a waveform timing object with a regular sample interval. - - Args: - sample_interval: The time difference between samples. - timestamp: A timestamp representing the start of an acquisition or a related - occurrence. - time_offset: The time difference between the timestamp and the time that the first - sample was acquired. - - Returns: - A waveform timing object. - """ return Timing(SampleIntervalMode.REGULAR, timestamp, time_offset, sample_interval) + @override @staticmethod - def create_with_irregular_interval( + def create_with_irregular_interval( # noqa: D102 - Missing docstring in public method - override timestamps: Sequence[dt.datetime], ) -> Timing: - """Create a waveform timing object with an irregular sample interval. - - Args: - timestamps: A sequence containing a timestamp for each sample in the waveform, - specifying the time that the sample was acquired. - - Returns: - A waveform timing object. - """ return Timing(SampleIntervalMode.IRREGULAR, timestamps=timestamps) + @override @staticmethod def _get_datetime_type() -> type[dt.datetime]: return dt.datetime + @override @staticmethod def _get_timedelta_type() -> type[dt.timedelta]: return dt.timedelta + @override @staticmethod def _get_default_time_offset() -> dt.timedelta: return Timing._DEFAULT_TIME_OFFSET @@ -102,4 +78,3 @@ def __init__( Timing.empty = Timing.create_with_no_interval() -"""A waveform timing object with no timestamp, time offset, or sample interval.""" diff --git a/src/nitypes/waveform/_warnings.py b/src/nitypes/waveform/_warnings.py new file mode 100644 index 00000000..0651e1ed --- /dev/null +++ b/src/nitypes/waveform/_warnings.py @@ -0,0 +1,27 @@ +from __future__ import annotations + + +class ScalingMismatchWarning(RuntimeWarning): + """Warning used when appending waveforms with mismatched scaling information.""" + + pass + + +class TimingMismatchWarning(RuntimeWarning): + """Warning used when appending waveforms with mismatched timing information.""" + + pass + + +def sample_interval_mismatch() -> TimingMismatchWarning: + """Create a TimingMismatchWarning about appending waveforms with mismatched sample intervals.""" + return TimingMismatchWarning( + "The sample interval of one or more waveforms does not match the sample interval of the current waveform." + ) + + +def scale_mode_mismatch() -> ScalingMismatchWarning: + """Create a ScalingMismatchwarning about appending waveforms with mismatched scale modes.""" + return ScalingMismatchWarning( + "The scale mode of one or more waveforms does not match the scale mode of the current waveform." + ) diff --git a/tests/unit/waveform/_timing/test_precision.py b/tests/unit/waveform/_timing/test_precision.py index 90e29e0d..37ef1cc1 100644 --- a/tests/unit/waveform/_timing/test_precision.py +++ b/tests/unit/waveform/_timing/test_precision.py @@ -166,25 +166,79 @@ def test___sample_interval_and_time_offset___create_with_regular_interval___crea ############################################################################### # create_with_irregular_interval ############################################################################### -def test___timestamps___create_with_irregular_interval___creates_waveform_timing_with_timestamps() -> ( +@pytest.mark.parametrize( + "time_offsets", + [ + [], + [ht.timedelta(0)], + [ht.timedelta(0), ht.timedelta(0)], + [ht.timedelta(0), ht.timedelta(1)], + [ht.timedelta(0), ht.timedelta(1), ht.timedelta(2)], + [ + ht.timedelta(0), + ht.timedelta(1), + ht.timedelta(2), + ht.timedelta(3), + ], + [ + ht.timedelta(3), + ht.timedelta(2), + ht.timedelta(1), + ht.timedelta(0), + ], + [ht.timedelta(0, 0, 1), ht.timedelta(0, 1, 0), ht.timedelta(1, 0, 0)], + [ht.timedelta(1, 0, 0), ht.timedelta(0, 1, 0), ht.timedelta(0, 0, 1)], + [ + ht.timedelta(0), + ht.timedelta(1), + ht.timedelta(1), + ht.timedelta(2), + ], + ], +) +def test___monotonic_timestamps___create_with_irregular_interval___creates_waveform_timing_with_timestamps( + time_offsets: list[ht.timedelta], +) -> None: + start_time = ht.datetime.now(dt.timezone.utc) + timestamps = [start_time + offset for offset in time_offsets] + + timing = PrecisionTiming.create_with_irregular_interval(timestamps) + + assert_type(timing, PrecisionTiming) + assert timing.sample_interval_mode == SampleIntervalMode.IRREGULAR + assert timing._timestamps == timestamps + + +@pytest.mark.parametrize( + "time_offsets", + [ + [ht.timedelta(0), ht.timedelta(1), ht.timedelta(0)], + [ht.timedelta(1), ht.timedelta(0), ht.timedelta(1)], + ], +) +def test___non_monotonic_timestamps___create_with_irregular_interval___raises_value_error( + time_offsets: list[ht.timedelta], +) -> None: + start_time = ht.datetime.now(dt.timezone.utc) + timestamps = [start_time + offset for offset in time_offsets] + + with pytest.raises(ValueError) as exc: + _ = PrecisionTiming.create_with_irregular_interval(timestamps) + + assert exc.value.args[0].startswith("The timestamps must be in ascending or descending order.") + + +def test___timestamps_tuple___create_with_irregular_interval___creates_waveform_timing_with_timestamps() -> ( None ): start_time = ht.datetime.now(dt.timezone.utc) - timestamps = [ - start_time, - start_time + ht.timedelta(seconds=1), - start_time + ht.timedelta(seconds=2.3), - start_time + ht.timedelta(seconds=2.5), - ] + timestamps = (start_time,) timing = PrecisionTiming.create_with_irregular_interval(timestamps) assert_type(timing, PrecisionTiming) - assert not timing.has_timestamp - assert timing.time_offset == ht.timedelta() - assert timing._sample_interval is None assert timing.sample_interval_mode == SampleIntervalMode.IRREGULAR - assert timing._timestamps == timestamps + assert timing._timestamps == list(timestamps) ############################################################################### @@ -375,3 +429,76 @@ def test___different_value___equality___not_equal( ) def test___various_values___repr___looks_ok(value: PrecisionTiming, expected_repr: str) -> None: assert repr(value) == expected_repr + + +############################################################################### +# _append_timing +############################################################################### +@pytest.mark.parametrize( + "left_offsets, right_offsets", + [ + ([], []), + ([ht.timedelta(0)], []), + ([ht.timedelta(0), ht.timedelta(1)], []), + ([ht.timedelta(0), ht.timedelta(1), ht.timedelta(2)], []), + ([ht.timedelta(0)], [ht.timedelta(1)]), + ([ht.timedelta(0)], [ht.timedelta(1), ht.timedelta(2)]), + ( + [ht.timedelta(0), ht.timedelta(1)], + [ht.timedelta(2), ht.timedelta(3)], + ), + ( + [ht.timedelta(3), ht.timedelta(2)], + [ht.timedelta(1), ht.timedelta(0)], + ), + ( + [ht.timedelta(0), ht.timedelta(1)], + [ht.timedelta(1), ht.timedelta(2)], + ), + ( + [ht.timedelta(2), ht.timedelta(1)], + [ht.timedelta(1), ht.timedelta(0)], + ), + ], +) +def test___monotonic_timestamps___append_timing___appends_timestamps( + left_offsets: list[ht.timedelta], + right_offsets: list[ht.timedelta], +) -> None: + start_time = ht.datetime.now(dt.timezone.utc) + left_timestamps = [start_time + offset for offset in left_offsets] + right_timestamps = [start_time + offset for offset in right_offsets] + left_timing = PrecisionTiming.create_with_irregular_interval(left_timestamps) + right_timing = PrecisionTiming.create_with_irregular_interval(right_timestamps) + + new_timing = left_timing._append_timing(right_timing) + + assert_type(new_timing, PrecisionTiming) + assert isinstance(new_timing, PrecisionTiming) + assert new_timing._timestamps == left_timestamps + right_timestamps + + +@pytest.mark.parametrize( + "left_offsets, right_offsets", + [ + ([ht.timedelta(0)], [ht.timedelta(1), ht.timedelta(0)]), + ([ht.timedelta(1)], [ht.timedelta(0), ht.timedelta(2)]), + ([ht.timedelta(0), ht.timedelta(1)], [ht.timedelta(0)]), + ([ht.timedelta(1), ht.timedelta(0)], [ht.timedelta(1)]), + ([ht.timedelta(0), ht.timedelta(1)], [ht.timedelta(2), ht.timedelta(0)]), + ], +) +def test___non_monotonic_timestamps___append_timing___raises_value_error( + left_offsets: list[ht.timedelta], + right_offsets: list[ht.timedelta], +) -> None: + start_time = ht.datetime.now(dt.timezone.utc) + left_timestamps = [start_time + offset for offset in left_offsets] + right_timestamps = [start_time + offset for offset in right_offsets] + left_timing = PrecisionTiming.create_with_irregular_interval(left_timestamps) + right_timing = PrecisionTiming.create_with_irregular_interval(right_timestamps) + + with pytest.raises(ValueError) as exc: + _ = left_timing._append_timing(right_timing) + + assert exc.value.args[0].startswith("The timestamps must be in ascending or descending order.") diff --git a/tests/unit/waveform/_timing/test_standard.py b/tests/unit/waveform/_timing/test_standard.py index 1960d761..cf5c2ce6 100644 --- a/tests/unit/waveform/_timing/test_standard.py +++ b/tests/unit/waveform/_timing/test_standard.py @@ -165,25 +165,79 @@ def test___sample_interval_and_time_offset___create_with_regular_interval___crea ############################################################################### # create_with_irregular_interval ############################################################################### -def test___timestamps___create_with_irregular_interval___creates_waveform_timing_with_timestamps() -> ( +@pytest.mark.parametrize( + "time_offsets", + [ + [], + [dt.timedelta(0)], + [dt.timedelta(0), dt.timedelta(0)], + [dt.timedelta(0), dt.timedelta(1)], + [dt.timedelta(0), dt.timedelta(1), dt.timedelta(2)], + [ + dt.timedelta(0), + dt.timedelta(1), + dt.timedelta(2), + dt.timedelta(3), + ], + [ + dt.timedelta(3), + dt.timedelta(2), + dt.timedelta(1), + dt.timedelta(0), + ], + [dt.timedelta(0, 0, 1), dt.timedelta(0, 1, 0), dt.timedelta(1, 0, 0)], + [dt.timedelta(1, 0, 0), dt.timedelta(0, 1, 0), dt.timedelta(0, 0, 1)], + [ + dt.timedelta(0), + dt.timedelta(1), + dt.timedelta(1), + dt.timedelta(2), + ], + ], +) +def test___monotonic_timestamps___create_with_irregular_interval___creates_waveform_timing_with_timestamps( + time_offsets: list[dt.timedelta], +) -> None: + start_time = dt.datetime.now(dt.timezone.utc) + timestamps = [start_time + offset for offset in time_offsets] + + timing = Timing.create_with_irregular_interval(timestamps) + + assert_type(timing, Timing) + assert timing.sample_interval_mode == SampleIntervalMode.IRREGULAR + assert timing._timestamps == timestamps + + +@pytest.mark.parametrize( + "time_offsets", + [ + [dt.timedelta(0), dt.timedelta(1), dt.timedelta(0)], + [dt.timedelta(1), dt.timedelta(0), dt.timedelta(1)], + ], +) +def test___non_monotonic_timestamps___create_with_irregular_interval___raises_value_error( + time_offsets: list[dt.timedelta], +) -> None: + start_time = dt.datetime.now(dt.timezone.utc) + timestamps = [start_time + offset for offset in time_offsets] + + with pytest.raises(ValueError) as exc: + _ = Timing.create_with_irregular_interval(timestamps) + + assert exc.value.args[0].startswith("The timestamps must be in ascending or descending order.") + + +def test___timestamps_tuple___create_with_irregular_interval___creates_waveform_timing_with_timestamps() -> ( None ): start_time = dt.datetime.now(dt.timezone.utc) - timestamps = [ - start_time, - start_time + dt.timedelta(seconds=1), - start_time + dt.timedelta(seconds=2.3), - start_time + dt.timedelta(seconds=2.5), - ] + timestamps = (start_time,) timing = Timing.create_with_irregular_interval(timestamps) assert_type(timing, Timing) - assert not timing.has_timestamp - assert timing.time_offset == dt.timedelta() - assert timing._sample_interval is None assert timing.sample_interval_mode == SampleIntervalMode.IRREGULAR - assert timing._timestamps == timestamps + assert timing._timestamps == list(timestamps) ############################################################################### @@ -359,3 +413,76 @@ def test___different_value___equality___not_equal( ) def test___various_values___repr___looks_ok(value: Timing, expected_repr: str) -> None: assert repr(value) == expected_repr + + +############################################################################### +# _append_timing +############################################################################### +@pytest.mark.parametrize( + "left_offsets, right_offsets", + [ + ([], []), + ([dt.timedelta(0)], []), + ([dt.timedelta(0), dt.timedelta(1)], []), + ([dt.timedelta(0), dt.timedelta(1), dt.timedelta(2)], []), + ([dt.timedelta(0)], [dt.timedelta(1)]), + ([dt.timedelta(0)], [dt.timedelta(1), dt.timedelta(2)]), + ( + [dt.timedelta(0), dt.timedelta(1)], + [dt.timedelta(2), dt.timedelta(3)], + ), + ( + [dt.timedelta(3), dt.timedelta(2)], + [dt.timedelta(1), dt.timedelta(0)], + ), + ( + [dt.timedelta(0), dt.timedelta(1)], + [dt.timedelta(1), dt.timedelta(2)], + ), + ( + [dt.timedelta(2), dt.timedelta(1)], + [dt.timedelta(1), dt.timedelta(0)], + ), + ], +) +def test___monotonic_timestamps___append_timing___appends_timestamps( + left_offsets: list[dt.timedelta], + right_offsets: list[dt.timedelta], +) -> None: + start_time = dt.datetime.now(dt.timezone.utc) + left_timestamps = [start_time + offset for offset in left_offsets] + right_timestamps = [start_time + offset for offset in right_offsets] + left_timing = Timing.create_with_irregular_interval(left_timestamps) + right_timing = Timing.create_with_irregular_interval(right_timestamps) + + new_timing = left_timing._append_timing(right_timing) + + assert_type(new_timing, Timing) + assert isinstance(new_timing, Timing) + assert new_timing._timestamps == left_timestamps + right_timestamps + + +@pytest.mark.parametrize( + "left_offsets, right_offsets", + [ + ([dt.timedelta(0)], [dt.timedelta(1), dt.timedelta(0)]), + ([dt.timedelta(1)], [dt.timedelta(0), dt.timedelta(2)]), + ([dt.timedelta(0), dt.timedelta(1)], [dt.timedelta(0)]), + ([dt.timedelta(1), dt.timedelta(0)], [dt.timedelta(1)]), + ([dt.timedelta(0), dt.timedelta(1)], [dt.timedelta(2), dt.timedelta(0)]), + ], +) +def test___non_monotonic_timestamps___append_timing___raises_value_error( + left_offsets: list[dt.timedelta], + right_offsets: list[dt.timedelta], +) -> None: + start_time = dt.datetime.now(dt.timezone.utc) + left_timestamps = [start_time + offset for offset in left_offsets] + right_timestamps = [start_time + offset for offset in right_offsets] + left_timing = Timing.create_with_irregular_interval(left_timestamps) + right_timing = Timing.create_with_irregular_interval(right_timestamps) + + with pytest.raises(ValueError) as exc: + _ = left_timing._append_timing(right_timing) + + assert exc.value.args[0].startswith("The timestamps must be in ascending or descending order.") diff --git a/tests/unit/waveform/test_analog_waveform.py b/tests/unit/waveform/test_analog_waveform.py index 88e807ee..53dff0d5 100644 --- a/tests/unit/waveform/test_analog_waveform.py +++ b/tests/unit/waveform/test_analog_waveform.py @@ -18,8 +18,12 @@ LinearScaleMode, NoneScaleMode, PrecisionTiming, + SampleIntervalMode, ScaleMode, + ScalingMismatchWarning, Timing, + TimingMismatchError, + TimingMismatchWarning, ) @@ -955,6 +959,41 @@ def test___waveform_with_precision_timing___get_timing___converts_timing() -> No ) +def test___waveform_with_cached_timing___get_timing___returns_cached_timing() -> None: + waveform = AnalogWaveform() + waveform.timing = Timing.create_with_regular_interval( + dt.timedelta(milliseconds=1), dt.datetime(2025, 1, 1), dt.timedelta(seconds=1) + ) + precision_timing_before = waveform.precision_timing + + precision_timing = waveform.precision_timing + + assert precision_timing is precision_timing_before + assert precision_timing == PrecisionTiming.create_with_regular_interval( + ht.timedelta(milliseconds=1), ht.datetime(2025, 1, 1), ht.timedelta(seconds=1) + ) + + +def test___waveform_with_cached_timing___set_timing___clears_cached_timing() -> None: + waveform = AnalogWaveform() + waveform.timing = Timing.create_with_regular_interval( + dt.timedelta(milliseconds=1), dt.datetime(2025, 1, 1), dt.timedelta(seconds=1) + ) + precision_timing_before = waveform.precision_timing + + waveform.timing = Timing.create_with_regular_interval( + dt.timedelta(milliseconds=2), dt.datetime(2025, 1, 2), dt.timedelta(seconds=2) + ) + + precision_timing_after = waveform.precision_timing + assert precision_timing_before == PrecisionTiming.create_with_regular_interval( + ht.timedelta(milliseconds=1), ht.datetime(2025, 1, 1), ht.timedelta(seconds=1) + ) + assert precision_timing_after == PrecisionTiming.create_with_regular_interval( + ht.timedelta(milliseconds=2), ht.datetime(2025, 1, 2), ht.timedelta(seconds=2) + ) + + ############################################################################### # scale_mode ############################################################################### @@ -964,3 +1003,418 @@ def test___waveform___scale_mode___defaults_to_no_scaling() -> None: assert_type(waveform.scale_mode, ScaleMode) assert isinstance(waveform.scale_mode, NoneScaleMode) assert waveform.scale_mode is NO_SCALING + + +############################################################################### +# append array +############################################################################### +def test___empty_ndarray___append___no_effect() -> None: + waveform = AnalogWaveform.from_array_1d([0, 1, 2], np.int32) + array = np.array([], np.int32) + + waveform.append(array) + + assert list(waveform.raw_data) == [0, 1, 2] + + +def test___int32_ndarray___append___appends_array() -> None: + waveform = AnalogWaveform.from_array_1d([0, 1, 2], np.int32) + array = np.array([3, 4, 5], np.int32) + + waveform.append(array) + + assert list(waveform.raw_data) == [0, 1, 2, 3, 4, 5] + + +def test___float64_ndarray___append___appends_array() -> None: + waveform = AnalogWaveform.from_array_1d([0, 1, 2], np.float64) + array = np.array([3, 4, 5], np.float64) + + waveform.append(array) + + assert list(waveform.raw_data) == [0, 1, 2, 3, 4, 5] + + +def test___ndarray_with_mismatched_dtype___append___raises_type_error() -> None: + waveform = AnalogWaveform.from_array_1d([0, 1, 2], np.float64) + array = np.array([3, 4, 5], np.int32) + + with pytest.raises(TypeError) as exc: + waveform.append(array) # type: ignore[arg-type] + + assert exc.value.args[0].startswith( + "The data type of the input array must match the waveform data type." + ) + + +def test___ndarray_2d___append___raises_value_error() -> None: + waveform = AnalogWaveform.from_array_1d([0, 1, 2], np.float64) + array = np.array([[3, 4, 5], [6, 7, 8]], np.float64) + + with pytest.raises(ValueError) as exc: + waveform.append(array) + + assert exc.value.args[0].startswith("The input array must be a one-dimensional array.") + + +def test___irregular_waveform_and_int32_ndarray_with_timestamps___append___appends_array() -> None: + start_time = dt.datetime.now(dt.timezone.utc) + waveform_offsets = [dt.timedelta(0), dt.timedelta(1), dt.timedelta(2)] + waveform_timestamps = [start_time + offset for offset in waveform_offsets] + waveform = AnalogWaveform.from_array_1d([0, 1, 2], np.int32) + waveform.timing = Timing.create_with_irregular_interval(waveform_timestamps) + array_offsets = [dt.timedelta(3), dt.timedelta(4), dt.timedelta(5)] + array_timestamps = [start_time + offset for offset in array_offsets] + array = np.array([3, 4, 5], np.int32) + + waveform.append(array, array_timestamps) + + assert list(waveform.raw_data) == [0, 1, 2, 3, 4, 5] + assert waveform.timing.sample_interval_mode == SampleIntervalMode.IRREGULAR + assert waveform.timing._timestamps == waveform_timestamps + array_timestamps + + +def test___irregular_waveform_and_int32_ndarray_without_timestamps___append___raises_timing_mismatch_error_and_does_not_append() -> ( + None +): + start_time = dt.datetime.now(dt.timezone.utc) + waveform_offsets = [dt.timedelta(0), dt.timedelta(1), dt.timedelta(2)] + waveform_timestamps = [start_time + offset for offset in waveform_offsets] + waveform = AnalogWaveform.from_array_1d([0, 1, 2], np.int32) + waveform.timing = Timing.create_with_irregular_interval(waveform_timestamps) + array = np.array([3, 4, 5], np.int32) + + with pytest.raises(TimingMismatchError) as exc: + waveform.append(array) + + assert exc.value.args[0].startswith( + "The timestamps argument is required when appending to a waveform with irregular timing." + ) + assert list(waveform.raw_data) == [0, 1, 2] + assert waveform.timing.sample_interval_mode == SampleIntervalMode.IRREGULAR + assert waveform.timing._timestamps == waveform_timestamps + + +def test___irregular_waveform_and_int32_ndarray_with_wrong_timestamp_count___append___raises_value_error_and_does_not_append() -> ( + None +): + start_time = dt.datetime.now(dt.timezone.utc) + waveform_offsets = [dt.timedelta(0), dt.timedelta(1), dt.timedelta(2)] + waveform_timestamps = [start_time + offset for offset in waveform_offsets] + waveform = AnalogWaveform.from_array_1d([0, 1, 2], np.int32) + waveform.timing = Timing.create_with_irregular_interval(waveform_timestamps) + array_offsets = [dt.timedelta(3), dt.timedelta(4)] + array_timestamps = [start_time + offset for offset in array_offsets] + array = np.array([3, 4, 5], np.int32) + + with pytest.raises(ValueError) as exc: + waveform.append(array, array_timestamps) + + assert exc.value.args[0].startswith( + "The number of irregular timestamps must be equal to the input array length." + ) + assert list(waveform.raw_data) == [0, 1, 2] + assert waveform.timing.sample_interval_mode == SampleIntervalMode.IRREGULAR + assert waveform.timing._timestamps == waveform_timestamps + + +def test___regular_waveform_and_int32_ndarray_with_timestamps___append___raises_runtime_error_and_does_not_append() -> ( + None +): + start_time = dt.datetime.now(dt.timezone.utc) + waveform = AnalogWaveform.from_array_1d([0, 1, 2], np.int32) + waveform.timing = Timing.create_with_regular_interval(dt.timedelta(milliseconds=1)) + array_offsets = [dt.timedelta(3), dt.timedelta(4), dt.timedelta(5)] + array_timestamps = [start_time + offset for offset in array_offsets] + array = np.array([3, 4, 5], np.int32) + + with pytest.raises(ValueError) as exc: + waveform.append(array, array_timestamps) + + assert exc.value.args[0].startswith("The timestamps argument is not supported.") + assert list(waveform.raw_data) == [0, 1, 2] + assert waveform.timing.sample_interval_mode == SampleIntervalMode.REGULAR + assert waveform.timing.sample_interval == dt.timedelta(milliseconds=1) + + +############################################################################### +# append waveform +############################################################################### +def test___empty_waveform___append___no_effect() -> None: + waveform = AnalogWaveform.from_array_1d([0, 1, 2], np.int32) + other = AnalogWaveform(dtype=np.int32) + + waveform.append(other) + + assert list(waveform.raw_data) == [0, 1, 2] + + +def test___int32_waveform___append___appends_waveform() -> None: + waveform = AnalogWaveform.from_array_1d([0, 1, 2], np.int32) + other = AnalogWaveform.from_array_1d([3, 4, 5], np.int32) + + waveform.append(other) + + assert list(waveform.raw_data) == [0, 1, 2, 3, 4, 5] + + +def test___float64_waveform___append___appends_waveform() -> None: + waveform = AnalogWaveform.from_array_1d([0, 1, 2], np.float64) + other = AnalogWaveform.from_array_1d([3, 4, 5], np.float64) + + waveform.append(other) + + assert list(waveform.raw_data) == [0, 1, 2, 3, 4, 5] + + +def test___waveform_with_mismatched_dtype___append___raises_type_error() -> None: + waveform = AnalogWaveform.from_array_1d([0, 1, 2], np.float64) + other = AnalogWaveform.from_array_1d([3, 4, 5], np.int32) + + with pytest.raises(TypeError) as exc: + waveform.append(other) # type: ignore[arg-type] + + assert exc.value.args[0].startswith( + "The data type of the input waveform must match the waveform data type." + ) + + +def test___irregular_waveform_and_irregular_waveform___append___appends_waveform() -> None: + start_time = dt.datetime.now(dt.timezone.utc) + waveform_offsets = [dt.timedelta(0), dt.timedelta(1), dt.timedelta(2)] + waveform_timestamps = [start_time + offset for offset in waveform_offsets] + waveform = AnalogWaveform.from_array_1d([0, 1, 2], np.int32) + waveform.timing = Timing.create_with_irregular_interval(waveform_timestamps) + other_offsets = [dt.timedelta(3), dt.timedelta(4), dt.timedelta(5)] + other_timestamps = [start_time + offset for offset in other_offsets] + other = AnalogWaveform.from_array_1d([3, 4, 5], np.int32) + other.timing = Timing.create_with_irregular_interval(other_timestamps) + + waveform.append(other) + + assert list(waveform.raw_data) == [0, 1, 2, 3, 4, 5] + assert waveform.timing.sample_interval_mode == SampleIntervalMode.IRREGULAR + assert waveform.timing._timestamps == waveform_timestamps + other_timestamps + + +def test___irregular_waveform_and_regular_waveform___append___raises_timing_mismatch_error() -> ( + None +): + start_time = dt.datetime.now(dt.timezone.utc) + waveform_offsets = [dt.timedelta(0), dt.timedelta(1), dt.timedelta(2)] + waveform_timestamps = [start_time + offset for offset in waveform_offsets] + waveform = AnalogWaveform.from_array_1d([0, 1, 2], np.int32) + waveform.timing = Timing.create_with_irregular_interval(waveform_timestamps) + other = AnalogWaveform.from_array_1d([3, 4, 5], np.int32) + + with pytest.raises(TimingMismatchError) as exc: + waveform.append(other) + + assert exc.value.args[0].startswith( + "The timing of one or more waveforms does not match the timing of the current waveform." + ) + assert list(waveform.raw_data) == [0, 1, 2] + assert waveform.timing.sample_interval_mode == SampleIntervalMode.IRREGULAR + assert waveform.timing._timestamps == waveform_timestamps + + +def test___regular_waveform_and_irregular_waveform___append___raises_timing_mismatch_error() -> ( + None +): + start_time = dt.datetime.now(dt.timezone.utc) + waveform = AnalogWaveform.from_array_1d([0, 1, 2], np.int32) + waveform.timing = Timing.create_with_regular_interval(dt.timedelta(milliseconds=1)) + other_offsets = [dt.timedelta(3), dt.timedelta(4), dt.timedelta(5)] + other_timestamps = [start_time + offset for offset in other_offsets] + other = AnalogWaveform.from_array_1d([3, 4, 5], np.int32) + other.timing = Timing.create_with_irregular_interval(other_timestamps) + + with pytest.raises(TimingMismatchError) as exc: + waveform.append(other) + + assert exc.value.args[0].startswith( + "The timing of one or more waveforms does not match the timing of the current waveform." + ) + assert list(waveform.raw_data) == [0, 1, 2] + assert waveform.timing.sample_interval_mode == SampleIntervalMode.REGULAR + assert waveform.timing.sample_interval == dt.timedelta(milliseconds=1) + + +def test___regular_waveform_and_regular_waveform_with_different_sample_interval___append___appends_waveform_with_timing_mismatch_warning() -> ( + None +): + waveform = AnalogWaveform.from_array_1d([0, 1, 2], np.int32) + waveform.timing = Timing.create_with_regular_interval(dt.timedelta(milliseconds=1)) + other = AnalogWaveform.from_array_1d([3, 4, 5], np.int32) + other.timing = Timing.create_with_regular_interval(dt.timedelta(milliseconds=2)) + + with pytest.warns(TimingMismatchWarning): + waveform.append(other) + + assert list(waveform.raw_data) == [0, 1, 2, 3, 4, 5] + assert waveform.timing.sample_interval_mode == SampleIntervalMode.REGULAR + assert waveform.timing.sample_interval == dt.timedelta(milliseconds=1) + + +def test___regular_waveform_and_regular_waveform_with_different_extended_properties___append___merges_extended_properties() -> ( + None +): + waveform = AnalogWaveform.from_array_1d([0, 1, 2], np.int32) + waveform.extended_properties["A"] = 1 + waveform.extended_properties["B"] = 2 + other = AnalogWaveform.from_array_1d([3, 4, 5], np.int32) + other.extended_properties["B"] = 3 + other.extended_properties["C"] = 4 + + waveform.append(other) + + assert list(waveform.raw_data) == [0, 1, 2, 3, 4, 5] + assert waveform.extended_properties == {"A": 1, "B": 2, "C": 4} + + +@pytest.mark.xfail(reason="Needs __eq__ from https://github.com/ni/nitypes-python/pull/11") +def test___regular_waveform_and_regular_waveform_with_different_scale_mode___append___appends_waveform_with_scaling_mismatch_warning() -> ( + None +): + waveform = AnalogWaveform.from_array_1d([0, 1, 2], np.int32) + waveform.scale_mode = LinearScaleMode(1.0, 0.0) + other = AnalogWaveform.from_array_1d([3, 4, 5], np.int32) + other.scale_mode = LinearScaleMode(2.0, 0.0) + + with pytest.warns(ScalingMismatchWarning): + waveform.append(other) + + assert list(waveform.raw_data) == [0, 1, 2, 3, 4, 5] + assert waveform.scale_mode == LinearScaleMode(1.0, 0.0) + + +############################################################################### +# append waveforms +############################################################################### +def test___empty_waveform_list___append___no_effect() -> None: + waveform = AnalogWaveform.from_array_1d([0, 1, 2], np.int32) + other: list[AnalogWaveform[np.int32]] = [] + + waveform.append(other) + + assert list(waveform.raw_data) == [0, 1, 2] + + +def test___int32_waveform_list___append___appends_waveform() -> None: + waveform = AnalogWaveform.from_array_1d([0, 1, 2], np.int32) + other = [ + AnalogWaveform.from_array_1d([3, 4, 5], np.int32), + AnalogWaveform.from_array_1d([6], np.int32), + AnalogWaveform.from_array_1d([7, 8], np.int32), + ] + + waveform.append(other) + + assert list(waveform.raw_data) == [0, 1, 2, 3, 4, 5, 6, 7, 8] + + +def test___float64_waveform_tuple___append___appends_waveform() -> None: + waveform = AnalogWaveform.from_array_1d([0, 1, 2], np.float64) + other = ( + AnalogWaveform.from_array_1d([3, 4, 5], np.float64), + AnalogWaveform.from_array_1d([6, 7, 8], np.float64), + ) + + waveform.append(other) + + assert list(waveform.raw_data) == [0, 1, 2, 3, 4, 5, 6, 7, 8] + + +def test___waveform_list_with_mismatched_dtype___append___raises_type_error_and_does_not_append() -> ( + None +): + waveform = AnalogWaveform.from_array_1d([0, 1, 2], np.float64) + other = [ + AnalogWaveform.from_array_1d([3, 4, 5], np.float64), + AnalogWaveform.from_array_1d([6, 7, 8], np.int32), + ] + + with pytest.raises(TypeError) as exc: + waveform.append(other) # type: ignore[arg-type] + + assert exc.value.args[0].startswith( + "The data type of the input waveform must match the waveform data type." + ) + assert list(waveform.raw_data) == [0, 1, 2] + + +def test___irregular_waveform_and_irregular_waveform_list___append___appends_waveform() -> None: + start_time = dt.datetime.now(dt.timezone.utc) + waveform_offsets = [dt.timedelta(0), dt.timedelta(1), dt.timedelta(2)] + waveform_timestamps = [start_time + offset for offset in waveform_offsets] + waveform = AnalogWaveform.from_array_1d([0, 1, 2], np.int32) + waveform.timing = Timing.create_with_irregular_interval(waveform_timestamps) + other1_offsets = [dt.timedelta(3), dt.timedelta(4), dt.timedelta(5)] + other1_timestamps = [start_time + offset for offset in other1_offsets] + other1 = AnalogWaveform.from_array_1d([3, 4, 5], np.int32) + other1.timing = Timing.create_with_irregular_interval(other1_timestamps) + other2_offsets = [dt.timedelta(6), dt.timedelta(7), dt.timedelta(8)] + other2_timestamps = [start_time + offset for offset in other2_offsets] + other2 = AnalogWaveform.from_array_1d([6, 7, 8], np.int32) + other2.timing = Timing.create_with_irregular_interval(other2_timestamps) + other = [other1, other2] + + waveform.append(other) + + assert list(waveform.raw_data) == [0, 1, 2, 3, 4, 5, 6, 7, 8] + assert waveform.timing.sample_interval_mode == SampleIntervalMode.IRREGULAR + assert ( + waveform.timing._timestamps == waveform_timestamps + other1_timestamps + other2_timestamps + ) + + +def test___irregular_waveform_and_regular_waveform_list___append___raises_timing_mismatch_error_and_does_not_append() -> ( + None +): + start_time = dt.datetime.now(dt.timezone.utc) + waveform_offsets = [dt.timedelta(0), dt.timedelta(1), dt.timedelta(2)] + waveform_timestamps = [start_time + offset for offset in waveform_offsets] + waveform = AnalogWaveform.from_array_1d([0, 1, 2], np.int32) + waveform.timing = Timing.create_with_irregular_interval(waveform_timestamps) + other1_offsets = [dt.timedelta(3), dt.timedelta(4), dt.timedelta(5)] + other1_timestamps = [start_time + offset for offset in other1_offsets] + other1 = AnalogWaveform.from_array_1d([3, 4, 5], np.int32) + other1.timing = Timing.create_with_irregular_interval(other1_timestamps) + other2 = AnalogWaveform.from_array_1d([6, 7, 8], np.int32) + other2.timing = Timing.create_with_regular_interval(dt.timedelta(milliseconds=1)) + other = [other1, other2] + + with pytest.raises(TimingMismatchError) as exc: + waveform.append(other) + + assert exc.value.args[0].startswith( + "The timing of one or more waveforms does not match the timing of the current waveform." + ) + assert list(waveform.raw_data) == [0, 1, 2] + assert waveform.timing.sample_interval_mode == SampleIntervalMode.IRREGULAR + assert waveform.timing._timestamps == waveform_timestamps + + +def test___regular_waveform_and_irregular_waveform_list___append___raises_runtime_error_and_does_not_append() -> ( + None +): + start_time = dt.datetime.now(dt.timezone.utc) + waveform = AnalogWaveform.from_array_1d([0, 1, 2], np.int32) + waveform.timing = Timing.create_with_regular_interval(dt.timedelta(milliseconds=1)) + other1 = AnalogWaveform.from_array_1d([3, 4, 5], np.int32) + other1.timing = Timing.create_with_regular_interval(dt.timedelta(milliseconds=1)) + other2_offsets = [dt.timedelta(3), dt.timedelta(4), dt.timedelta(5)] + other2_timestamps = [start_time + offset for offset in other2_offsets] + other2 = AnalogWaveform.from_array_1d([3, 4, 5], np.int32) + other2.timing = Timing.create_with_irregular_interval(other2_timestamps) + other = [other1, other2] + + with pytest.raises(RuntimeError) as exc: + waveform.append(other) + + assert exc.value.args[0].startswith( + "The timing of one or more waveforms does not match the timing of the current waveform." + ) + assert list(waveform.raw_data) == [0, 1, 2] + assert waveform.timing.sample_interval_mode == SampleIntervalMode.REGULAR + assert waveform.timing.sample_interval == dt.timedelta(milliseconds=1)