waveform: Add timing

pyproject.toml

@@ -14,7 +14,8 @@ numpy = [
   {version = ">=1.26", python = ">=3.12,<3.13"},
   {version = ">=2.1", python = "^3.13"},
 ]
-hightime = "^0.2.2"
+# hightime = "^0.2.2"
+hightime = { git = "https://github.com/ni/hightime.git" }
 
 [tool.poetry.group.lint.dependencies]
 bandit = { version = ">=1.7", extras = ["toml"] }
@@ -43,13 +44,6 @@ plugins = "numpy.typing.mypy_plugin"
 namespace_packages = true
 strict = true
 
-[[tool.mypy.overrides]]
-module = [
-  # https://github.com/ni/hightime/issues/4 - Add type annotations
-  "hightime.*",
-]
-ignore_missing_imports = true
-
 [tool.bandit]
 skips = [
   "B101", # assert_used

src/nitypes/time/__init__.py

@@ -0,0 +1,5 @@
+"""Time data types for NI Python APIs."""
+
+from nitypes.time._conversion import convert_datetime, convert_timedelta
+
+__all__ = ["convert_datetime", "convert_timedelta"]

src/nitypes/time/_conversion.py

@@ -0,0 +1,136 @@
+from __future__ import annotations
+
+import datetime as dt
+import sys
+from collections.abc import Callable
+from functools import singledispatch
+from typing import Any, TypeVar, Union, cast
+
+import hightime as ht
+
+if sys.version_info >= (3, 10):
+    from typing import TypeAlias
+else:
+    from typing_extensions import TypeAlias
+
+_AnyDateTime: TypeAlias = Union[dt.datetime, ht.datetime]
+_TDateTime = TypeVar("_TDateTime", dt.datetime, ht.datetime)
+
+_AnyTimeDelta: TypeAlias = Union[dt.timedelta, ht.timedelta]
+_TTimeDelta = TypeVar("_TTimeDelta", dt.timedelta, ht.timedelta)
+
+
+def convert_datetime(requested_type: type[_TDateTime], value: _AnyDateTime, /) -> _TDateTime:
+    """Convert a datetime object to the specified type."""
+    convert_func = _CONVERT_DATETIME_FOR_TYPE.get(requested_type)
+    if convert_func is None:
+        raise TypeError(
+            "The requested type must be a datetime type.\n\n" f"Requested type: {requested_type}"
+        )
+    return cast(_TDateTime, convert_func(value))
+
+
+@singledispatch
+def _convert_to_dt_datetime(value: object, /) -> dt.datetime:
+    raise TypeError("The value must be a datetime.\n\n" f"Provided value: {value}")
+
+
+@_convert_to_dt_datetime.register
+def _(value: dt.datetime, /) -> dt.datetime:
+    return value
+
+
+@_convert_to_dt_datetime.register
+def _(value: ht.datetime, /) -> dt.datetime:
+    return dt.datetime(
+        value.year,
+        value.month,
+        value.day,
+        value.hour,
+        value.minute,
+        value.second,
+        value.microsecond,
+        value.tzinfo,
+        fold=value.fold,
+    )
+
+
+@singledispatch
+def _convert_to_ht_datetime(value: object, /) -> ht.datetime:
+    raise TypeError("The value must be a datetime.\n\n" f"Provided value: {value}")
+
+
+@_convert_to_ht_datetime.register
+def _(value: dt.datetime, /) -> ht.datetime:
+    return ht.datetime(
+        value.year,
+        value.month,
+        value.day,
+        value.hour,
+        value.minute,
+        value.second,
+        value.microsecond,
+        value.tzinfo,
+        fold=value.fold,
+    )
+
+
+@_convert_to_ht_datetime.register
+def _(value: ht.datetime, /) -> ht.datetime:
+    return value
+
+
+_CONVERT_DATETIME_FOR_TYPE: dict[type[Any], Callable[[object], object]] = {
+    dt.datetime: _convert_to_dt_datetime,
+    ht.datetime: _convert_to_ht_datetime,
+}
+
+
+def convert_timedelta(requested_type: type[_TTimeDelta], value: _AnyTimeDelta, /) -> _TTimeDelta:
+    """Convert a timedelta object to the specified type."""
+    convert_func = _CONVERT_TIMEDELTA_FOR_TYPE.get(requested_type)
+    if convert_func is None:
+        raise TypeError(
+            "The requested type must be a timedelta type.\n\n" f"Requested type: {requested_type}"
+        )
+    return cast(_TTimeDelta, convert_func(value))
+
+
+@singledispatch
+def _convert_to_dt_timedelta(value: object, /) -> dt.timedelta:
+    raise TypeError("The value must be a timedelta.\n\n" f"Provided value: {value}")
+
+
+@_convert_to_dt_timedelta.register
+def _(value: dt.timedelta, /) -> dt.timedelta:
+    return value
+
+
+@_convert_to_dt_timedelta.register
+def _(value: ht.timedelta, /) -> dt.timedelta:
+    return dt.timedelta(value.days, value.seconds, value.microseconds)
+
+
+@singledispatch
+def _convert_to_ht_timedelta(value: object, /) -> ht.timedelta:
+    raise TypeError("The value must be a timedelta.\n\n" f"Provided value: {value}")
+
+
+@_convert_to_ht_timedelta.register
+def _(value: dt.timedelta, /) -> ht.timedelta:
+    return ht.timedelta(
+        value.days,
+        value.seconds,
+        value.microseconds,
+    )
+
+
+@_convert_to_ht_timedelta.register
+def _(value: ht.timedelta, /) -> ht.timedelta:
+    return value
+
+
+_CONVERT_TIMEDELTA_FOR_TYPE: dict[type[Any], Callable[[object], object]] = {
+    dt.timedelta: _convert_to_dt_timedelta,
+    ht.timedelta: _convert_to_ht_timedelta,
+}

src/nitypes/waveform/__init__.py

@@ -5,9 +5,25 @@
     ExtendedPropertyDictionary,
     ExtendedPropertyValue,
 )
+from nitypes.waveform._timing._base import BaseTiming, SampleIntervalMode
+from nitypes.waveform._timing._precision import PrecisionTiming
+from nitypes.waveform._timing._standard import Timing
 
 __all__ = [
     "AnalogWaveform",
+    "BaseTiming",
     "ExtendedPropertyDictionary",
     "ExtendedPropertyValue",
+    "PrecisionTiming",
+    "SampleIntervalMode",
+    "Timing",
 ]
+
+# Hide that it was defined in a helper file
+AnalogWaveform.__module__ = __name__
+BaseTiming.__module__ = __name__
+ExtendedPropertyDictionary.__module__ = __name__
+# ExtendedPropertyValue is a TypeAlias
+PrecisionTiming.__module__ = __name__
+SampleIntervalMode.__module__ = __name__
+Timing.__module__ = __name__

src/nitypes/waveform/_analog_waveform.py

@@ -12,6 +12,9 @@
     UNIT_DESCRIPTION,
     ExtendedPropertyDictionary,
 )
+from nitypes.waveform._timing._conversion import convert_timing
+from nitypes.waveform._timing._precision import PrecisionTiming
+from nitypes.waveform._timing._standard import Timing
 from nitypes.waveform._utils import arg_to_uint, validate_dtype
 
 if sys.version_info < (3, 10):
@@ -211,12 +214,22 @@ def from_array_2d(
             for i in range(len(array))
         ]
 
-    __slots__ = ["_data", "_start_index", "_sample_count", "_extended_properties", "__weakref__"]
+    __slots__ = [
+        "_data",
+        "_start_index",
+        "_sample_count",
+        "_extended_properties",
+        "_timing",
+        "_precision_timing",
+        "__weakref__",
+    ]
 
     _data: npt.NDArray[_ScalarType_co]
     _start_index: int
     _sample_count: int
     _extended_properties: ExtendedPropertyDictionary
+    _timing: Timing | None
+    _precision_timing: PrecisionTiming | None
 
     # If neither dtype nor _data is specified, the type parameter defaults to np.float64.
     @overload
@@ -332,6 +345,8 @@ def _init_with_new_array(
         self._start_index = start_index
         self._sample_count = sample_count
         self._extended_properties = ExtendedPropertyDictionary()
+        self._timing = Timing.empty
+        self._precision_timing = None
 
     def _init_with_provided_array(
         self,
@@ -384,6 +399,8 @@ def _init_with_provided_array(
         self._start_index = start_index
         self._sample_count = sample_count
         self._extended_properties = ExtendedPropertyDictionary()
+        self._timing = Timing.empty
+        self._precision_timing = None
 
     @property
     def raw_data(self) -> npt.NDArray[_ScalarType_co]:
@@ -464,3 +481,62 @@ def unit_description(self, value: str) -> None:
                 "The unit description must be a str.\n\n" f"Unit description: {value!r}"
             )
         self._extended_properties[UNIT_DESCRIPTION] = value
+
+    @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
+
+    @timing.setter
+    def timing(self, value: Timing) -> None:
+        if not isinstance(value, Timing):
+            raise TypeError("The timing information must be a Timing object.")
+        self._timing = value
+        self._precision_timing = None
+
+    @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
+
+    @property
+    def precision_timing(self) -> PrecisionTiming:
+        """The precision timing information of the analog 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
+        with up to yoctosecond precision. If the timing information is set using
+        AnalogWaveform.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.
+        """
+        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
+
+    @precision_timing.setter
+    def precision_timing(self, value: PrecisionTiming) -> None:
+        if not isinstance(value, PrecisionTiming):
+            raise TypeError("The precision timing information must be a PrecisionTiming object.")
+        self._precision_timing = value
+        self._timing = None

src/nitypes/waveform/_timing/__init__.py

@@ -0,0 +1 @@
+"""Waveform timing data types for NI Python APIs."""