diff --git a/docs/conf.py b/docs/conf.py index c6095044..bf0169ea 100644 --- a/docs/conf.py +++ b/docs/conf.py @@ -46,6 +46,7 @@ autoapi_options = list(autoapi.extension._DEFAULT_OPTIONS) autoapi_options.remove("private-members") # note: remove this to include "_" members in docs autoapi_dirs = [root_path / "src" / "nitypes"] +autoapi_python_class_content = "both" autoapi_type = "python" autodoc_typehints = "description" diff --git a/src/nitypes/_arguments.py b/src/nitypes/_arguments.py index ba2f3d1a..44eb5b24 100644 --- a/src/nitypes/_arguments.py +++ b/src/nitypes/_arguments.py @@ -133,6 +133,39 @@ def arg_to_uint( return value +def is_dtype(dtype: npt.DTypeLike, supported_dtypes: tuple[npt.DTypeLike, ...]) -> bool: + """Check a dtype-like object against a tuple of supported dtype-like objects. + + Unlike :any:`numpy.isdtype`, this function supports structured data types. + + >>> is_dtype(np.float64, (np.float64, np.intc, np.long,)) + True + >>> is_dtype("float64", (np.float64, np.intc, np.long,)) + True + >>> is_dtype(np.float64, (np.byte, np.short, np.intc, np.int_, np.long, np.longlong)) + False + >>> a_type = np.dtype([('a', np.int32)]) + >>> b_type = np.dtype([('b', np.int32)]) + >>> is_dtype(a_type, (np.float64, np.intc, a_type,)) + True + >>> is_dtype(b_type, (np.float64, np.intc, a_type,)) + False + >>> is_dtype("i2, i2", (np.float64, np.intc, a_type,)) + False + >>> is_dtype("i4", (np.float64, np.intc, a_type,)) + False + >>> is_dtype("i4", (np.float64, np.intc, a_type, np.dtype("i4"),)) + True + """ + if not isinstance(dtype, (type, np.dtype)): + dtype = np.dtype(dtype) + + if isinstance(dtype, np.dtype) and dtype.fields: + return dtype in supported_dtypes + + return np.isdtype(dtype, supported_dtypes) + + def validate_dtype(dtype: npt.DTypeLike, supported_dtypes: tuple[npt.DTypeLike, ...]) -> None: """Validate a dtype-like object against a tuple of supported dtype-like objects. @@ -145,10 +178,20 @@ def validate_dtype(dtype: npt.DTypeLike, supported_dtypes: tuple[npt.DTypeLike, Data type: float64 Supported data types: int8, int16, int32, int64 + >>> a_type = np.dtype([('a', np.int32)]) + >>> b_type = np.dtype([('b', np.int32)]) + >>> validate_dtype(a_type, (np.float64, np.intc, a_type,)) + >>> validate_dtype(b_type, (np.float64, np.intc, a_type,)) + Traceback (most recent call last): + ... + TypeError: The requested data type is not supported. + + Data type: [('b', '>> AnalogWaveform() +nitypes.waveform.AnalogWaveform(0) +>>> AnalogWaveform(5) +nitypes.waveform.AnalogWaveform(5, raw_data=array([0., 0., 0., 0., 0.])) + +To construct an analog waveform from a NumPy array, use the :any:`AnalogWaveform.from_array_1d` +method. + +>>> import numpy as np +>>> AnalogWaveform.from_array_1d(np.array([1.0, 2.0, 3.0])) +nitypes.waveform.AnalogWaveform(3, raw_data=array([1., 2., 3.])) + +You can also use :any:`AnalogWaveform.from_array_1d` to construct an analog waveform from a +sequence, such as a list. In this case, you must specify the NumPy data type. + +>>> AnalogWaveform.from_array_1d([1.0, 2.0, 3.0], np.float64) +nitypes.waveform.AnalogWaveform(3, raw_data=array([1., 2., 3.])) + +The 2D version, :any:`AnalogWaveform.from_array_2d`, constructs a list of waveforms, one for each +row of data in the array or nested sequence. + +>>> nested_list = [[1.0, 2.0, 3.0], [4.0, 5.0, 6.0]] +>>> AnalogWaveform.from_array_2d(nested_list, np.float64) # doctest: +NORMALIZE_WHITESPACE +[nitypes.waveform.AnalogWaveform(3, raw_data=array([1., 2., 3.])), + nitypes.waveform.AnalogWaveform(3, raw_data=array([4., 5., 6.]))] + +Scaling analog data +------------------- + +By default, analog waveforms contain floating point data in :any:`numpy.float64` format, but they +can also be used to scale raw integer data to floating-point: + +>>> scale_mode = LinearScaleMode(gain=2.0, offset=0.5) +>>> wfm = AnalogWaveform.from_array_1d([1, 2, 3], np.int32, scale_mode=scale_mode) +>>> wfm # doctest: +NORMALIZE_WHITESPACE +nitypes.waveform.AnalogWaveform(3, int32, raw_data=array([1, 2, 3], dtype=int32), + scale_mode=nitypes.waveform.LinearScaleMode(2.0, 0.5)) +>>> wfm.raw_data +array([1, 2, 3], dtype=int32) +>>> wfm.scaled_data +array([2.5, 4.5, 6.5]) + +Complex Waveforms +================= + +A complex waveform represents a single complex-number signal, such as I/Q data, with timing +information and extended properties such as units. + +Constructing complex waveforms +------------------------------ + +To construct a complex waveform, use the :any:`ComplexWaveform` class: + +>>> ComplexWaveform.from_array_1d([1 + 2j, 3 + 4j], np.complex128) +nitypes.waveform.ComplexWaveform(2, complex128, raw_data=array([1.+2.j, 3.+4.j])) + +Scaling complex-number data +--------------------------- + +Complex waveforms support scaling raw integer data to floating-point. Python and NumPy do not +have native support for complex integers, so this uses the :any:`ComplexInt32DType` structured data +type. + +>>> from nitypes.complex import ComplexInt32DType +>>> wfm = ComplexWaveform.from_array_1d([(1, 2), (3, 4)], ComplexInt32DType, scale_mode=scale_mode) +>>> wfm # doctest: +NORMALIZE_WHITESPACE +nitypes.waveform.ComplexWaveform(2, void32, raw_data=array([(1, 2), (3, 4)], + dtype=[('real', '>> wfm.raw_data +array([(1, 2), (3, 4)], dtype=[('real', '>> wfm.scaled_data +array([2.5+4.j, 6.5+8.j]) +""" + +from nitypes.waveform._analog import AnalogWaveform +from nitypes.waveform._complex import ComplexWaveform from nitypes.waveform._exceptions import TimingMismatchError from nitypes.waveform._extended_properties import ( ExtendedPropertyDictionary, ExtendedPropertyValue, ) +from nitypes.waveform._numeric import NumericWaveform from nitypes.waveform._scaling import ( NO_SCALING, LinearScaleMode, @@ -23,11 +111,13 @@ __all__ = [ "AnalogWaveform", "BaseTiming", + "ComplexWaveform", "ExtendedPropertyDictionary", "ExtendedPropertyValue", "LinearScaleMode", "NO_SCALING", "NoneScaleMode", + "NumericWaveform", "PrecisionTiming", "SampleIntervalMode", "ScaleMode", @@ -40,11 +130,13 @@ # Hide that it was defined in a helper file AnalogWaveform.__module__ = __name__ BaseTiming.__module__ = __name__ +ComplexWaveform.__module__ = __name__ ExtendedPropertyDictionary.__module__ = __name__ # ExtendedPropertyValue is a TypeAlias LinearScaleMode.__module__ = __name__ # NO_SCALING is a constant NoneScaleMode.__module__ = __name__ +NumericWaveform.__module__ = __name__ PrecisionTiming.__module__ = __name__ SampleIntervalMode.__module__ = __name__ ScaleMode.__module__ = __name__ diff --git a/src/nitypes/waveform/_analog.py b/src/nitypes/waveform/_analog.py new file mode 100644 index 00000000..4813104f --- /dev/null +++ b/src/nitypes/waveform/_analog.py @@ -0,0 +1,363 @@ +from __future__ import annotations + +from collections.abc import Mapping, Sequence +from typing import Any, SupportsIndex, Union, overload + +import numpy as np +import numpy.typing as npt +from typing_extensions import TypeVar, final, override + +from nitypes.waveform._extended_properties import ExtendedPropertyValue +from nitypes.waveform._numeric import NumericWaveform, _TOtherScaled +from nitypes.waveform._scaling import ScaleMode +from nitypes.waveform._timing import PrecisionTiming, Timing + +# _TRaw and _TRaw_co specify the type of the raw_data array. AnalogWaveform accepts a narrower set +# of types than NumericWaveform. +_TRaw = TypeVar("_TRaw", bound=Union[np.floating, np.integer]) +_TRaw_co = TypeVar("_TRaw_co", bound=Union[np.floating, np.integer], covariant=True) + +# 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). +_RAW_DTYPES = ( + # Floating point + np.single, + np.double, + # Signed integers + np.byte, + np.short, + np.intc, + np.int_, + np.long, + np.longlong, + # Unsigned integers + np.ubyte, + np.ushort, + np.uintc, + np.uint, + np.ulong, + np.ulonglong, +) + +_SCALED_DTYPES = ( + # Floating point + np.single, + np.double, +) + + +@final +class AnalogWaveform(NumericWaveform[_TRaw_co, np.float64]): + """An analog waveform, which encapsulates analog data and timing information.""" + + @override + @staticmethod + def _get_default_raw_dtype() -> type[np.generic] | np.dtype[np.generic]: + return np.float64 + + @override + @staticmethod + def _get_default_scaled_dtype() -> type[np.generic] | np.dtype[np.generic]: + return np.float64 + + @override + @staticmethod + def _get_supported_raw_dtypes() -> tuple[npt.DTypeLike, ...]: + return _RAW_DTYPES + + @override + @staticmethod + def _get_supported_scaled_dtypes() -> tuple[npt.DTypeLike, ...]: + return _SCALED_DTYPES + + # Override from_array_1d, from_array_2d, and __init__ in order to use overloads to control type + # inference. + @overload + @classmethod + def from_array_1d( + cls, + array: npt.NDArray[_TRaw], + dtype: None = ..., + *, + copy: bool = ..., + start_index: SupportsIndex | None = ..., + sample_count: SupportsIndex | None = ..., + extended_properties: Mapping[str, ExtendedPropertyValue] | None = ..., + timing: Timing | PrecisionTiming | None = ..., + scale_mode: ScaleMode | None = ..., + ) -> AnalogWaveform[_TRaw]: ... + + @overload + @classmethod + def from_array_1d( + cls, + array: npt.NDArray[Any] | Sequence[Any], + dtype: type[_TRaw] | np.dtype[_TRaw] = ..., + *, + copy: bool = ..., + start_index: SupportsIndex | None = ..., + sample_count: SupportsIndex | None = ..., + extended_properties: Mapping[str, ExtendedPropertyValue] | None = ..., + timing: Timing | PrecisionTiming | None = ..., + scale_mode: ScaleMode | None = ..., + ) -> AnalogWaveform[_TRaw]: ... + + @overload + @classmethod + def from_array_1d( + cls, + array: npt.NDArray[Any] | Sequence[Any], + dtype: npt.DTypeLike = ..., + *, + copy: bool = ..., + start_index: SupportsIndex | None = ..., + sample_count: SupportsIndex | None = ..., + extended_properties: Mapping[str, ExtendedPropertyValue] | None = ..., + timing: Timing | PrecisionTiming | None = ..., + scale_mode: ScaleMode | None = ..., + ) -> AnalogWaveform[Any]: ... + + @override + @classmethod + def from_array_1d( + cls, + array: npt.NDArray[Any] | Sequence[Any], + dtype: npt.DTypeLike = None, + *, + copy: bool = True, + start_index: SupportsIndex | None = 0, + sample_count: SupportsIndex | None = None, + extended_properties: Mapping[str, ExtendedPropertyValue] | None = None, + timing: Timing | PrecisionTiming | None = None, + scale_mode: ScaleMode | None = None, + ) -> AnalogWaveform[Any]: + """Construct an analog waveform from a one-dimensional array or sequence. + + Args: + array: The waveform data as a one-dimensional array or a sequence. + dtype: The NumPy data type for the waveform data. This argument is required + when array is a sequence. + copy: Specifies whether to copy the array or save a reference to it. + start_index: The sample index at which the waveform data begins. + sample_count: The number of samples in the waveform. + extended_properties: The extended properties of the waveform. + timing: The timing information of the waveform. + scale_mode: The scale mode of the waveform. + + Returns: + An analog waveform containing the specified data. + """ + return super(AnalogWaveform, cls).from_array_1d( + array, + dtype, + copy=copy, + start_index=start_index, + sample_count=sample_count, + extended_properties=extended_properties, + timing=timing, + scale_mode=scale_mode, + ) + + @overload + @classmethod + def from_array_2d( + cls, + array: npt.NDArray[_TRaw], + dtype: None = ..., + *, + copy: bool = ..., + start_index: SupportsIndex | None = ..., + sample_count: SupportsIndex | None = ..., + extended_properties: Mapping[str, ExtendedPropertyValue] | None = ..., + timing: Timing | PrecisionTiming | None = ..., + scale_mode: ScaleMode | None = ..., + ) -> list[AnalogWaveform[_TRaw]]: ... + + @overload + @classmethod + def from_array_2d( + cls, + array: npt.NDArray[Any] | Sequence[Sequence[Any]], + dtype: type[_TRaw] | np.dtype[_TRaw] = ..., + *, + copy: bool = ..., + start_index: SupportsIndex | None = ..., + sample_count: SupportsIndex | None = ..., + extended_properties: Mapping[str, ExtendedPropertyValue] | None = ..., + timing: Timing | PrecisionTiming | None = ..., + scale_mode: ScaleMode | None = ..., + ) -> list[AnalogWaveform[_TRaw]]: ... + + @overload + @classmethod + def from_array_2d( + cls, + array: npt.NDArray[Any] | Sequence[Sequence[Any]], + dtype: npt.DTypeLike = ..., + *, + copy: bool = ..., + start_index: SupportsIndex | None = ..., + sample_count: SupportsIndex | None = ..., + extended_properties: Mapping[str, ExtendedPropertyValue] | None = ..., + timing: Timing | PrecisionTiming | None = ..., + scale_mode: ScaleMode | None = ..., + ) -> list[AnalogWaveform[Any]]: ... + + @override + @classmethod + def from_array_2d( + cls, + array: npt.NDArray[Any] | Sequence[Sequence[Any]], + dtype: npt.DTypeLike = None, + *, + copy: bool = True, + start_index: SupportsIndex | None = 0, + sample_count: SupportsIndex | None = None, + extended_properties: Mapping[str, ExtendedPropertyValue] | None = None, + timing: Timing | PrecisionTiming | None = None, + scale_mode: ScaleMode | None = None, + ) -> list[AnalogWaveform[Any]]: + """Construct a list of analog waveforms from a two-dimensional array or nested sequence. + + Args: + array: The waveform data as a two-dimensional array or a nested sequence. + dtype: The NumPy data type for the waveform data. This argument is required + when array is a sequence. + copy: Specifies whether to copy the array or save a reference to it. + start_index: The sample index at which the waveform data begins. + sample_count: The number of samples in the waveform. + extended_properties: The extended properties of the waveform. + timing: The timing information of the waveform. + scale_mode: The scale mode of the waveform. + + Returns: + A list containing an analog waveform for each row of the specified data. + + When constructing multiple waveforms, the same extended properties, timing + information, and scale mode are applied to all waveforms. Consider assigning + these properties after construction. + """ + return super(AnalogWaveform, cls).from_array_2d( + array, + dtype, + copy=copy, + start_index=start_index, + sample_count=sample_count, + extended_properties=extended_properties, + timing=timing, + scale_mode=scale_mode, + ) + + __slots__ = () + + # If neither dtype nor raw_data is specified, _TRaw_co defaults to np.float64. + @overload + def __init__( # noqa: D107 - Missing docstring in __init__ (auto-generated noqa) + self: AnalogWaveform[np.float64], + sample_count: SupportsIndex | None = ..., + dtype: None = ..., + *, + raw_data: None = ..., + start_index: SupportsIndex | None = ..., + capacity: SupportsIndex | None = ..., + extended_properties: Mapping[str, ExtendedPropertyValue] | None = ..., + timing: Timing | PrecisionTiming | None = ..., + scale_mode: ScaleMode | None = ..., + ) -> None: ... + + @overload + def __init__( # noqa: D107 - Missing docstring in __init__ (auto-generated noqa) + self: AnalogWaveform[_TRaw], + sample_count: SupportsIndex | None = ..., + dtype: type[_TRaw] | np.dtype[_TRaw] = ..., + *, + raw_data: None = ..., + start_index: SupportsIndex | None = ..., + capacity: SupportsIndex | None = ..., + extended_properties: Mapping[str, ExtendedPropertyValue] | None = ..., + timing: Timing | PrecisionTiming | None = ..., + scale_mode: ScaleMode | None = ..., + ) -> None: ... + + @overload + def __init__( # noqa: D107 - Missing docstring in __init__ (auto-generated noqa) + self: AnalogWaveform[_TRaw], + sample_count: SupportsIndex | None = ..., + dtype: None = ..., + *, + raw_data: npt.NDArray[_TRaw] = ..., + start_index: SupportsIndex | None = ..., + capacity: SupportsIndex | None = ..., + extended_properties: Mapping[str, ExtendedPropertyValue] | None = ..., + timing: Timing | PrecisionTiming | None = ..., + scale_mode: ScaleMode | None = ..., + ) -> None: ... + + @overload + def __init__( # noqa: D107 - Missing docstring in __init__ (auto-generated noqa) + self: AnalogWaveform[Any], + sample_count: SupportsIndex | None = ..., + dtype: npt.DTypeLike = ..., + *, + raw_data: npt.NDArray[Any] | None = ..., + start_index: SupportsIndex | None = ..., + capacity: SupportsIndex | None = ..., + extended_properties: Mapping[str, ExtendedPropertyValue] | None = ..., + timing: Timing | PrecisionTiming | None = ..., + scale_mode: ScaleMode | None = ..., + ) -> None: ... + + def __init__( + self, + sample_count: SupportsIndex | None = None, + dtype: npt.DTypeLike = None, + *, + raw_data: npt.NDArray[Any] | None = None, + start_index: SupportsIndex | None = None, + capacity: SupportsIndex | None = None, + extended_properties: Mapping[str, ExtendedPropertyValue] | None = None, + copy_extended_properties: bool = True, + timing: Timing | PrecisionTiming | None = None, + scale_mode: ScaleMode | None = None, + ) -> None: + """Construct an analog waveform. + + Args: + sample_count: The number of samples in the analog waveform. + dtype: The NumPy data type for the analog waveform data. If not specified, the data + type defaults to np.float64. + raw_data: A NumPy ndarray to use for sample storage. The analog waveform takes ownership + of this array. If not specified, an ndarray is created based on the specified dtype, + start index, sample count, and capacity. + start_index: The sample index at which the analog waveform data begins. + sample_count: The number of samples in the analog waveform. + capacity: The number of samples to allocate. Pre-allocating a larger buffer optimizes + appending samples to the waveform. + extended_properties: The extended properties of the analog waveform. + copy_extended_properties: Specifies whether to copy the extended properties or take + ownership. + timing: The timing information of the analog waveform. + scale_mode: The scale mode of the analog waveform. + + Returns: + An analog waveform. + """ + return super().__init__( + sample_count, + dtype, + raw_data=raw_data, + start_index=start_index, + capacity=capacity, + extended_properties=extended_properties, + copy_extended_properties=copy_extended_properties, + timing=timing, + scale_mode=scale_mode, + ) + + @override + def _convert_data( + self, + dtype: npt.DTypeLike | type[_TOtherScaled] | np.dtype[_TOtherScaled], + raw_data: npt.NDArray[_TRaw_co], + ) -> npt.NDArray[_TOtherScaled]: + return raw_data.astype(dtype) diff --git a/src/nitypes/waveform/_complex.py b/src/nitypes/waveform/_complex.py new file mode 100644 index 00000000..ae52d5ce --- /dev/null +++ b/src/nitypes/waveform/_complex.py @@ -0,0 +1,351 @@ +from __future__ import annotations + +from collections.abc import Mapping, Sequence +from typing import Any, SupportsIndex, Union, overload + +import numpy as np +import numpy.typing as npt +from typing_extensions import TypeVar, final, override + +from nitypes.complex import ComplexInt32Base, ComplexInt32DType, convert_complex +from nitypes.waveform._extended_properties import ExtendedPropertyValue +from nitypes.waveform._numeric import NumericWaveform, _TOtherScaled +from nitypes.waveform._scaling import ScaleMode +from nitypes.waveform._timing import PrecisionTiming, Timing + +# _TRaw and _TRaw_co specify the type of the raw_data array. ComplexWaveform accepts a narrower +# set of types than NumericWaveform. Note that ComplexInt32Base is an alias for np.void, but other +# structured data types are rejected at run time. +_TRaw = TypeVar("_TRaw", bound=Union[np.complexfloating, ComplexInt32Base]) +_TRaw_co = TypeVar("_TRaw_co", bound=Union[np.complexfloating, ComplexInt32Base], covariant=True) + +# 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). +_RAW_DTYPES = ( + # Complex floating point + np.csingle, + np.cdouble, + # Complex integers + ComplexInt32DType, +) + +_SCALED_DTYPES = ( + # Complex floating point + np.csingle, + np.cdouble, +) + + +@final +class ComplexWaveform(NumericWaveform[_TRaw_co, np.complex128]): + """A complex waveform, which encapsulates complex data and timing information.""" + + @override + @staticmethod + def _get_default_raw_dtype() -> type[np.generic] | np.dtype[np.generic]: + return np.complex128 + + @override + @staticmethod + def _get_default_scaled_dtype() -> type[np.generic] | np.dtype[np.generic]: + return np.complex128 + + @override + @staticmethod + def _get_supported_raw_dtypes() -> tuple[npt.DTypeLike, ...]: + return _RAW_DTYPES + + @override + @staticmethod + def _get_supported_scaled_dtypes() -> tuple[npt.DTypeLike, ...]: + return _SCALED_DTYPES + + @overload + @classmethod + def from_array_1d( + cls, + array: npt.NDArray[_TRaw], + dtype: None = ..., + *, + copy: bool = ..., + start_index: SupportsIndex | None = ..., + sample_count: SupportsIndex | None = ..., + extended_properties: Mapping[str, ExtendedPropertyValue] | None = ..., + timing: Timing | PrecisionTiming | None = ..., + scale_mode: ScaleMode | None = ..., + ) -> ComplexWaveform[_TRaw]: ... + + @overload + @classmethod + def from_array_1d( + cls, + array: npt.NDArray[Any] | Sequence[Any], + dtype: type[_TRaw] | np.dtype[_TRaw] = ..., + *, + copy: bool = ..., + start_index: SupportsIndex | None = ..., + sample_count: SupportsIndex | None = ..., + extended_properties: Mapping[str, ExtendedPropertyValue] | None = ..., + timing: Timing | PrecisionTiming | None = ..., + scale_mode: ScaleMode | None = ..., + ) -> ComplexWaveform[_TRaw]: ... + + @overload + @classmethod + def from_array_1d( + cls, + array: npt.NDArray[Any] | Sequence[Any], + dtype: npt.DTypeLike = ..., + *, + copy: bool = ..., + start_index: SupportsIndex | None = ..., + sample_count: SupportsIndex | None = ..., + extended_properties: Mapping[str, ExtendedPropertyValue] | None = ..., + timing: Timing | PrecisionTiming | None = ..., + scale_mode: ScaleMode | None = ..., + ) -> ComplexWaveform[Any]: ... + + @override + @classmethod + def from_array_1d( + cls, + array: npt.NDArray[Any] | Sequence[Any], + dtype: npt.DTypeLike = None, + *, + copy: bool = True, + start_index: SupportsIndex | None = 0, + sample_count: SupportsIndex | None = None, + extended_properties: Mapping[str, ExtendedPropertyValue] | None = None, + timing: Timing | PrecisionTiming | None = None, + scale_mode: ScaleMode | None = None, + ) -> ComplexWaveform[Any]: + """Construct a complex waveform from a one-dimensional array or sequence. + + Args: + array: The waveform data as a one-dimensional array or a sequence. + dtype: The NumPy data type for the waveform data. This argument is required + when array is a sequence. + copy: Specifies whether to copy the array or save a reference to it. + start_index: The sample index at which the waveform data begins. + sample_count: The number of samples in the waveform. + extended_properties: The extended properties of the waveform. + timing: The timing information of the waveform. + scale_mode: The scale mode of the waveform. + + Returns: + A complex waveform containing the specified data. + """ + return super(ComplexWaveform, cls).from_array_1d( + array, + dtype, + copy=copy, + start_index=start_index, + sample_count=sample_count, + extended_properties=extended_properties, + timing=timing, + scale_mode=scale_mode, + ) + + @overload + @classmethod + def from_array_2d( + cls, + array: npt.NDArray[_TRaw], + dtype: None = ..., + *, + copy: bool = ..., + start_index: SupportsIndex | None = ..., + sample_count: SupportsIndex | None = ..., + extended_properties: Mapping[str, ExtendedPropertyValue] | None = ..., + timing: Timing | PrecisionTiming | None = ..., + scale_mode: ScaleMode | None = ..., + ) -> list[ComplexWaveform[_TRaw]]: ... + + @overload + @classmethod + def from_array_2d( + cls, + array: npt.NDArray[Any] | Sequence[Sequence[Any]], + dtype: type[_TRaw] | np.dtype[_TRaw] = ..., + *, + copy: bool = ..., + start_index: SupportsIndex | None = ..., + sample_count: SupportsIndex | None = ..., + extended_properties: Mapping[str, ExtendedPropertyValue] | None = ..., + timing: Timing | PrecisionTiming | None = ..., + scale_mode: ScaleMode | None = ..., + ) -> list[ComplexWaveform[_TRaw]]: ... + + @overload + @classmethod + def from_array_2d( + cls, + array: npt.NDArray[Any] | Sequence[Sequence[Any]], + dtype: npt.DTypeLike = ..., + *, + copy: bool = ..., + start_index: SupportsIndex | None = ..., + sample_count: SupportsIndex | None = ..., + extended_properties: Mapping[str, ExtendedPropertyValue] | None = ..., + timing: Timing | PrecisionTiming | None = ..., + scale_mode: ScaleMode | None = ..., + ) -> list[ComplexWaveform[Any]]: ... + + @override + @classmethod + def from_array_2d( + cls, + array: npt.NDArray[Any] | Sequence[Sequence[Any]], + dtype: npt.DTypeLike = None, + *, + copy: bool = True, + start_index: SupportsIndex | None = 0, + sample_count: SupportsIndex | None = None, + extended_properties: Mapping[str, ExtendedPropertyValue] | None = None, + timing: Timing | PrecisionTiming | None = None, + scale_mode: ScaleMode | None = None, + ) -> list[ComplexWaveform[Any]]: + """Construct a list of complex waveforms from a two-dimensional array or nested sequence. + + Args: + array: The waveform data as a two-dimensional array or a nested sequence. + dtype: The NumPy data type for the waveform data. This argument is required + when array is a sequence. + copy: Specifies whether to copy the array or save a reference to it. + start_index: The sample index at which the waveform data begins. + sample_count: The number of samples in the waveform. + extended_properties: The extended properties of the waveform. + timing: The timing information of the waveform. + scale_mode: The scale mode of the waveform. + + Returns: + A list containing a complex waveform for each row of the specified data. + + When constructing multiple waveforms, the same extended properties, timing + information, and scale mode are applied to all waveforms. Consider assigning + these properties after construction. + """ + return super(ComplexWaveform, cls).from_array_2d( + array, + dtype, + copy=copy, + start_index=start_index, + sample_count=sample_count, + extended_properties=extended_properties, + timing=timing, + scale_mode=scale_mode, + ) + + __slots__ = () + + # If neither dtype nor raw_data is specified, _TRaw_co defaults to np.complex128. + @overload + def __init__( # noqa: D107 - Missing docstring in __init__ (auto-generated noqa) + self: ComplexWaveform[np.complex128], + sample_count: SupportsIndex | None = ..., + dtype: None = ..., + *, + raw_data: None = ..., + start_index: SupportsIndex | None = ..., + capacity: SupportsIndex | None = ..., + extended_properties: Mapping[str, ExtendedPropertyValue] | None = ..., + timing: Timing | PrecisionTiming | None = ..., + scale_mode: ScaleMode | None = ..., + ) -> None: ... + + @overload + def __init__( # noqa: D107 - Missing docstring in __init__ (auto-generated noqa) + self: ComplexWaveform[_TRaw], + sample_count: SupportsIndex | None = ..., + dtype: type[_TRaw] | np.dtype[_TRaw] = ..., + *, + raw_data: None = ..., + start_index: SupportsIndex | None = ..., + capacity: SupportsIndex | None = ..., + extended_properties: Mapping[str, ExtendedPropertyValue] | None = ..., + timing: Timing | PrecisionTiming | None = ..., + scale_mode: ScaleMode | None = ..., + ) -> None: ... + + @overload + def __init__( # noqa: D107 - Missing docstring in __init__ (auto-generated noqa) + self: ComplexWaveform[_TRaw], + sample_count: SupportsIndex | None = ..., + dtype: None = ..., + *, + raw_data: npt.NDArray[_TRaw] = ..., + start_index: SupportsIndex | None = ..., + capacity: SupportsIndex | None = ..., + extended_properties: Mapping[str, ExtendedPropertyValue] | None = ..., + timing: Timing | PrecisionTiming | None = ..., + scale_mode: ScaleMode | None = ..., + ) -> None: ... + + @overload + def __init__( # noqa: D107 - Missing docstring in __init__ (auto-generated noqa) + self: ComplexWaveform[Any], + sample_count: SupportsIndex | None = ..., + dtype: npt.DTypeLike = ..., + *, + raw_data: npt.NDArray[Any] | None = ..., + start_index: SupportsIndex | None = ..., + capacity: SupportsIndex | None = ..., + extended_properties: Mapping[str, ExtendedPropertyValue] | None = ..., + timing: Timing | PrecisionTiming | None = ..., + scale_mode: ScaleMode | None = ..., + ) -> None: ... + + def __init__( + self, + sample_count: SupportsIndex | None = None, + dtype: npt.DTypeLike = None, + *, + raw_data: npt.NDArray[Any] | None = None, + start_index: SupportsIndex | None = None, + capacity: SupportsIndex | None = None, + extended_properties: Mapping[str, ExtendedPropertyValue] | None = None, + copy_extended_properties: bool = True, + timing: Timing | PrecisionTiming | None = None, + scale_mode: ScaleMode | None = None, + ) -> None: + """Construct a complex waveform. + + Args: + sample_count: The number of samples in the waveform. + dtype: The NumPy data type for the waveform data. If not specified, the data + type defaults to np.complex128. + raw_data: A NumPy ndarray to use for sample storage. The waveform takes ownership + of this array. If not specified, an ndarray is created based on the specified dtype, + start index, sample count, and capacity. + start_index: The sample index at which the waveform data begins. + sample_count: The number of samples in the waveform. + capacity: The number of samples to allocate. Pre-allocating a larger buffer optimizes + appending samples to the waveform. + extended_properties: The extended properties of the waveform. + copy_extended_properties: Specifies whether to copy the extended properties or take + ownership. + timing: The timing information of the waveform. + scale_mode: The scale mode of the waveform. + + Returns: + A complex waveform. + """ + return super().__init__( + sample_count, + dtype, + raw_data=raw_data, + start_index=start_index, + capacity=capacity, + extended_properties=extended_properties, + copy_extended_properties=copy_extended_properties, + timing=timing, + scale_mode=scale_mode, + ) + + @override + def _convert_data( + self, + dtype: npt.DTypeLike | type[_TOtherScaled] | np.dtype[_TOtherScaled], + raw_data: npt.NDArray[_TRaw_co], + ) -> npt.NDArray[_TOtherScaled]: + return convert_complex(dtype, raw_data) diff --git a/src/nitypes/waveform/_analog_waveform.py b/src/nitypes/waveform/_numeric.py similarity index 68% rename from src/nitypes/waveform/_analog_waveform.py rename to src/nitypes/waveform/_numeric.py index aa0dc5f1..cef7a01c 100644 --- a/src/nitypes/waveform/_analog_waveform.py +++ b/src/nitypes/waveform/_numeric.py @@ -3,13 +3,14 @@ import datetime as dt import sys import warnings +from abc import ABC, abstractmethod from collections.abc import Mapping, Sequence -from typing import Any, Generic, SupportsIndex, TypeVar, Union, cast, overload +from typing import Any, Generic, SupportsIndex, Union, cast, overload import hightime as ht import numpy as np import numpy.typing as npt -from typing_extensions import Self, TypeAlias +from typing_extensions import Self, TypeAlias, TypeVar from nitypes._arguments import arg_to_uint, validate_dtype, validate_unsupported_arg from nitypes._exceptions import invalid_arg_type, invalid_array_ndim @@ -30,99 +31,63 @@ if sys.version_info < (3, 10): import array as std_array +# _TRaw_co specifies the type of the raw_data array. It is not limited to supported +# types. Requesting an unsupported type raises TypeError at run time. +_TRaw_co = TypeVar("_TRaw_co", bound=np.generic, covariant=True) -_ScalarType = TypeVar("_ScalarType", bound=np.generic) -_ScalarType_co = TypeVar("_ScalarType_co", bound=np.generic, covariant=True) +# _TScaled_co specifies the type of the scaled_data property. +_TScaled_co = TypeVar("_TScaled_co", bound=np.generic, covariant=True) + +# _TOtherScaled is for the get_scaled_data() method, which supports both single and +# double precision. +_TOtherScaled = TypeVar("_TOtherScaled", bound=np.generic) _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 = ( - # Floating point - np.single, - np.double, - # Signed integers - np.byte, - np.short, - np.intc, - np.int_, - np.long, - np.longlong, - # Unsigned integers - np.ubyte, - np.ushort, - np.uintc, - np.uint, - np.ulong, - np.ulonglong, -) - -_SCALED_DTYPES = ( - # Floating point - np.single, - 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. # - npt.ArrayLike accepts some types that np.asarray() does not, such as buffers, so we are # explicitly using npt.NDArray | Sequence instead of npt.ArrayLike. -# - _ScalarType is bound to np.generic, so Sequence[_ScalarType] will not match list[int]. -# - We are not using PEP 696 – Type Defaults for Type Parameters because it makes the type parameter -# default to np.float64 in some cases where it should be inferred as Any, such as when dtype is -# specified as a str. +# - _TRaw_co is bound to np.generic, so Sequence[_TRaw_co] will not match list[int]. +# - We are not using PEP 696 – Type Defaults for Type Parameters for type variables on functions +# because it makes the type parameter default to np.float64 in some cases where it should be +# inferred as Any, such as when dtype is specified as a str. PEP 696 seems more appropriate for +# type variables on classes. -class AnalogWaveform(Generic[_ScalarType_co]): - """An analog waveform, which encapsulates analog data and timing information.""" +class NumericWaveform(ABC, Generic[_TRaw_co, _TScaled_co]): + """A numeric waveform, which encapsulates numeric data and timing information. + + This is an abstract base class. To create a numeric waveform, use :any:`AnalogWaveform` or + :any:`ComplexWaveform`. + """ - @overload @staticmethod - def from_array_1d( - array: npt.NDArray[_ScalarType], - dtype: None = ..., - *, - copy: bool = ..., - start_index: SupportsIndex | None = ..., - sample_count: SupportsIndex | None = ..., - extended_properties: Mapping[str, ExtendedPropertyValue] | None = ..., - timing: Timing | PrecisionTiming | None = ..., - scale_mode: ScaleMode | None = ..., - ) -> AnalogWaveform[_ScalarType]: ... + @abstractmethod + def _get_default_raw_dtype() -> type[np.generic] | np.dtype[np.generic]: + raise NotImplementedError - @overload @staticmethod - def from_array_1d( - array: npt.NDArray[Any] | Sequence[Any], - dtype: type[_ScalarType] | np.dtype[_ScalarType] = ..., - *, - copy: bool = ..., - start_index: SupportsIndex | None = ..., - sample_count: SupportsIndex | None = ..., - extended_properties: Mapping[str, ExtendedPropertyValue] | None = ..., - timing: Timing | PrecisionTiming | None = ..., - scale_mode: ScaleMode | None = ..., - ) -> AnalogWaveform[_ScalarType]: ... + @abstractmethod + def _get_default_scaled_dtype() -> type[np.generic] | np.dtype[np.generic]: + raise NotImplementedError - @overload @staticmethod - def from_array_1d( - array: npt.NDArray[Any] | Sequence[Any], - dtype: npt.DTypeLike = ..., - *, - copy: bool = ..., - start_index: SupportsIndex | None = ..., - sample_count: SupportsIndex | None = ..., - extended_properties: Mapping[str, ExtendedPropertyValue] | None = ..., - timing: Timing | PrecisionTiming | None = ..., - scale_mode: ScaleMode | None = ..., - ) -> AnalogWaveform[Any]: ... + @abstractmethod + def _get_supported_raw_dtypes() -> tuple[npt.DTypeLike, ...]: + raise NotImplementedError @staticmethod + @abstractmethod + def _get_supported_scaled_dtypes() -> tuple[npt.DTypeLike, ...]: + raise NotImplementedError + + @classmethod def from_array_1d( + cls, array: npt.NDArray[Any] | Sequence[Any], dtype: npt.DTypeLike = None, *, @@ -132,22 +97,22 @@ def from_array_1d( extended_properties: Mapping[str, ExtendedPropertyValue] | None = None, timing: Timing | PrecisionTiming | None = None, scale_mode: ScaleMode | None = None, - ) -> AnalogWaveform[_ScalarType]: - """Construct an analog waveform from a one-dimensional array or sequence. + ) -> Self: + """Construct a waveform from a one-dimensional array or sequence. Args: - array: The analog waveform data as a one-dimensional array or a sequence. - dtype: The NumPy data type for the analog waveform data. This argument is required + array: The waveform data as a one-dimensional array or a sequence. + dtype: The NumPy data type for the waveform data. This argument is required when array is a sequence. copy: Specifies whether to copy the array or save a reference to it. - start_index: The sample index at which the analog waveform data begins. - sample_count: The number of samples in the analog waveform. - extended_properties: The extended properties of the analog waveform. - timing: The timing information of the analog waveform. - scale_mode: The scale mode of the analog waveform. + start_index: The sample index at which the waveform data begins. + sample_count: The number of samples in the waveform. + extended_properties: The extended properties of the waveform. + timing: The timing information of the waveform. + scale_mode: The scale mode of the waveform. Returns: - An analog waveform containing the specified data. + A waveform containing the specified data. """ if isinstance(array, np.ndarray): if array.ndim != 1: @@ -162,7 +127,7 @@ def from_array_1d( else: raise invalid_arg_type("input array", "one-dimensional array or sequence", array) - return AnalogWaveform( + return cls( raw_data=np.asarray(array, dtype, copy=copy), start_index=start_index, sample_count=sample_count, @@ -171,50 +136,9 @@ def from_array_1d( scale_mode=scale_mode, ) - @overload - @staticmethod - def from_array_2d( - array: npt.NDArray[_ScalarType], - dtype: None = ..., - *, - copy: bool = ..., - start_index: SupportsIndex | None = ..., - sample_count: SupportsIndex | None = ..., - extended_properties: Mapping[str, ExtendedPropertyValue] | None = ..., - timing: Timing | PrecisionTiming | None = ..., - scale_mode: ScaleMode | None = ..., - ) -> list[AnalogWaveform[_ScalarType]]: ... - - @overload - @staticmethod - def from_array_2d( - array: npt.NDArray[Any] | Sequence[Sequence[Any]], - dtype: type[_ScalarType] | np.dtype[_ScalarType] = ..., - *, - copy: bool = ..., - start_index: SupportsIndex | None = ..., - sample_count: SupportsIndex | None = ..., - extended_properties: Mapping[str, ExtendedPropertyValue] | None = ..., - timing: Timing | PrecisionTiming | None = ..., - scale_mode: ScaleMode | None = ..., - ) -> list[AnalogWaveform[_ScalarType]]: ... - - @overload - @staticmethod - def from_array_2d( - array: npt.NDArray[Any] | Sequence[Sequence[Any]], - dtype: npt.DTypeLike = ..., - *, - copy: bool = ..., - start_index: SupportsIndex | None = ..., - sample_count: SupportsIndex | None = ..., - extended_properties: Mapping[str, ExtendedPropertyValue] | None = ..., - timing: Timing | PrecisionTiming | None = ..., - scale_mode: ScaleMode | None = ..., - ) -> list[AnalogWaveform[Any]]: ... - - @staticmethod + @classmethod def from_array_2d( + cls, array: npt.NDArray[Any] | Sequence[Sequence[Any]], dtype: npt.DTypeLike = None, *, @@ -224,25 +148,25 @@ def from_array_2d( extended_properties: Mapping[str, ExtendedPropertyValue] | None = None, timing: Timing | PrecisionTiming | None = None, scale_mode: ScaleMode | None = None, - ) -> list[AnalogWaveform[_ScalarType]]: - """Construct a list of analog waveforms from a two-dimensional array or nested sequence. + ) -> list[Self]: + """Construct a list of waveforms from a two-dimensional array or nested sequence. Args: - array: The analog waveform data as a two-dimensional array or a nested sequence. - dtype: The NumPy data type for the analog waveform data. This argument is required + array: The waveform data as a two-dimensional array or a nested sequence. + dtype: The NumPy data type for the waveform data. This argument is required when array is a sequence. copy: Specifies whether to copy the array or save a reference to it. - start_index: The sample index at which the analog waveform data begins. - sample_count: The number of samples in the analog waveform. - extended_properties: The extended properties of the analog waveform. - timing: The timing information of the analog waveform. - scale_mode: The scale mode of the analog waveform. + start_index: The sample index at which the waveform data begins. + sample_count: The number of samples in the waveform. + extended_properties: The extended properties of the waveform. + timing: The timing information of the waveform. + scale_mode: The scale mode of the waveform. Returns: - A list containing an analog waveform for each row of the specified data. + A list containing a waveform for each row of the specified data. - When constructing multiple analog waveforms, the same extended properties, timing - information, and scale mode are applied to all analog waveforms. Consider assigning + When constructing multiple waveforms, the same extended properties, timing + information, and scale mode are applied to all waveforms. Consider assigning these properties after construction. """ if isinstance(array, np.ndarray): @@ -259,7 +183,7 @@ def from_array_2d( raise invalid_arg_type("input array", "two-dimensional array or nested sequence", array) return [ - AnalogWaveform( + cls( raw_data=np.asarray(array[i], dtype, copy=copy), start_index=start_index, sample_count=sample_count, @@ -281,7 +205,7 @@ def from_array_2d( "__weakref__", ] - _data: npt.NDArray[_ScalarType_co] + _data: npt.NDArray[_TRaw_co] _start_index: int _sample_count: int _extended_properties: ExtendedPropertyDictionary @@ -289,69 +213,12 @@ def from_array_2d( _converted_timing_cache: dict[type[_AnyTiming], _AnyTiming] _scale_mode: ScaleMode - # If neither dtype nor _data is specified, the type parameter defaults to np.float64. - @overload - def __init__( # noqa: D107 - Missing docstring in __init__ (auto-generated noqa) - self: AnalogWaveform[np.float64], - sample_count: SupportsIndex | None = ..., - dtype: None = ..., - *, - raw_data: None = ..., - start_index: SupportsIndex | None = ..., - capacity: SupportsIndex | None = ..., - extended_properties: Mapping[str, ExtendedPropertyValue] | None = ..., - timing: Timing | PrecisionTiming | None = ..., - scale_mode: ScaleMode | None = ..., - ) -> None: ... - - @overload - def __init__( # noqa: D107 - Missing docstring in __init__ (auto-generated noqa) - self: AnalogWaveform[_ScalarType_co], - sample_count: SupportsIndex | None = ..., - dtype: type[_ScalarType_co] | np.dtype[_ScalarType_co] = ..., - *, - raw_data: None = ..., - start_index: SupportsIndex | None = ..., - capacity: SupportsIndex | None = ..., - extended_properties: Mapping[str, ExtendedPropertyValue] | None = ..., - timing: Timing | PrecisionTiming | None = ..., - scale_mode: ScaleMode | None = ..., - ) -> None: ... - - @overload - def __init__( # noqa: D107 - Missing docstring in __init__ (auto-generated noqa) - self: AnalogWaveform[_ScalarType_co], - sample_count: SupportsIndex | None = ..., - dtype: None = ..., - *, - raw_data: npt.NDArray[_ScalarType_co] | None = ..., - start_index: SupportsIndex | None = ..., - capacity: SupportsIndex | None = ..., - extended_properties: Mapping[str, ExtendedPropertyValue] | None = ..., - timing: Timing | PrecisionTiming | None = ..., - scale_mode: ScaleMode | None = ..., - ) -> None: ... - - @overload - def __init__( # noqa: D107 - Missing docstring in __init__ (auto-generated noqa) - self: AnalogWaveform[Any], - sample_count: SupportsIndex | None = ..., - dtype: npt.DTypeLike = ..., - *, - raw_data: npt.NDArray[Any] | None = ..., - start_index: SupportsIndex | None = ..., - capacity: SupportsIndex | None = ..., - extended_properties: Mapping[str, ExtendedPropertyValue] | None = ..., - timing: Timing | PrecisionTiming | None = ..., - scale_mode: ScaleMode | None = ..., - ) -> None: ... - def __init__( self, sample_count: SupportsIndex | None = None, dtype: npt.DTypeLike = None, *, - raw_data: npt.NDArray[_ScalarType_co] | None = None, + raw_data: npt.NDArray[_TRaw_co] | None = None, start_index: SupportsIndex | None = None, capacity: SupportsIndex | None = None, extended_properties: Mapping[str, ExtendedPropertyValue] | None = None, @@ -359,27 +226,26 @@ def __init__( timing: Timing | PrecisionTiming | None = None, scale_mode: ScaleMode | None = None, ) -> None: - """Construct an analog waveform. + """Construct a numeric waveform. Args: - sample_count: The number of samples in the analog waveform. - dtype: The NumPy data type for the analog waveform data. If not specified, the data - type defaults to np.float64. - raw_data: A NumPy ndarray to use for sample storage. The analog waveform takes ownership + sample_count: The number of samples in the waveform. + dtype: The NumPy data type for the waveform data. + raw_data: A NumPy ndarray to use for sample storage. The waveform takes ownership of this array. If not specified, an ndarray is created based on the specified dtype, start index, sample count, and capacity. - start_index: The sample index at which the analog waveform data begins. - sample_count: The number of samples in the analog waveform. + start_index: The sample index at which the waveform data begins. + sample_count: The number of samples in the waveform. capacity: The number of samples to allocate. Pre-allocating a larger buffer optimizes appending samples to the waveform. - extended_properties: The extended properties of the analog waveform. + extended_properties: The extended properties of the waveform. copy_extended_properties: Specifies whether to copy the extended properties or take ownership. - timing: The timing information of the analog waveform. - scale_mode: The scale mode of the analog waveform. + timing: The timing information of the waveform. + scale_mode: The scale mode of the waveform. Returns: - An analog waveform. + A numeric waveform. """ if raw_data is None: self._init_with_new_array( @@ -424,8 +290,8 @@ def _init_with_new_array( capacity = arg_to_uint("capacity", capacity, sample_count) if dtype is None: - dtype = np.float64 - validate_dtype(dtype, _ANALOG_DTYPES) + dtype = self.__class__._get_default_raw_dtype() + validate_dtype(dtype, self.__class__._get_supported_raw_dtypes()) if start_index > capacity: raise ValueError( @@ -447,7 +313,7 @@ def _init_with_new_array( def _init_with_provided_array( self, - data: npt.NDArray[_ScalarType_co], + data: npt.NDArray[_TRaw_co], dtype: npt.DTypeLike = None, *, start_index: SupportsIndex | None = None, @@ -467,7 +333,7 @@ def _init_with_provided_array( f"Array data type: {data.dtype}\n" f"Requested data type: {np.dtype(dtype)}" ) - validate_dtype(dtype, _ANALOG_DTYPES) + validate_dtype(dtype, self.__class__._get_supported_raw_dtypes()) capacity = arg_to_uint("capacity", capacity, len(data)) if capacity != len(data): @@ -499,21 +365,21 @@ def _init_with_provided_array( self._sample_count = sample_count @property - def raw_data(self) -> npt.NDArray[_ScalarType_co]: - """The raw analog waveform data.""" + def raw_data(self) -> npt.NDArray[_TRaw_co]: + """The raw waveform data.""" return self._data[self._start_index : self._start_index + self._sample_count] def get_raw_data( self, start_index: SupportsIndex | None = 0, sample_count: SupportsIndex | None = None - ) -> npt.NDArray[_ScalarType_co]: - """Get a subset of the raw analog waveform data. + ) -> npt.NDArray[_TRaw_co]: + """Get a subset of the raw waveform data. Args: start_index: The sample index at which the data begins. sample_count: The number of samples to return. Returns: - A subset of the raw analog waveform data. + A subset of the raw waveform data. """ start_index = arg_to_uint("sample index", start_index, 0) if start_index > self.sample_count: @@ -535,15 +401,17 @@ def get_raw_data( return self.raw_data[start_index : start_index + sample_count] @property - def scaled_data(self) -> npt.NDArray[np.float64]: - """The scaled analog waveform data. + def scaled_data(self) -> npt.NDArray[_TScaled_co]: + """The scaled waveform data. - This property converts all of the waveform samples to float64 and scales them. To scale a - subset of the waveform or convert to float32, use the get_scaled_data() method instead. + This property converts all of the waveform samples from the raw data type to the scaled + data type and scales them using :any:`scale_mode`. To scale a subset of the waveform or + scale to single-precision floating point, use the :any:`get_scaled_data` method + instead. """ return self.get_scaled_data() - # If dtype is not specified, _ScaledDataType defaults to np.float64. + # If dtype is not specified, the dtype defaults to _TScaled_co. @overload def get_scaled_data( # noqa: D107 - Missing docstring in __init__ (auto-generated noqa) self, @@ -551,16 +419,16 @@ def get_scaled_data( # noqa: D107 - Missing docstring in __init__ (auto-generat *, start_index: SupportsIndex | None = ..., sample_count: SupportsIndex | None = ..., - ) -> npt.NDArray[np.float64]: ... + ) -> npt.NDArray[_TScaled_co]: ... @overload def get_scaled_data( # noqa: D107 - Missing docstring in __init__ (auto-generated noqa) self, - dtype: type[_ScalarType] | np.dtype[_ScalarType] = ..., + dtype: type[_TOtherScaled] | np.dtype[_TOtherScaled] = ..., *, start_index: SupportsIndex | None = ..., sample_count: SupportsIndex | None = ..., - ) -> npt.NDArray[_ScalarType]: ... + ) -> npt.NDArray[_TOtherScaled]: ... @overload def get_scaled_data( # noqa: D107 - Missing docstring in __init__ (auto-generated noqa) @@ -578,7 +446,7 @@ def get_scaled_data( start_index: SupportsIndex | None = 0, sample_count: SupportsIndex | None = None, ) -> npt.NDArray[Any]: - """Get a subset of the scaled analog waveform data with the specified dtype. + """Get a subset of the scaled waveform data with the specified dtype. Args: dtype: The NumPy data type to use for scaled data. @@ -586,24 +454,32 @@ def get_scaled_data( sample_count: The number of samples to scale. Returns: - A subset of the scaled analog waveform data. + A subset of the scaled waveform data. """ if dtype is None: - dtype = np.float64 - validate_dtype(dtype, _SCALED_DTYPES) + dtype = self.__class__._get_default_scaled_dtype() + validate_dtype(dtype, self.__class__._get_supported_scaled_dtypes()) raw_data = self.get_raw_data(start_index, sample_count) - converted_data = raw_data.astype(dtype) + converted_data: npt.NDArray[Any] = self._convert_data(dtype, raw_data) return self._scale_mode._transform_data(converted_data) + @abstractmethod + def _convert_data( + self, + dtype: npt.DTypeLike | type[_TOtherScaled] | np.dtype[_TOtherScaled], + raw_data: npt.NDArray[_TRaw_co], + ) -> npt.NDArray[_TOtherScaled]: + raise NotImplementedError + @property def sample_count(self) -> int: - """The number of samples in the analog waveform.""" + """The number of samples in the waveform.""" return self._sample_count @property def capacity(self) -> int: - """The total capacity available for analog waveform data. + """The total capacity available for waveform data. Setting the capacity resizes the underlying NumPy array in-place. @@ -626,18 +502,18 @@ def capacity(self, value: int) -> None: self._data.resize(value, refcheck=False) @property - def dtype(self) -> np.dtype[_ScalarType_co]: - """The NumPy dtype for the analog waveform data.""" + def dtype(self) -> np.dtype[_TRaw_co]: + """The NumPy dtype for the waveform data.""" return self._data.dtype @property def extended_properties(self) -> ExtendedPropertyDictionary: - """The extended properties for the analog waveform.""" + """The extended properties for the waveform.""" return self._extended_properties @property def channel_name(self) -> str: - """The name of the device channel from which the analog waveform was acquired.""" + """The name of the device channel from which the waveform was acquired.""" value = self._extended_properties.get(CHANNEL_NAME, "") assert isinstance(value, str) return value @@ -650,7 +526,7 @@ def channel_name(self, value: str) -> None: @property def unit_description(self) -> str: - """The unit of measurement, such as volts, of the analog waveform.""" + """The unit of measurement, such as volts, of the waveform.""" value = self._extended_properties.get(UNIT_DESCRIPTION, "") assert isinstance(value, str) return value @@ -685,7 +561,7 @@ def _validate_timing(self, value: _TTiming) -> None: @property def timing(self) -> Timing: - """The timing information of the analog waveform. + """The timing information of the waveform. The default value is Timing.empty. """ @@ -705,19 +581,20 @@ def is_precision_timing_initialized(self) -> bool: @property def precision_timing(self) -> PrecisionTiming: - """The precision timing information of the analog waveform. + """The precision timing information of the waveform. The default value is PrecisionTiming.empty. - Use AnalogWaveform.precision_timing instead of AnalogWaveform.timing to obtain timing - information with higher precision than AnalogWaveform.timing. If the timing information is - set using AnalogWaveform.precision_timing, then this property returns timing information + Use :any:`precision_timing` instead of :any:`timing` to obtain timing + information with higher precision than :any:`timing`. If the timing information is + set using :any:`precision_timing`, then this property returns timing information with up to yoctosecond precision. If the timing information is set using - AnalogWaveform.timing, then the timing information returned has up to microsecond precision. + :any:`timing`, then the timing information returned has up to microsecond + precision. Accessing this property can potentially decrease performance if the timing information is - set using AnalogWaveform.timing. Use AnalogWaveform.is_precision_timing_initialized to - determine if AnalogWaveform.precision_timing has been initialized. + set using :any:`timing`. Use :any:`is_precision_timing_initialized` to + determine if :any:`precision_timing` has been initialized. """ return self._get_timing(PrecisionTiming) @@ -730,7 +607,7 @@ def precision_timing(self, value: PrecisionTiming) -> None: @property def scale_mode(self) -> ScaleMode: - """The scale mode of the analog waveform.""" + """The scale mode of the waveform.""" return self._scale_mode @scale_mode.setter @@ -742,14 +619,14 @@ def scale_mode(self, value: ScaleMode) -> None: def append( self, other: ( - npt.NDArray[_ScalarType_co] - | AnalogWaveform[_ScalarType_co] - | Sequence[AnalogWaveform[_ScalarType_co]] + npt.NDArray[_TRaw_co] + | NumericWaveform[_TRaw_co, _TScaled_co] + | Sequence[NumericWaveform[_TRaw_co, _TScaled_co]] ), /, timestamps: Sequence[dt.datetime] | Sequence[ht.datetime] | None = None, ) -> None: - """Append data to the analog waveform. + """Append data to the waveform. Args: other: The array or waveform(s) to append. @@ -789,10 +666,10 @@ def append( """ if isinstance(other, np.ndarray): self._append_array(other, timestamps) - elif isinstance(other, AnalogWaveform): + elif isinstance(other, NumericWaveform): validate_unsupported_arg("timestamps", timestamps) self._append_waveform(other) - elif isinstance(other, Sequence) and all(isinstance(x, AnalogWaveform) for x in other): + elif isinstance(other, Sequence) and all(isinstance(x, NumericWaveform) for x in other): validate_unsupported_arg("timestamps", timestamps) self._append_waveforms(other) else: @@ -800,7 +677,7 @@ def append( def _append_array( self, - array: npt.NDArray[_ScalarType_co], + array: npt.NDArray[_TRaw_co], timestamps: Sequence[dt.datetime] | Sequence[ht.datetime] | None = None, ) -> None: if array.dtype != self.dtype: @@ -823,10 +700,12 @@ def _append_array( self._data[offset : offset + len(array)] = array self._sample_count += len(array) - def _append_waveform(self, waveform: AnalogWaveform[_ScalarType_co]) -> None: + def _append_waveform(self, waveform: NumericWaveform[_TRaw_co, _TScaled_co]) -> None: self._append_waveforms([waveform]) - def _append_waveforms(self, waveforms: Sequence[AnalogWaveform[_ScalarType_co]]) -> None: + def _append_waveforms( + self, waveforms: Sequence[NumericWaveform[_TRaw_co, _TScaled_co]] + ) -> None: for waveform in waveforms: if waveform.dtype != self.dtype: raise input_waveform_data_type_mismatch(waveform.dtype, self.dtype) @@ -854,7 +733,7 @@ def _increase_capacity(self, amount: int) -> None: def load_data( self, - array: npt.NDArray[_ScalarType_co], + array: npt.NDArray[_TRaw_co], *, copy: bool = True, start_index: SupportsIndex | None = 0, @@ -865,8 +744,8 @@ def load_data( Args: array: A NumPy array containing the data to load. copy: Specifies whether to copy the array or save a reference to it. - start_index: The sample index at which the analog waveform data begins. - sample_count: The number of samples in the analog waveform. + start_index: The sample index at which the waveform data begins. + sample_count: The number of samples in the waveform. """ if isinstance(array, np.ndarray): self._load_array(array, copy=copy, start_index=start_index, sample_count=sample_count) @@ -875,7 +754,7 @@ def load_data( def _load_array( self, - array: npt.NDArray[_ScalarType_co], + array: npt.NDArray[_TRaw_co], *, copy: bool = True, start_index: SupportsIndex | None = 0, diff --git a/tests/unit/complex/test_dtypes.py b/tests/unit/complex/test_dtypes.py index f9e4df2e..a454d7a0 100644 --- a/tests/unit/complex/test_dtypes.py +++ b/tests/unit/complex/test_dtypes.py @@ -51,3 +51,21 @@ def test___complexint32_array_and_int16_array___add___raises_type_error() -> Non with pytest.raises(TypeError): _ = left + right # type: ignore[operator] + + +def test___unknown_structured_dtype___equality___not_equal() -> None: + dtype = np.dtype([("a", np.int16), ("b", np.int16)]) + + assert dtype != ComplexInt32DType + + +def test___duplicate_structured_dtype___equality___equal() -> None: + dtype = np.dtype([("real", np.int16), ("imag", np.int16)]) + + assert dtype == ComplexInt32DType + + +def test___structured_dtype_str___equality___not_equal() -> None: + dtype = np.dtype("i2, i2") + + assert dtype != ComplexInt32DType diff --git a/tests/unit/waveform/_scaling/test_linear.py b/tests/unit/waveform/_scaling/test_linear.py index 06a1dd71..484710bf 100644 --- a/tests/unit/waveform/_scaling/test_linear.py +++ b/tests/unit/waveform/_scaling/test_linear.py @@ -82,6 +82,28 @@ def test___float64_ndarray___transform_data___returns_float64_scaled_data() -> N assert list(scaled_data) == [4.0, 7.0, 10.0, 13.0] +def test___complex64_ndarray___transform_data___returns_complex64_scaled_data() -> None: + raw_data = np.array([1 + 2j, 3 - 4j], np.complex64) + scale_mode = LinearScaleMode(3.0, 4.0) + + scaled_data = scale_mode._transform_data(raw_data) + + assert_type(scaled_data, npt.NDArray[np.complex64]) + assert isinstance(scaled_data, np.ndarray) and scaled_data.dtype == np.complex64 + assert list(scaled_data) == [7 + 6j, 13 - 12j] + + +def test___complex128_ndarray___transform_data___returns_complex128_scaled_data() -> None: + raw_data = np.array([1 + 2j, 3 - 4j], np.complex128) + scale_mode = LinearScaleMode(3.0, 4.0) + + scaled_data = scale_mode._transform_data(raw_data) + + assert_type(scaled_data, npt.NDArray[np.complex128]) + assert isinstance(scaled_data, np.ndarray) and scaled_data.dtype == np.complex128 + assert list(scaled_data) == [7 + 6j, 13 - 12j] + + @pytest.mark.parametrize( "left, right", [ diff --git a/tests/unit/waveform/_scaling/test_none.py b/tests/unit/waveform/_scaling/test_none.py index 62e71c21..3453432e 100644 --- a/tests/unit/waveform/_scaling/test_none.py +++ b/tests/unit/waveform/_scaling/test_none.py @@ -45,6 +45,26 @@ def test___float64_ndarray___transform_data___returns_float64_scaled_data() -> N assert list(scaled_data) == [0.0, 1.0, 2.0, 3.0] +def test___complex64_ndarray___transform_data___returns_complex64_scaled_data() -> None: + raw_data = np.array([1 + 2j, 3 - 4j], np.complex64) + + scaled_data = NO_SCALING._transform_data(raw_data) + + assert_type(scaled_data, npt.NDArray[np.complex64]) + assert isinstance(scaled_data, np.ndarray) and scaled_data.dtype == np.complex64 + assert list(scaled_data) == [1 + 2j, 3 - 4j] + + +def test___complex128_ndarray___transform_data___returns_complex128_scaled_data() -> None: + raw_data = np.array([1 + 2j, 3 - 4j], np.complex128) + + scaled_data = NO_SCALING._transform_data(raw_data) + + assert_type(scaled_data, npt.NDArray[np.complex128]) + assert isinstance(scaled_data, np.ndarray) and scaled_data.dtype == np.complex128 + assert list(scaled_data) == [1 + 2j, 3 - 4j] + + def test___scale_mode___repr___looks_ok() -> None: assert repr(NO_SCALING) == "nitypes.waveform.NoneScaleMode()" diff --git a/tests/unit/waveform/test_analog_waveform.py b/tests/unit/waveform/test_analog_waveform.py index 67410680..13048316 100644 --- a/tests/unit/waveform/test_analog_waveform.py +++ b/tests/unit/waveform/test_analog_waveform.py @@ -88,13 +88,29 @@ def test___sample_count_dtype_and_capacity___create___creates_waveform_with_samp assert_type(waveform, AnalogWaveform[np.int32]) -def test___sample_count_and_unsupported_dtype___create___raises_type_error() -> None: +@pytest.mark.parametrize("dtype", [np.complex128, np.str_, np.void, "i2, i2"]) +def test___sample_count_and_unsupported_dtype___create___raises_type_error( + dtype: npt.DTypeLike, +) -> None: with pytest.raises(TypeError) as exc: - _ = AnalogWaveform(10, np.complex128) + _ = AnalogWaveform(10, dtype) assert exc.value.args[0].startswith("The requested data type is not supported.") +def test___dtype_str_with_unsupported_traw_hint___create___mypy_type_var_warning() -> None: + waveform1: AnalogWaveform[np.complex128] = AnalogWaveform(dtype="int32") # type: ignore[type-var] + waveform2: AnalogWaveform[np.str_] = AnalogWaveform(dtype="int32") # type: ignore[type-var] + waveform3: AnalogWaveform[np.void] = AnalogWaveform(dtype="int32") # type: ignore[type-var] + _ = waveform1, waveform2, waveform3 + + +def test___dtype_str_with_traw_hint___create___narrows_traw() -> None: + waveform: AnalogWaveform[np.int32] = AnalogWaveform(dtype="int32") + + assert_type(waveform, AnalogWaveform[np.int32]) + + ############################################################################### # from_array_1d ############################################################################### @@ -205,7 +221,7 @@ def test___iterable___from_array_1d___raises_type_error() -> None: def test___ndarray_with_unsupported_dtype___from_array_1d___raises_type_error() -> None: - data = np.zeros(3, np.complex128) + data = np.zeros(3, np.str_) with pytest.raises(TypeError) as exc: _ = AnalogWaveform.from_array_1d(data) @@ -456,7 +472,7 @@ def test___iterable_list___from_array_2d___raises_type_error() -> None: def test___ndarray_with_unsupported_dtype___from_array_2d___raises_type_error() -> None: - data = np.zeros((2, 3), np.complex128) + data = np.zeros((2, 3), np.str_) with pytest.raises(TypeError) as exc: _ = AnalogWaveform.from_array_2d(data) @@ -787,6 +803,7 @@ def test___unsupported_dtype___get_scaled_data___raises_type_error() -> None: _ = waveform.get_scaled_data(np.int32) assert exc.value.args[0].startswith("The requested data type is not supported.") + assert "Data type: int32" in exc.value.args[0] assert "Supported data types: float32, float64" in exc.value.args[0] diff --git a/tests/unit/waveform/test_complex_waveform.py b/tests/unit/waveform/test_complex_waveform.py new file mode 100644 index 00000000..25df9d94 --- /dev/null +++ b/tests/unit/waveform/test_complex_waveform.py @@ -0,0 +1,324 @@ +from __future__ import annotations + +from typing import Any + +import numpy as np +import numpy.typing as npt +import pytest +from typing_extensions import assert_type + +from nitypes.complex import ComplexInt32Base, ComplexInt32DType +from nitypes.waveform import ComplexWaveform, LinearScaleMode + + +############################################################################### +# create +############################################################################### +def test___sample_count_and_complexint32_dtype___create___creates_waveform_with_sample_count_and_dtype() -> ( + None +): + waveform = ComplexWaveform(10, ComplexInt32DType) + + assert waveform.sample_count == waveform.capacity == len(waveform.raw_data) == 10 + assert waveform.dtype == ComplexInt32DType + assert_type(waveform, ComplexWaveform[ComplexInt32Base]) + + +def test___sample_count_and_complex64_dtype___create___creates_waveform_with_sample_count_and_dtype() -> ( + None +): + waveform = ComplexWaveform(10, np.complex64) + + assert waveform.sample_count == waveform.capacity == len(waveform.raw_data) == 10 + assert waveform.dtype == np.complex64 + assert_type(waveform, ComplexWaveform[np.complex64]) + + +def test___sample_count_and_complex128_dtype___create___creates_waveform_with_sample_count_and_dtype() -> ( + None +): + waveform = ComplexWaveform(10, np.complex128) + + assert waveform.sample_count == waveform.capacity == len(waveform.raw_data) == 10 + assert waveform.dtype == np.complex128 + assert_type(waveform, ComplexWaveform[np.complex128]) + + +def test___sample_count_and_unknown_structured_dtype___create___raises_type_error() -> None: + dtype = np.dtype([("a", np.int16), ("b", np.int32)]) + + with pytest.raises(TypeError) as exc: + waveform = ComplexWaveform(10, dtype) + + # ComplexWaveform currently cannot distinguish between ComplexInt32DType and other + # structured data types at type-checking time. + assert_type(waveform, ComplexWaveform[ComplexInt32Base]) + + assert exc.value.args[0].startswith("The requested data type is not supported.") + assert "Data type: [('a', ' None: + with pytest.raises(TypeError) as exc: + _ = ComplexWaveform(10, "i2, i2") + + assert exc.value.args[0].startswith("The requested data type is not supported.") + assert "Data type: [('f0', ' None: + with pytest.raises(TypeError) as exc: + _ = ComplexWaveform(10, dtype) + + assert exc.value.args[0].startswith("The requested data type is not supported.") + + +def test___dtype_str_with_unsupported_traw_hint___create___mypy_type_var_warning() -> None: + waveform1: ComplexWaveform[np.int32] = ComplexWaveform(dtype="complex64") # type: ignore[type-var] + waveform2: ComplexWaveform[np.float64] = ComplexWaveform(dtype="complex64") # type: ignore[type-var] + waveform3: ComplexWaveform[np.str_] = ComplexWaveform(dtype="complex64") # type: ignore[type-var] + _ = waveform1, waveform2, waveform3 + + +def test___dtype_str_with_traw_hint___create___narrows_traw() -> None: + waveform: ComplexWaveform[np.complex64] = ComplexWaveform(dtype="complex64") + + assert_type(waveform, ComplexWaveform[np.complex64]) + + +############################################################################### +# from_array_1d +############################################################################### +def test___complexint32_ndarray___from_array_1d___creates_waveform_with_complexint32_dtype() -> ( + None +): + data = np.array([(1, 2), (3, -4)], ComplexInt32DType) + + waveform = ComplexWaveform.from_array_1d(data) + + assert waveform.raw_data.tolist() == data.tolist() + assert waveform.dtype == ComplexInt32DType + assert_type(waveform, ComplexWaveform[ComplexInt32Base]) + + +def test___complex64_ndarray___from_array_1d___creates_waveform_with_complex64_dtype() -> None: + data = np.array([1.1 + 2.2j, 3.3 - 4.4j], np.complex64) + + waveform = ComplexWaveform.from_array_1d(data) + + assert waveform.raw_data.tolist() == data.tolist() + assert waveform.dtype == np.complex64 + assert_type(waveform, ComplexWaveform[np.complex64]) + + +def test___complex128_ndarray___from_array_1d___creates_waveform_with_complex128_dtype() -> None: + data = np.array([1.1 + 2.2j, 3.3 - 4.4j], np.complex128) + + waveform = ComplexWaveform.from_array_1d(data) + + assert waveform.raw_data.tolist() == data.tolist() + assert waveform.dtype == np.complex128 + assert_type(waveform, ComplexWaveform[np.complex128]) + + +def test___complex_list_with_dtype___from_array_1d___creates_waveform_with_specified_dtype() -> ( + None +): + data = [1.1 + 2.2j, 3.3 - 4.4j] + + waveform = ComplexWaveform.from_array_1d(data, np.complex64) + + assert waveform.raw_data.tolist() == pytest.approx(data) + assert waveform.dtype == np.complex64 + assert_type(waveform, ComplexWaveform[np.complex64]) + + +def test___complex_list_with_dtype_str___from_array_1d___creates_waveform_with_specified_dtype() -> ( + None +): + data = [1.1 + 2.2j, 3.3 - 4.4j] + + waveform = ComplexWaveform.from_array_1d(data, "complex64") + + assert waveform.raw_data.tolist() == pytest.approx(data) + assert waveform.dtype == np.complex64 + assert_type(waveform, ComplexWaveform[Any]) # dtype not inferred from string + + +############################################################################### +# from_array_2d +############################################################################### +def test___complexint32_ndarray___from_array_2d___creates_waveform_with_complexint32_dtype() -> ( + None +): + data = np.array([[(1, 2), (3, -4), (-5, 6)], [(-7 - 8), (9, 10), (11, 12)]], ComplexInt32DType) + + waveforms = ComplexWaveform.from_array_2d(data) + + assert len(waveforms) == 2 + for i in range(len(waveforms)): + assert waveforms[i].raw_data.tolist() == data[i].tolist() + assert waveforms[i].dtype == ComplexInt32DType + assert_type(waveforms[i], ComplexWaveform[ComplexInt32Base]) + + +def test___complex64_ndarray___from_array_2d___creates_waveform_with_complex64_dtype() -> None: + data = np.array([[1 + 2j, 3 - 4j, -5 + 6j], [-7 - 8j, 9 + 10j, 11 + 12j]], np.complex64) + + waveforms = ComplexWaveform.from_array_2d(data) + + assert len(waveforms) == 2 + for i in range(len(waveforms)): + assert waveforms[i].raw_data.tolist() == data[i].tolist() + assert waveforms[i].dtype == np.complex64 + assert_type(waveforms[i], ComplexWaveform[np.complex64]) + + +def test___complex128_ndarray___from_array_2d___creates_waveform_with_complex128_dtype() -> None: + data = np.array([[1 + 2j, 3 - 4j, -5 + 6j], [-7 - 8j, 9 + 10j, 11 + 12j]], np.complex128) + + waveforms = ComplexWaveform.from_array_2d(data) + + assert len(waveforms) == 2 + for i in range(len(waveforms)): + assert waveforms[i].raw_data.tolist() == data[i].tolist() + assert waveforms[i].dtype == np.complex128 + assert_type(waveforms[i], ComplexWaveform[np.complex128]) + + +def test___complex_list_list_with_dtype___from_array_2d___creates_waveform_with_specified_dtype() -> ( + None +): + data = [[1 + 2j, 3 - 4j, -5 + 6j], [-7 - 8j, 9 + 10j, 11 + 12j]] + + waveforms = ComplexWaveform.from_array_2d(data, np.complex64) + + assert len(waveforms) == 2 + for i in range(len(waveforms)): + assert waveforms[i].raw_data.tolist() == data[i] + assert waveforms[i].dtype == np.complex64 + assert_type(waveforms[i], ComplexWaveform[np.complex64]) + + +def test___int_list_list_with_dtype_str___from_array_2d___creates_waveform_with_specified_dtype() -> ( + None +): + data = [[1 + 2j, 3 - 4j, -5 + 6j], [-7 - 8j, 9 + 10j, 11 + 12j]] + + waveforms = ComplexWaveform.from_array_2d(data, "complex64") + + assert len(waveforms) == 2 + for i in range(len(waveforms)): + assert waveforms[i].raw_data.tolist() == data[i] + assert waveforms[i].dtype == np.complex64 + assert_type(waveforms[i], ComplexWaveform[Any]) # dtype not inferred from string + + +############################################################################### +# raw_data +############################################################################### +def test___complexint32_waveform___raw_data___returns_complexint32_data() -> None: + waveform = ComplexWaveform.from_array_1d([(1, 2), (3, -4)], ComplexInt32DType) + + raw_data = waveform.raw_data + + assert_type(raw_data, npt.NDArray[ComplexInt32Base]) + assert isinstance(raw_data, np.ndarray) and raw_data.dtype == ComplexInt32DType + assert [x.item() for x in raw_data] == [(1, 2), (3, -4)] + + +def test___complex64_waveform___raw_data___returns_complex64_data() -> None: + waveform = ComplexWaveform.from_array_1d([1 + 2j, 3 - 4j], np.complex64) + + raw_data = waveform.raw_data + + assert_type(raw_data, npt.NDArray[np.complex64]) + assert isinstance(raw_data, np.ndarray) and raw_data.dtype == np.complex64 + assert list(raw_data) == [1 + 2j, 3 - 4j] + + +############################################################################### +# scaled_data +############################################################################### +def test___complexint32_waveform___scaled_data___converts_to_complex128() -> None: + waveform = ComplexWaveform.from_array_1d([(1, 2), (3, -4)], ComplexInt32DType) + + scaled_data = waveform.scaled_data + + assert_type(scaled_data, npt.NDArray[np.complex128]) + assert isinstance(scaled_data, np.ndarray) and scaled_data.dtype == np.complex128 + assert list(scaled_data) == [1 + 2j, 3 - 4j] + + +def test___complex64_waveform___scaled_data___converts_to_complex128() -> None: + waveform = ComplexWaveform.from_array_1d([1 + 2j, 3 - 4j], np.complex64) + + scaled_data = waveform.scaled_data + + assert_type(scaled_data, npt.NDArray[np.complex128]) + assert isinstance(scaled_data, np.ndarray) and scaled_data.dtype == np.complex128 + assert list(scaled_data) == [1 + 2j, 3 - 4j] + + +def test___complexint32_waveform_with_linear_scale___scaled_data___converts_to_complex128() -> None: + waveform = ComplexWaveform.from_array_1d([(1, 2), (3, -4)], ComplexInt32DType) + waveform.scale_mode = LinearScaleMode(2.0, 0.5) + + scaled_data = waveform.scaled_data + + assert_type(scaled_data, npt.NDArray[np.complex128]) + assert isinstance(scaled_data, np.ndarray) and scaled_data.dtype == np.complex128 + assert list(scaled_data) == [2.5 + 4j, 6.5 - 8j] + + +def test___complex64_waveform_with_linear_scale___scaled_data___converts_to_complex128() -> None: + waveform = ComplexWaveform.from_array_1d([1 + 2j, 3 - 4j], np.complex64) + waveform.scale_mode = LinearScaleMode(2.0, 0.5) + + scaled_data = waveform.scaled_data + + assert_type(scaled_data, npt.NDArray[np.complex128]) + assert isinstance(scaled_data, np.ndarray) and scaled_data.dtype == np.complex128 + assert list(scaled_data) == [2.5 + 4j, 6.5 - 8j] + + +############################################################################### +# get_scaled_data +############################################################################### +def test___complexint32_waveform_with_complex64_dtype___get_scaled_data___converts_to_complex64() -> ( + None +): + waveform = ComplexWaveform.from_array_1d([(1, 2), (3, -4)], ComplexInt32DType) + + scaled_data = waveform.get_scaled_data(np.complex64) + + assert_type(scaled_data, npt.NDArray[np.complex64]) + assert isinstance(scaled_data, np.ndarray) and scaled_data.dtype == np.complex64 + assert list(scaled_data) == [1 + 2j, 3 - 4j] + + +def test___complex64_waveform_with_complex64_dtype___get_scaled_data___does_not_convert() -> None: + waveform = ComplexWaveform.from_array_1d([1 + 2j, 3 - 4j], np.complex64) + + scaled_data = waveform.get_scaled_data(np.complex64) + + assert_type(scaled_data, npt.NDArray[np.complex64]) + assert isinstance(scaled_data, np.ndarray) and scaled_data.dtype == np.complex64 + assert list(scaled_data) == [1 + 2j, 3 - 4j] + + +def test___complexint32_waveform_with_unknown_structured_dtype___get_scaled_data___raises_type_error() -> ( + None +): + waveform = ComplexWaveform.from_array_1d([(1, 2), (3, -4)], ComplexInt32DType) + dtype = np.dtype([("a", np.int16), ("b", np.int16)]) + + with pytest.raises(TypeError) as exc: + _ = waveform.get_scaled_data(dtype) + + assert exc.value.args[0].startswith("The requested data type is not supported.") + assert "Data type: [('a', '