Add lib and grpc time data types and conversions to enable future NI-DAQmx time support

generated/nidaqmx/_grpc_time.py

@@ -1,5 +1,7 @@
 from datetime import timezone
+from datetime import datetime as std_datetime
 from hightime import datetime as ht_datetime
+from typing import Union
 
 from google.protobuf.internal.well_known_types import Timestamp as GrpcTimestamp
 
@@ -12,7 +14,7 @@
 _YS_PER_FS = 10**9
 
 
-def convert_time_to_timestamp(dt, ts):
+def convert_time_to_timestamp(dt: Union[std_datetime, ht_datetime], ts: GrpcTimestamp) -> None:
     utc_dt = dt.astimezone(tz=timezone.utc)
     seconds = int(utc_dt.timestamp())
 
@@ -30,7 +32,7 @@ def convert_time_to_timestamp(dt, ts):
     ts.FromNanoseconds(seconds * _NS_PER_S + nanos)
 
 
-def convert_timestamp_to_time(ts, tzinfo=None):
+def convert_timestamp_to_time(ts: GrpcTimestamp, tzinfo: timezone = None) -> ht_datetime:
     total_nanos = ts.ToNanoseconds()
     seconds, nanos = divmod(total_nanos, _NS_PER_S)
 

generated/nidaqmx/_lib_time.py

@@ -1,7 +1,9 @@
 import ctypes
 import functools
 from datetime import timezone
+from datetime import datetime as std_datetime
 from hightime import datetime as ht_datetime
+from typing import Union
 
 
 @functools.total_ordering
@@ -26,7 +28,7 @@ class AbsoluteTime(ctypes.Structure):
     MAX_YS = 10**9
 
     @classmethod
-    def from_datetime(cls, dt):
+    def from_datetime(cls, dt: Union[std_datetime, ht_datetime]) -> "AbsoluteTime":
         utc_dt = dt.astimezone(tz=timezone.utc)
 
         # First, calculate whole seconds by converting from the 1970 to 1904 epoch.
@@ -54,7 +56,7 @@ def from_datetime(cls, dt):
 
         return AbsoluteTime(lsb=lsb, msb=timestamp_1904_epoch)
 
-    def to_datetime(self, tzinfo=None):
+    def to_datetime(self, tzinfo: timezone = None):
         # First, calculate whole seconds by converting from the 1904 to 1970 epoch.
         timestamp_1904_epoch = self.msb
         was_positive = timestamp_1904_epoch > 0
@@ -82,13 +84,13 @@ def to_datetime(self, tzinfo=None):
         # Then convert to requested timezone
         return dt.astimezone(tz=tzinfo)
 
-    def __str__(self):
+    def __str__(self) -> str:
         return f"AbsoluteTime(lsb=0x{self.lsb:x}, msb=0x{self.msb:x})"
 
-    def __eq__(self, other):
+    def __eq__(self, other) -> bool:
         return self.msb == other.msb and self.lsb == other.lsb
 
-    def __lt__(self, other):
+    def __lt__(self, other) -> bool:
         if self.msb == other.msb:
             return self.lsb < other.lsb
         else:

src/handwritten/_grpc_time.py

@@ -1,5 +1,7 @@
 from datetime import timezone
+from datetime import datetime as std_datetime
 from hightime import datetime as ht_datetime
+from typing import Union
 
 from google.protobuf.internal.well_known_types import Timestamp as GrpcTimestamp
 
@@ -12,7 +14,7 @@
 _YS_PER_FS = 10**9
 
 
-def convert_time_to_timestamp(dt, ts):
+def convert_time_to_timestamp(dt: Union[std_datetime, ht_datetime], ts: GrpcTimestamp) -> None:
     utc_dt = dt.astimezone(tz=timezone.utc)
     seconds = int(utc_dt.timestamp())
 
@@ -30,7 +32,7 @@ def convert_time_to_timestamp(dt, ts):
     ts.FromNanoseconds(seconds * _NS_PER_S + nanos)
 
 
-def convert_timestamp_to_time(ts, tzinfo=None):
+def convert_timestamp_to_time(ts: GrpcTimestamp, tzinfo: timezone = None) -> ht_datetime:
     total_nanos = ts.ToNanoseconds()
     seconds, nanos = divmod(total_nanos, _NS_PER_S)
 

src/handwritten/_lib_time.py

@@ -1,7 +1,9 @@
 import ctypes
 import functools
 from datetime import timezone
+from datetime import datetime as std_datetime
 from hightime import datetime as ht_datetime
+from typing import Union
 
 
 @functools.total_ordering
@@ -26,7 +28,7 @@ class AbsoluteTime(ctypes.Structure):
     MAX_YS = 10**9
 
     @classmethod
-    def from_datetime(cls, dt):
+    def from_datetime(cls, dt: Union[std_datetime, ht_datetime]) -> "AbsoluteTime":
         utc_dt = dt.astimezone(tz=timezone.utc)
 
         # First, calculate whole seconds by converting from the 1970 to 1904 epoch.
@@ -54,7 +56,7 @@ def from_datetime(cls, dt):
 
         return AbsoluteTime(lsb=lsb, msb=timestamp_1904_epoch)
 
-    def to_datetime(self, tzinfo=None):
+    def to_datetime(self, tzinfo: timezone = None):
         # First, calculate whole seconds by converting from the 1904 to 1970 epoch.
         timestamp_1904_epoch = self.msb
         was_positive = timestamp_1904_epoch > 0
@@ -82,13 +84,13 @@ def to_datetime(self, tzinfo=None):
         # Then convert to requested timezone
         return dt.astimezone(tz=tzinfo)
 
-    def __str__(self):
+    def __str__(self) -> str:
         return f"AbsoluteTime(lsb=0x{self.lsb:x}, msb=0x{self.msb:x})"
 
-    def __eq__(self, other):
+    def __eq__(self, other) -> bool:
         return self.msb == other.msb and self.lsb == other.lsb
 
-    def __lt__(self, other):
+    def __lt__(self, other) -> bool:
         if self.msb == other.msb:
             return self.lsb < other.lsb
         else:

tests/unit/_time_utils.py

@@ -4,9 +4,9 @@
 from hightime import datetime as ht_datetime
 
 # Jan 1, 2002 = 32 years + 8 leapdays = 11688 days = 1009843200 seconds
-JAN_01_2022_TIMESTAMP_1970_EPOCH = 0x3C30FC00
+JAN_01_2002_TIMESTAMP_1970_EPOCH = 0x3C30FC00
 # Jan 1, 2002 = 98 years + 25 leapdays = 35795 days = 3092688000 seconds
-JAN_01_2022_TIMESTAMP_1904_EPOCH = 0xB856AC80
+JAN_01_2002_TIMESTAMP_1904_EPOCH = 0xB856AC80
 
-JAN_01_2022_DATETIME = std_datetime(2002, 1, 1, tzinfo=timezone.utc)
-JAN_01_2022_HIGHTIME = ht_datetime(2002, 1, 1, tzinfo=timezone.utc)
+JAN_01_2002_DATETIME = std_datetime(2002, 1, 1, tzinfo=timezone.utc)
+JAN_01_2002_HIGHTIME = ht_datetime(2002, 1, 1, tzinfo=timezone.utc)

tests/unit/test_grpc_time.py

@@ -6,24 +6,47 @@
 from hightime import datetime as ht_datetime
 
 import nidaqmx._grpc_time as grpc_time
+import nidaqmx._stubs.nidaqmx_pb2 as nidaqmx_pb2
 from tests.unit._time_utils import (
-    JAN_01_2022_TIMESTAMP_1970_EPOCH,
-    JAN_01_2022_DATETIME,
-    JAN_01_2022_HIGHTIME,
+    JAN_01_2002_TIMESTAMP_1970_EPOCH,
+    JAN_01_2002_DATETIME,
+    JAN_01_2002_HIGHTIME,
 )
 
 
-@pytest.mark.parametrize("from_dt", [(JAN_01_2022_DATETIME), (JAN_01_2022_HIGHTIME)])
+@pytest.mark.parametrize("from_dt", [(JAN_01_2002_DATETIME), (JAN_01_2002_HIGHTIME)])
 def test___utc_datetime___convert_to_timestamp___is_reversible(from_dt):
     to_ts = GrpcTimestamp()
     grpc_time.convert_time_to_timestamp(from_dt, to_ts)
     roundtrip_dt = grpc_time.convert_timestamp_to_time(to_ts, tzinfo=timezone.utc)
 
     total_nanoseconds = to_ts.ToNanoseconds()
     seconds, nanos = divmod(total_nanoseconds, grpc_time._NS_PER_S)
-    assert seconds == JAN_01_2022_TIMESTAMP_1970_EPOCH
+    assert seconds == JAN_01_2002_TIMESTAMP_1970_EPOCH
     assert nanos == 0
-    assert roundtrip_dt == JAN_01_2022_HIGHTIME
+    assert roundtrip_dt == JAN_01_2002_HIGHTIME
+
+
+@pytest.mark.parametrize("request_dt", [(JAN_01_2002_DATETIME), (JAN_01_2002_HIGHTIME)])
+def test___utc_datetime___convert_to_grpc_request___succeeds(request_dt):
+    request = nidaqmx_pb2.CfgTimeStartTrigRequest()
+
+    grpc_time.convert_time_to_timestamp(request_dt, request.when)
+
+    total_nanoseconds = request.when.ToNanoseconds()
+    seconds, nanos = divmod(total_nanoseconds, grpc_time._NS_PER_S)
+    assert seconds == JAN_01_2002_TIMESTAMP_1970_EPOCH
+    assert nanos == 0
+
+
+@pytest.mark.parametrize("response_dt", [(JAN_01_2002_DATETIME), (JAN_01_2002_HIGHTIME)])
+def test___grpc_response___convert_to_timestamp___succeeds(response_dt):
+    response = nidaqmx_pb2.GetStartTrigTrigWhenResponse()
+    grpc_time.convert_time_to_timestamp(response_dt, response.data)
+
+    to_dt = grpc_time.convert_timestamp_to_time(response.data, tzinfo=timezone.utc)
+
+    assert to_dt == JAN_01_2002_HIGHTIME
 
 
 @pytest.mark.parametrize(
@@ -48,26 +71,26 @@ def test___tz_datetime___convert_to_timestamp___is_reversible(
     grpc_time.convert_time_to_timestamp(from_dt, to_ts)
     roundtrip_dt = grpc_time.convert_timestamp_to_time(to_ts, tzinfo=tzinfo)
 
-    assert to_ts.seconds == JAN_01_2022_TIMESTAMP_1970_EPOCH + expected_offset
+    assert to_ts.seconds == JAN_01_2002_TIMESTAMP_1970_EPOCH + expected_offset
     assert to_ts.nanos == 0
     assert from_dt == roundtrip_dt
 
 
 @pytest.mark.parametrize(
     "base_dt, microsecond, nanoseconds",
     [
-        (JAN_01_2022_DATETIME, 0, 0),
-        (JAN_01_2022_DATETIME, 1, 1000),
-        (JAN_01_2022_DATETIME, 250000, 250000000),
-        (JAN_01_2022_DATETIME, 500000, 500000000),
-        (JAN_01_2022_DATETIME, 750000, 750000000),
-        (JAN_01_2022_DATETIME, 999999, 999999000),
-        (JAN_01_2022_HIGHTIME, 0, 0),
-        (JAN_01_2022_HIGHTIME, 1, 1000),
-        (JAN_01_2022_HIGHTIME, 250000, 250000000),
-        (JAN_01_2022_HIGHTIME, 500000, 500000000),
-        (JAN_01_2022_HIGHTIME, 750000, 750000000),
-        (JAN_01_2022_HIGHTIME, 999999, 999999000),
+        (JAN_01_2002_DATETIME, 0, 0),
+        (JAN_01_2002_DATETIME, 1, 1000),
+        (JAN_01_2002_DATETIME, 250000, 250000000),
+        (JAN_01_2002_DATETIME, 500000, 500000000),
+        (JAN_01_2002_DATETIME, 750000, 750000000),
+        (JAN_01_2002_DATETIME, 999999, 999999000),
+        (JAN_01_2002_HIGHTIME, 0, 0),
+        (JAN_01_2002_HIGHTIME, 1, 1000),
+        (JAN_01_2002_HIGHTIME, 250000, 250000000),
+        (JAN_01_2002_HIGHTIME, 500000, 500000000),
+        (JAN_01_2002_HIGHTIME, 750000, 750000000),
+        (JAN_01_2002_HIGHTIME, 999999, 999999000),
     ],
 )
 def test___datetime_with_microseconds___convert_to_timestamp___is_reversible(
@@ -79,26 +102,26 @@ def test___datetime_with_microseconds___convert_to_timestamp___is_reversible(
     grpc_time.convert_time_to_timestamp(from_dt, to_ts)
     roundtrip_dt = grpc_time.convert_timestamp_to_time(to_ts, tzinfo=timezone.utc)
 
-    assert to_ts.seconds == JAN_01_2022_TIMESTAMP_1970_EPOCH
+    assert to_ts.seconds == JAN_01_2002_TIMESTAMP_1970_EPOCH
     assert to_ts.nanos == nanoseconds
     assert roundtrip_dt.microsecond == microsecond
 
 
 @pytest.mark.parametrize(
     "base_dt, femtosecond, nanoseconds",
     [
-        (JAN_01_2022_HIGHTIME, 0, 0),
-        (JAN_01_2022_HIGHTIME, 1, 0),
+        (JAN_01_2002_HIGHTIME, 0, 0),
+        (JAN_01_2002_HIGHTIME, 1, 0),
         # If femtoseconds get high enough, then it should round up
-        (JAN_01_2022_HIGHTIME, 500000, 1),
-        (JAN_01_2022_HIGHTIME, 999999, 1),
+        (JAN_01_2002_HIGHTIME, 500000, 1),
+        (JAN_01_2002_HIGHTIME, 999999, 1),
         # And of course, whole nanos
-        (JAN_01_2022_HIGHTIME, 1000000, 1),
-        (JAN_01_2022_HIGHTIME, 1000001, 1),
-        (JAN_01_2022_HIGHTIME, 1500000, 2),
-        (JAN_01_2022_HIGHTIME, 1999999, 2),
-        (JAN_01_2022_HIGHTIME, 2000000, 2),
-        (JAN_01_2022_HIGHTIME, 2000001, 2),
+        (JAN_01_2002_HIGHTIME, 1000000, 1),
+        (JAN_01_2002_HIGHTIME, 1000001, 1),
+        (JAN_01_2002_HIGHTIME, 1500000, 2),
+        (JAN_01_2002_HIGHTIME, 1999999, 2),
+        (JAN_01_2002_HIGHTIME, 2000000, 2),
+        (JAN_01_2002_HIGHTIME, 2000001, 2),
     ],
 )
 def test___datetime_with_femtoseconds___convert_to_timestamp___is_reversible(
@@ -110,7 +133,7 @@ def test___datetime_with_femtoseconds___convert_to_timestamp___is_reversible(
     grpc_time.convert_time_to_timestamp(from_dt, to_ts)
     roundtrip_dt = grpc_time.convert_timestamp_to_time(to_ts, tzinfo=timezone.utc)
 
-    assert to_ts.seconds == JAN_01_2022_TIMESTAMP_1970_EPOCH
+    assert to_ts.seconds == JAN_01_2002_TIMESTAMP_1970_EPOCH
     assert to_ts.nanos == nanoseconds
     # we lost femtosecond precision coercing to nanoseconds.
     assert roundtrip_dt.femtosecond == nanoseconds * (10**6)