Fix ntp/osc timestamp/timetag conversion and TimedMessage time.

Author: smrg-lm

pythonosc/osc_message.py

@@ -53,7 +53,7 @@ def _parse_datagram(self) -> None:
                 elif param == "m":  # MIDI.
                     val, index = osc_types.get_midi(self._dgram, index)
                 elif param == "t":  # osc time tag:
-                    val, index = osc_types.get_ttag(self._dgram, index)
+                    val, index = osc_types.get_timetag(self._dgram, index)
                 elif param == "T":  # True.
                     val = True
                 elif param == "F":  # False.

pythonosc/osc_packet.py

@@ -3,7 +3,6 @@
 It lets you access easily to OscMessage and OscBundle instances in the packet.
 """
 
-import calendar
 import collections
 import time
 
@@ -22,11 +21,11 @@
     field_names=('time', 'message'))
 
 
-def _timed_msg_of_bundle(bundle: osc_bundle.OscBundle, now: int) -> List[TimedMessage]:
+def _timed_msg_of_bundle(bundle: osc_bundle.OscBundle, now: float) -> List[TimedMessage]:
     """Returns messages contained in nested bundles as a list of TimedMessage."""
     msgs = []
     for content in bundle:
-        if type(content) == osc_message.OscMessage:
+        if type(content) is osc_message.OscMessage:
             if (bundle.timestamp == osc_types.IMMEDIATELY or bundle.timestamp < now):
                 msgs.append(TimedMessage(now, content))
             else:
@@ -56,7 +55,7 @@ def __init__(self, dgram: bytes) -> None:
         Raises:
           - ParseError if the datagram could not be parsed.
         """
-        now = calendar.timegm(time.gmtime())
+        now = time.time()
         try:
             if osc_bundle.OscBundle.dgram_is_bundle(dgram):
                 self._messages = sorted(

pythonosc/parsing/ntp.py

@@ -6,12 +6,14 @@
 
 from typing import Union
 
-# conversion factor for fractional seconds (maximum value of fractional part)
-FRACTIONAL_CONVERSION = 2 ** 32
 
 # 63 zero bits followed by a one in the least signifigant bit is a special
 # case meaning "immediately."
-IMMEDIATELY = struct.pack('>q', 1)
+IMMEDIATELY = struct.pack('>Q', 1)
+
+# timetag * (1 / 2 ** 32) == l32bits + (r32bits / 1 ** 32)
+_NTP_TIMESTAMP_TO_SECONDS = 1. / 2. ** 32.
+_SECONDS_TO_NTP_TIMESTAMP = 2. ** 32.
 
 # From NTP lib.
 _SYSTEM_EPOCH = datetime.date(*time.gmtime(0)[0:3])
@@ -24,27 +26,33 @@ class NtpError(Exception):
   """Base class for ntp module errors."""
 
 
-def ntp_to_system_time(date: Union[int, float]) -> Union[int, float]:
-    """Convert a NTP time to system time.
-
-    System time is reprensented by seconds since the epoch in UTC.
+def ntp_to_system_time(timestamp: bytes) -> float:
+    """Convert a NTP timestamp to system time in seconds.
     """
-    return date - _NTP_DELTA
+    try:
+        timestamp = struct.unpack('>Q', timestamp)
+    except Exception as e:
+        raise NtpError(e)
+    return timestamp * _NTP_TIMESTAMP_TO_SECONDS - _NTP_DELTA
 
-def system_time_to_ntp(date: Union[int, float]) -> bytes:
-    """Convert a system time to NTP time.
 
-    System time is reprensented by seconds since the epoch in UTC.
+def system_time_to_ntp(seconds: float) -> bytes:
+    """Convert a system time in seconds to NTP timestamp.
     """
     try:
-      num_secs = int(date)
+      seconds = seconds + _NTP_DELTA
     except ValueError as e:
       raise NtpError(e)
-    
-    num_secs_ntp = num_secs + _NTP_DELTA
-    
-    sec_frac = float(date - num_secs)
-
-    picos = int(sec_frac * FRACTIONAL_CONVERSION)
-  
-    return struct.pack('>I', int(num_secs_ntp)) + struct.pack('>I', picos)
+    return struct.pack('>Q', int(seconds * _SECONDS_TO_NTP_TIMESTAMP))
+
+
+def ntp_time_to_system_epoch(seconds: float) -> float:
+    """Convert a NTP time in seconds to system time in seconds.
+    """
+    return seconds - _NTP_DELTA
+
+
+def system_time_to_ntp_epoch(seconds: float) -> float:
+    """Convert a system time in seconds to NTP time in seconds.
+    """
+    return seconds + _NTP_DELTA

pythonosc/parsing/osc_types.py

@@ -1,6 +1,5 @@
 """Functions to get OSC types from datagrams and vice versa"""
 
-import decimal
 import struct
 
 from pythonosc.parsing import ntp
@@ -22,9 +21,10 @@ class BuildError(Exception):
 
 # Datagram length in bytes for types that have a fixed size.
 _INT_DGRAM_LEN = 4
+_UINT64_DGRAM_LEN = 8
 _FLOAT_DGRAM_LEN = 4
 _DOUBLE_DGRAM_LEN = 8
-_DATE_DGRAM_LEN = _INT_DGRAM_LEN * 2
+_TIMETAG_DGRAM_LEN = 8
 # Strings and blob dgram length is always a multiple of 4 bytes.
 _STRING_DGRAM_PAD = 4
 _BLOB_DGRAM_PAD = 4
@@ -126,7 +126,31 @@ def get_int(dgram: bytes, start_index: int) -> Tuple[int, int]:
         raise ParseError('Could not parse datagram %s' % e)
 
 
-def get_ttag(dgram: bytes, start_index: int) -> Tuple[datetime, int]:
+def get_uint64(dgram: bytes, start_index: int) -> Tuple[int, int]:
+    """Get a 64-bit big-endian unsigned integer from the datagram.
+
+    Args:
+      dgram: A datagram packet.
+      start_index: An index where the integer starts in the datagram.
+
+    Returns:
+      A tuple containing the integer and the new end index.
+
+    Raises:
+      ParseError if the datagram could not be parsed.
+    """
+    try:
+        if len(dgram[start_index:]) < _UINT64_DGRAM_LEN:
+            raise ParseError('Datagram is too short')
+        return (
+            struct.unpack('>Q',
+                          dgram[start_index:start_index + _UINT64_DGRAM_LEN])[0],
+            start_index + _UINT64_DGRAM_LEN)
+    except (struct.error, TypeError) as e:
+        raise ParseError('Could not parse datagram %s' % e)
+
+
+def get_timetag(dgram: bytes, start_index: int) -> Tuple[datetime, int]:
     """Get a 64-bit OSC time tag from the datagram.
 
     Args:
@@ -140,29 +164,22 @@ def get_ttag(dgram: bytes, start_index: int) -> Tuple[datetime, int]:
     Raises:
       ParseError if the datagram could not be parsed.
     """
-
-    _TTAG_DGRAM_LEN = 8
-
     try:
-        if len(dgram[start_index:]) < _TTAG_DGRAM_LEN:
+        if len(dgram[start_index:]) < _TIMETAG_DGRAM_LEN:
             raise ParseError('Datagram is too short')
 
-        seconds, idx = get_int(dgram, start_index)
-        second_decimals, _ = get_int(dgram, idx)
-
-        if seconds < 0:
-            seconds += ntp.FRACTIONAL_CONVERSION
-
-        if second_decimals < 0:
-            second_decimals += ntp.FRACTIONAL_CONVERSION
+        timetag, _ = get_uint64(dgram, start_index)
+        seconds_float = timetag * ntp._NTP_TIMESTAMP_TO_SECONDS
+        seconds = int(seconds_float)
+        fraction = seconds_float - seconds
 
         hours, seconds = seconds // 3600, seconds % 3600
         minutes, seconds = seconds // 60, seconds % 60
 
-        utc = datetime.combine(ntp._NTP_EPOCH, datetime.min.time()) + timedelta(hours=hours, minutes=minutes,
-                                                                                seconds=seconds)
+        utc = (datetime.combine(ntp._NTP_EPOCH, datetime.min.time()) +
+               timedelta(hours=hours, minutes=minutes, seconds=seconds))
 
-        return (utc, second_decimals), start_index + _TTAG_DGRAM_LEN
+        return (utc, fraction), start_index + _TIMETAG_DGRAM_LEN
     except (struct.error, TypeError) as e:
         raise ParseError('Could not parse datagram %s' % e)
 
@@ -284,7 +301,7 @@ def write_blob(val: bytes) -> bytes:
     return dgram
 
 
-def get_date(dgram: bytes, start_index: int) -> Tuple[Union[int, float], int]:
+def get_date(dgram: bytes, start_index: int) -> Tuple[float, int]:
     """Get a 64-bit big-endian fixed-point time tag as a date from the datagram.
 
     According to the specifications, a date is represented as is:
@@ -304,16 +321,13 @@ def get_date(dgram: bytes, start_index: int) -> Tuple[Union[int, float], int]:
       ParseError if the datagram could not be parsed.
     """
     # Check for the special case first.
-    if dgram[start_index:start_index + _DATE_DGRAM_LEN] == ntp.IMMEDIATELY:
-        return IMMEDIATELY, start_index + _DATE_DGRAM_LEN
-    if len(dgram[start_index:]) < _DATE_DGRAM_LEN:
+    if dgram[start_index:start_index + _TIMETAG_DGRAM_LEN] == ntp.IMMEDIATELY:
+        return IMMEDIATELY, start_index + _TIMETAG_DGRAM_LEN
+    if len(dgram[start_index:]) < _TIMETAG_DGRAM_LEN:
         raise ParseError('Datagram is too short')
-    num_secs, start_index = get_int(dgram, start_index)
-    fraction, start_index = get_int(dgram, start_index)
-    # Sum seconds and fraction of second:
-    system_time = num_secs + (fraction / ntp.FRACTIONAL_CONVERSION)
-
-    return ntp.ntp_to_system_time(system_time), start_index
+    timetag, start_index = get_uint64(dgram, start_index)
+    seconds = timetag * ntp._NTP_TIMESTAMP_TO_SECONDS
+    return ntp.ntp_time_to_system_epoch(seconds), start_index
 
 
 def write_date(system_time: Union[int, float]) -> bytes: