Merge branch 'main' of https://github.com/robamu-org/py-spacepackets

CHANGELOG.md

@@ -8,12 +8,38 @@ and this project adheres to [Semantic Versioning](http://semver.org/).
 
 ## [unreleased]
 
+## [v0.8.1]
+
+- Named value for fetching global APID
+
+## [v0.8.0]
+
+- Update `PusServices` enumeration
+
+## [v0.7.1]
+
+- Added subservice enumerations for generic PUS Services 1, 3, 5 and 17
+
+## [v0.7.0]
+
+- Improvement for API of PUS TM1 and PUS TM17 base classes
+
+## [v0.6.2]
+
+- Fix in size pre-check of space packet parser `parse_space_packets`
+
+## [v0.6.1]
+
+- Add packet sizes in `__str__` method of PUS TM and TC
+- Some type corrections: Expect `bytes` instead of `bytearray` where applicable
+
 ## [v0.6.0]
 
 ### Added
 
 - Unified Space Data Link Protocol Packet implementations
 
+
 ### Changed
 
 - Assign default print format in TM and TC implementation

spacepackets/__init__.py

@@ -1 +1 @@
-__version__ = "0.6.0"
+__version__ = "0.8.1"

spacepackets/ccsds/spacepacket.py

@@ -252,7 +252,7 @@ def parse_space_packets(
         return tm_list
     # Packet ID detected
     while True:
-        if current_idx + 1 >= len(concatenated_packets):
+        if current_idx + 6 >= len(concatenated_packets):
             break
         current_packet_id = (
             concatenated_packets[current_idx] << 8
@@ -280,8 +280,7 @@ def __handle_packet_id_match(
 ) -> (int, int):
     next_packet_len_field = (
         concatenated_packets[current_idx + 4] << 8
-        | concatenated_packets[current_idx + 5]
-    )
+    ) | concatenated_packets[current_idx + 5]
     total_packet_len = get_total_space_packet_len_from_len_field(next_packet_len_field)
     # Might be part of packet. Put back into analysis queue as whole
     if total_packet_len > len(concatenated_packets):

spacepackets/ccsds/time.py

@@ -109,7 +109,7 @@ def pack(self) -> bytearray:
         return cds_packet
 
     @classmethod
-    def unpack(cls, time_field: bytearray) -> CdsShortTimestamp:
+    def unpack(cls, time_field: bytes) -> CdsShortTimestamp:
         if len(time_field) < cls.TIMESTAMP_SIZE:
             raise ValueError
         # TODO: check ID?

spacepackets/ecss/conf.py

@@ -2,6 +2,9 @@
 from .definitions import DEFAULT_MAX_TC_DATA_SIZE
 
 
+FETCH_GLOBAL_APID = -1
+
+
 class PusVersion(enum.IntEnum):
     # ESA PSS-07-101. Not supported by this package!
     ESA_PUS = 0

spacepackets/ecss/definitions.py

@@ -6,13 +6,14 @@
 
 
 class PusServices(enum.IntEnum):
-    SERVICE_1_VERIFICATION = 1
-    SERVICE_2_RAW_CMD = 2
-    SERVICE_3_HOUSEKEEPING = 3
-    SERVICE_5_EVENT = 5
-    SERVICE_6_MEMORY_MGMT = 6
-    SERVICE_8_FUNC_CMD = 8
-    SERVICE_9_TIME_MGMT = 9
-    SERVICE_17_TEST = 17
-    SERVICE_20_PARAMETER = 20
-    SERVICE_23_FILE_MGMT = 23
+    S1_VERIFICATION = 1
+    S2_RAW_CMD = 2
+    S3_HOUSEKEEPING = 3
+    S5_EVENT = 5
+    S6_MEMORY_MGMT = 6
+    S8_FUNC_CMD = 8
+    S9_TIME_MGMT = 9
+    S11_TC_SCHED = 11
+    S17_TEST = 17
+    S20_PARAMETER = 20
+    S23_FILE_MGMT = 23

spacepackets/ecss/service_17_test.py → spacepackets/ecss/pus_17_test.py

@@ -1,9 +1,14 @@
 from __future__ import annotations
-
+import enum
 from spacepackets.ecss.definitions import PusServices
 from spacepackets.ecss.tm import CdsShortTimestamp, PusVersion, PusTelemetry
 
 
+class Subservices(enum.IntEnum):
+    TC_PING = 1
+    TM_REPLY = 2
+
+
 class Service17TM:
     def __init__(
         self,
@@ -19,7 +24,7 @@ def __init__(
         destination_id: int = 0,
     ):
         self.pus_tm = PusTelemetry(
-            service=PusServices.SERVICE_17_TEST,
+            service=PusServices.S17_TEST,
             subservice=subservice,
             time=time,
             ssc=ssc,

spacepackets/ecss/service_1_verification.py → spacepackets/ecss/pus_1_verification.py

@@ -2,13 +2,26 @@
 """Deserialize PUS Service 1 Verification TM
 """
 from __future__ import annotations
+import enum
 import struct
 
 from spacepackets.ccsds.time import CdsShortTimestamp
+from spacepackets.ecss.definitions import PusServices
 from spacepackets.ecss.tm import PusVersion, PusTelemetry
 from spacepackets.log import get_console_logger
 
 
+class Subservices(enum.IntEnum):
+    TM_ACCEPTANCE_SUCCESS = 1
+    TM_ACCEPTANCE_FAILURE = 2
+    TM_START_SUCCESS = 3
+    TM_START_FAILURE = 4
+    TM_STEP_SUCCESS = 5
+    TM_STEP_FAILURE = 6
+    TM_COMPLETION_SUCCESS = 7
+    TM_COMPLETION_FAILURE = 8
+
+
 class Service1TM:
     """Service 1 TM class representation. Can be used to deserialize raw service 1 packets."""
 
@@ -28,7 +41,7 @@ def __init__(
         destination_id: int = 0,
     ):
         self.pus_tm = PusTelemetry(
-            service=1,
+            service=PusServices.S1_VERIFICATION,
             subservice=subservice,
             time=time,
             ssc=ssc,

spacepackets/ecss/pus_3_hk.py

@@ -0,0 +1,23 @@
+import enum
+
+
+class Subservices(enum.IntEnum):
+    TC_ENABLE_PERIODIC_HK_GEN = 5
+    TC_DISABLE_PERIODIC_HK_GEN = 6
+    TC_ENABLE_PERIODIC_DIAGNOSTICS_GEN = 7
+    TC_DISABLE_PERIODIC_DIAGNOSTICS_GEN = 8
+
+    TM_REPORT_HK_REPORT_STRUCTURES = 9
+    TM_REPORT_DIAG_REPORT_STRUCTURES = 11
+
+    TM_HK_DEFINITIONS_REPORT = 10
+    TM_DIAG_DEFINITION_REPORT = 12
+
+    TM_HK_REPORT = 25
+    TM_DIAGNOSTICS_REPORT = 26
+
+    TC_GENERATE_ONE_PARAMETER_REPORT = 27
+    TC_GENERATE_ONE_DIAGNOSTICS_REPORT = 28
+
+    TC_MODIFY_PARAMETER_REPORT_COLLECTION_INTERVAL = 31
+    TC_MODIFY_DIAGNOSTICS_REPORT_COLLECTION_INTERVAL = 32

spacepackets/ecss/pus_5_event.py

@@ -0,0 +1,17 @@
+import enum
+
+
+class Severity(enum.IntEnum):
+    INFO = 1
+    LOW = 2
+    MEDIUM = 3
+    HIGH = 4
+
+
+class Subservices(enum.IntEnum):
+    TM_INFO_EVENT = Severity.INFO
+    TM_LOW_SEVERITY_EVENT = Severity.LOW
+    TM_MEDIUM_SEVERITY_EVENT = Severity.MEDIUM
+    TM_HIGH_SEVERITY_EVENT = Severity.HIGH
+    TC_ENABLE_EVENT_REPORTING = 5
+    TC_DISABLE_EVENT_REPORTING = 6

spacepackets/ecss/tc.py

@@ -16,6 +16,7 @@
     PusVersion,
     get_pus_tc_version,
     get_max_tc_packet_size,
+    FETCH_GLOBAL_APID,
 )
 
 
@@ -168,7 +169,7 @@ def __init__(
         source_id: int = 0,
         pus_version: PusVersion = PusVersion.GLOBAL_CONFIG,
         ack_flags: int = 0b1111,
-        apid: int = -1,
+        apid: int = FETCH_GLOBAL_APID,
     ):
         """Initiate a PUS telecommand from the given parameters. The raw byte representation
         can then be retrieved with the :py:meth:`pack` function.
@@ -239,8 +240,9 @@ def __repr__(self):
     def __str__(self):
         """Returns string representation of a class instance."""
         return (
-            f"TC[{self.data_field_header.service_type}, "
-            f"{self.data_field_header.service_subtype}] with SSC {self.space_packet_header.ssc}"
+            f"PUS TC[{self.data_field_header.service_type}, "
+            f"{self.data_field_header.service_subtype}], APID {self.apid:#05x}, "
+            f"SSC {self.space_packet_header.ssc}, Size {self.packet_len}"
         )
 
     @property

spacepackets/ecss/tm.py

@@ -11,7 +11,12 @@
     PacketTypes,
 )
 from spacepackets.ccsds.time import CdsShortTimestamp, read_p_field
-from spacepackets.ecss.conf import get_pus_tm_version, PusVersion, get_default_tm_apid
+from spacepackets.ecss.conf import (
+    get_pus_tm_version,
+    PusVersion,
+    get_default_tm_apid,
+    FETCH_GLOBAL_APID,
+)
 
 
 def get_service_from_raw_pus_packet(raw_bytearray: bytearray) -> int:
@@ -43,7 +48,7 @@ def __init__(
         time: CdsShortTimestamp = None,
         ssc: int = 0,
         source_data: bytearray = bytearray([]),
-        apid: int = -1,
+        apid: int = FETCH_GLOBAL_APID,
         message_counter: int = 0,
         space_time_ref: int = 0b0000,
         destination_id: int = 0,
@@ -174,8 +179,8 @@ def unpack(
     def __str__(self):
         return (
             f"PUS TM[{self.secondary_packet_header.service_id},"
-            f"{self.secondary_packet_header.subservice_id}] with message counter "
-            f"{self.secondary_packet_header.message_counter}"
+            f"{self.secondary_packet_header.subservice_id}], APID {self.apid:#05x}, MSG Counter "
+            f"{self.secondary_packet_header.message_counter}, Size {self.packet_len}"
         )
 
     def __repr__(self):
@@ -389,7 +394,7 @@ def pack(self) -> bytearray:
 
     @classmethod
     def unpack(
-        cls, header_start: bytearray, pus_version: PusVersion
+        cls, header_start: bytes, pus_version: PusVersion
     ) -> PusTmSecondaryHeader:
         """Unpack the PUS TM secondary header from the raw packet starting at the header index.
         The user still needs to specify the PUS version because the version field is parsed

tests/test_ecss_pus.py

@@ -26,8 +26,8 @@
     get_service_from_raw_pus_packet,
     PusTmSecondaryHeader,
 )
-from spacepackets.ecss.service_17_test import Service17TM
-from spacepackets.ecss.service_1_verification import Service1TM
+from spacepackets.ecss.pus_17_test import Service17TM
+from spacepackets.ecss.pus_1_verification import Service1TM
 
 
 class TestTelecommand(TestCase):