Re-work ECSS TM time handling

CHANGELOG.md

@@ -8,6 +8,22 @@ and this project adheres to [Semantic Versioning](http://semver.org/).
 
 # [unreleased]
 
+# [v0.24.0] 2024-04-23
+
+## Changed
+
+- ECSS PUS telemetry time handling is now more generic and low level: Constructors expect
+  a simple `bytes` type while unpackers/readers expect the length of the timestamp. Determining
+  or knowing the size of the timestamp is now the task of the user.
+- `CdsShortTimestamp.from_now` renamed to `now`.
+
+## Added
+
+
+- `spacepackets.ecss.tm.PUS_TM_TIMESTAMP_OFFSET` constant which can be used as a look-ahead to
+  determine the timestamp length from a raw PUS TM packet.
+- `spacepackets.ccsds.CCSDS_HEADER_LEN` constant.
+
 # [v0.23.1] 2024-04-22
 
 ## Added

docs/examples.rst

@@ -16,7 +16,7 @@ PUS ping telemetry reply without a timestamp.
     cmd_as_bytes = ping_cmd.pack()
     print(f"Ping telecommand [17,1] (hex): [{cmd_as_bytes.hex(sep=',')}]")
 
-    ping_reply = PusTm(service=17, subservice=2, apid=0x01, time_provider=None)
+    ping_reply = PusTm(service=17, subservice=2, apid=0x01, timestamp=bytes())
     tm_as_bytes = ping_reply.pack()
     print(f"Ping reply [17,2] (hex): [{tm_as_bytes.hex(sep=',')}]")
 

spacepackets/ccsds/spacepacket.py

@@ -11,6 +11,7 @@
 from spacepackets.exceptions import BytesTooShortError
 
 SPACE_PACKET_HEADER_SIZE: Final = 6
+CCSDS_HEADER_LEN: Final = SPACE_PACKET_HEADER_SIZE
 SEQ_FLAG_MASK = 0xC000
 APID_MASK = 0x7FF
 PACKET_ID_MASK = 0x1FFF

spacepackets/ccsds/time/cds.py

@@ -84,11 +84,11 @@ def _calculate_unix_seconds(self):
 
     def _calculate_date_time(self):
         if self._unix_seconds < 0:
-            self._date_time = datetime.datetime(
+            self._datetime = datetime.datetime(
                 1970, 1, 1, tzinfo=datetime.timezone.utc
             ) + datetime.timedelta(seconds=self._unix_seconds)
         else:
-            self._date_time = datetime.datetime.fromtimestamp(
+            self._datetime = datetime.datetime.fromtimestamp(
                 self._unix_seconds, tz=datetime.timezone.utc
             )
 
@@ -108,7 +108,7 @@ def ccsds_days(self) -> int:
     def ms_of_day(self) -> int:
         return self._ms_of_day
 
-    def pack(self) -> bytearray:
+    def pack(self) -> bytes:
         cds_packet = bytearray()
         cds_packet.extend(self.__p_field)
         cds_packet.extend(struct.pack("!H", self._ccsds_days))
@@ -169,12 +169,14 @@ def __repr__(self):
         )
 
     def __str__(self):
-        return f"Date {self._date_time!r} with representation {self!r}"
+        return f"Date {self._datetime!r} with representation {self!r}"
 
-    def __eq__(self, other: CdsShortTimestamp):
-        return (self.ccsds_days == other.ccsds_days) and (
-            self.ms_of_day == other.ms_of_day
-        )
+    def __eq__(self, other: object):
+        if isinstance(other, CdsShortTimestamp):
+            return (self.ccsds_days == other.ccsds_days) and (
+                self.ms_of_day == other.ms_of_day
+            )
+        return False
 
     def __add__(self, timedelta: datetime.timedelta):
         """Allows adding timedelta to the CDS timestamp provider.
@@ -200,23 +202,33 @@ def __add__(self, timedelta: datetime.timedelta):
         return self
 
     @classmethod
-    def from_now(cls) -> CdsShortTimestamp:
+    def now(cls) -> CdsShortTimestamp:
         """Returns a seven byte CDS short timestamp with the current time."""
         return cls.from_date_time(datetime.datetime.now(tz=datetime.timezone.utc))
 
+    @classmethod
+    @deprecation.deprecated(
+        deprecated_in="0.24.0",
+        current_version=get_version(),
+        details="use now instead",
+    )
+    def from_now(cls) -> CdsShortTimestamp:
+        """Returns a seven byte CDS short timestamp with the current time."""
+        return cls.now()
+
     @classmethod
     @deprecation.deprecated(
         deprecated_in="0.14.0rc1",
         current_version=get_version(),
         details="use from_now instead",
     )
     def from_current_time(cls) -> CdsShortTimestamp:
-        return cls.from_now()
+        return cls.now()
 
     @classmethod
-    def from_date_time(cls, dt: datetime.datetime) -> CdsShortTimestamp:
+    def from_datetime(cls, dt: datetime.datetime) -> CdsShortTimestamp:
         instance = cls.empty(False)
-        instance._date_time = dt
+        instance._datetime = dt
         instance._unix_seconds = dt.timestamp()
         full_unix_secs = int(math.floor(instance._unix_seconds))
         subsec_millis = int((instance._unix_seconds - full_unix_secs) * 1000)
@@ -226,6 +238,15 @@ def from_date_time(cls, dt: datetime.datetime) -> CdsShortTimestamp:
         instance._ccsds_days = convert_unix_days_to_ccsds_days(unix_days)
         return instance
 
+    @classmethod
+    @deprecation.deprecated(
+        deprecated_in="0.24.0",
+        current_version=get_version(),
+        details="use from_datetime instead",
+    )
+    def from_date_time(cls, dt: datetime.datetime) -> CdsShortTimestamp:
+        return cls.from_datetime(dt)
+
     @staticmethod
     def ms_of_today(seconds_since_epoch: Optional[float] = None):
         if seconds_since_epoch is None:
@@ -238,5 +259,13 @@ def ms_of_today(seconds_since_epoch: Optional[float] = None):
     def as_unix_seconds(self) -> float:
         return self._unix_seconds
 
+    def as_datetime(self) -> datetime.datetime:
+        return self._datetime
+
+    @deprecation.deprecated(
+        deprecated_in="0.24.0",
+        current_version=get_version(),
+        details="use as_datetime instead",
+    )
     def as_date_time(self) -> datetime.datetime:
-        return self._date_time
+        return self.as_datetime()

spacepackets/ccsds/time/common.py

@@ -72,7 +72,7 @@ def len(self) -> int:
         return self.len_packed
 
     @abstractmethod
-    def pack(self) -> bytearray:
+    def pack(self) -> bytes:
         pass
 
     @abstractmethod

spacepackets/ecss/pus_17_test.py

@@ -1,10 +1,8 @@
 from __future__ import annotations
 import enum
-from typing import Optional
 
 from spacepackets import SpacePacketHeader
 from spacepackets.ccsds.spacepacket import PacketId, PacketSeqCtrl
-from spacepackets.ccsds.time import CcsdsTimeProvider
 from spacepackets.ecss.conf import FETCH_GLOBAL_APID
 from spacepackets.ecss.defs import PusService
 from spacepackets.ecss.tm import PusTm, AbstractPusTm
@@ -19,7 +17,7 @@ class Service17Tm(AbstractPusTm):
     def __init__(
         self,
         subservice: int,
-        time_provider: Optional[CcsdsTimeProvider],
+        timestamp: bytes,
         ssc: int = 0,
         source_data: bytes = bytes(),
         apid: int = FETCH_GLOBAL_APID,
@@ -30,7 +28,7 @@ def __init__(
         self.pus_tm = PusTm(
             service=PusService.S17_TEST,
             subservice=subservice,
-            time_provider=time_provider,
+            timestamp=timestamp,
             seq_count=ssc,
             source_data=source_data,
             apid=apid,
@@ -60,8 +58,8 @@ def service(self) -> int:
         return self.pus_tm.service
 
     @property
-    def time_provider(self) -> Optional[CcsdsTimeProvider]:
-        return self.pus_tm.time_provider
+    def timestamp(self) -> bytes:
+        return self.pus_tm.timestamp
 
     @property
     def subservice(self) -> int:
@@ -75,19 +73,17 @@ def pack(self) -> bytearray:
         return self.pus_tm.pack()
 
     @classmethod
-    def __empty(cls, time_provider: Optional[CcsdsTimeProvider]) -> Service17Tm:
-        return cls(subservice=0, time_provider=time_provider)
+    def __empty(cls) -> Service17Tm:
+        return cls(subservice=0, timestamp=bytes())
 
     @classmethod
-    def unpack(
-        cls, data: bytes, time_reader: Optional[CcsdsTimeProvider]
-    ) -> Service17Tm:
+    def unpack(cls, data: bytes, timestamp_len: int) -> Service17Tm:
         """
 
         :raises BytesTooShortError: Passed bytestream too short.
         :raises ValueError: Unsupported PUS version.
         :raises InvalidTmCrc16: Invalid CRC16.
         """
-        service_17_tm = cls.__empty(time_provider=time_reader)
-        service_17_tm.pus_tm = PusTm.unpack(data=data, time_reader=time_reader)
+        service_17_tm = cls.__empty()
+        service_17_tm.pus_tm = PusTm.unpack(data=data, timestamp_len=timestamp_len)
         return service_17_tm