#96 format for EosPayload driver health reporting

EosLib/format/__init__.py

@@ -1,3 +1,4 @@
 from EosLib.format.formats import telemetry_data, position, empty_format, cutdown, ping_format, valve, e_field, \
                                   science_data
+from EosLib.format.formats.health import driver_health_report
 from EosLib.format.definitions import Type as Type

EosLib/format/definitions.py

@@ -17,4 +17,5 @@ class Type(IntEnum):
     VALVE = 11
     E_FIELD = 12
     SCIENCE_DATA = 13
+    DRIVER_HEALTH_REPORT = 14
     ERROR = 255

EosLib/format/formats/health/driver_health_report.py

@@ -0,0 +1,75 @@
+from dataclasses import dataclass
+from enum import IntEnum, unique
+from typing_extensions import Self
+import struct
+
+from EosLib.format.base_format import BaseFormat
+from EosLib.format.definitions import Type
+
+@unique
+class ThreadStatus(IntEnum):
+    NONE = 0
+    INVALID = 1
+    REGISTERED = 2
+    ALIVE = 3
+    DEAD = 4
+
+
+@dataclass
+class DriverHealthReport(BaseFormat):
+
+    is_healthy:              bool       # true if healthy                                                (bool)
+    custom_state_bitvector:  int        # may be used by the driver for any custom purpose               (uchar)
+    num_threads:             int        # of threads, including main and mqtt (uchar)
+    thread_statuses:         list[int]  # list of ThreadStatuses of size num_threads - 1, starting with  (uchar[])
+                                        # mqtt and then registered threads in order of registration
+
+    @staticmethod
+    def _i_promise_all_abstract_methods_are_implemented() -> bool:
+        return True
+
+    @staticmethod
+    def get_format_type() -> Type:
+        return Type.DRIVER_HEALTH_REPORT
+
+    def get_validity(self) -> bool:
+        thread_status_bounds_all_good = True
+        for thread_status in self.thread_statuses:
+            try:
+                ThreadStatus(thread_status)
+            except ValueError:
+                thread_status_bounds_all_good = False
+                break
+
+        return (
+            0 <= self.custom_state_bitvector <= 255
+            and 2 <= self.num_threads <= 63  # there is some maximum much less than 255, so just giving a guess
+            and len(self.thread_statuses) == self.num_threads - 1
+            and thread_status_bounds_all_good
+        )
+
+    def encode(self) -> bytes:
+        format_string = "?BB" + "B"*(self.num_threads - 1)
+        return struct.pack(
+            format_string,
+            self.is_healthy,
+            self.custom_state_bitvector,
+            self.num_threads,
+            *self.thread_statuses,
+        )
+
+    @classmethod
+    def decode(cls, data: bytes) -> Self:
+        # start with the fields that will always be present (is_healthy, custom_bitvector, num_threads)
+        constant_part_format_string = "?BB"
+        offset = struct.calcsize(constant_part_format_string)
+        is_healthy, custom_state_bitvector, num_threads = struct.unpack(constant_part_format_string, data[:offset])
+
+        # now do the thread_status_array which is variable-length (depends on num_threads)
+        if not (2 <= num_threads <= 63):
+            raise ValueError(f"Failed to decode: num_threads must between 2 and 63, got {num_threads}")
+        thread_statuses_format_string = "B"*(num_threads - 1)
+        thread_statuses: list[int] = list(struct.unpack(thread_statuses_format_string, data[offset:]))
+        return DriverHealthReport(is_healthy, custom_state_bitvector, num_threads, thread_statuses)
+
+

setup.py

@@ -4,7 +4,7 @@
 from setuptools import find_packages
 
 setup(name='EosLib',
-      version='4.4.0',
+      version='4.5.0',
       description='Library of shared code between EosPayload and EosGround',
       author='Lightning From The Edge of Space',
       author_email='[email protected]',

tests/format/formats/format_test.py

@@ -44,6 +44,8 @@ def test_not_eq(self):
                 new_data_list[i] += 1
             elif isinstance(new_data_list[i], datetime.datetime):
                 new_data_list[i] += datetime.timedelta(1)
+            elif isinstance(new_data_list[i], list):
+                continue
 
             data_2 = self.get_format_class()(*new_data_list)
 
@@ -54,3 +56,6 @@ def test_encode_decode_bytes(self):
         base_position_bytes = base_format.encode()
         new_format = decode_factory.decode(self.get_good_format().get_format_type(), base_position_bytes)
         assert base_format == new_format
+
+    def test_get_validity_happy_path(self):
+        assert self.get_good_format().get_validity()

tests/format/formats/health/test_driver_health_report.py

@@ -0,0 +1,43 @@
+from EosLib.format.formats.health.driver_health_report import DriverHealthReport, ThreadStatus
+from tests.format.formats.format_test import CheckFormat
+
+
+class TestCutDown(CheckFormat):
+
+    def get_format_class(self):
+        return DriverHealthReport
+
+    def get_good_format_params(self):
+        return [
+            True,
+            0xB5,
+            3,
+            [ThreadStatus.ALIVE, ThreadStatus.DEAD]
+        ]
+
+    def test_get_validity(self):
+        # custom bitvector bounds
+        assert not DriverHealthReport(True, -1, 2, [ThreadStatus.NONE]).get_validity()
+        assert DriverHealthReport(True, 0, 2, [ThreadStatus.NONE]).get_validity()
+        assert DriverHealthReport(True, 255, 2, [ThreadStatus.NONE]).get_validity()
+        assert not DriverHealthReport(True, 256, 2, [ThreadStatus.NONE]).get_validity()
+
+        # num threads bounds
+        assert not DriverHealthReport(False, 0, 1, []).get_validity()
+        assert DriverHealthReport(False, 0, 2, [ThreadStatus.NONE]).get_validity()
+        assert DriverHealthReport(False, 0, 63, [ThreadStatus.NONE]*62).get_validity()
+        assert not DriverHealthReport(False, 0, 64, []).get_validity()
+
+        # thread_status eum values
+        thread_status_values = [elem.value for elem in ThreadStatus]
+        min_thread_status = min(thread_status_values)
+        max_thread_status = max(thread_status_values)
+        assert not DriverHealthReport(True, 0, 2, [min_thread_status - 1]).get_validity()
+        assert DriverHealthReport(True, 0, 2, [min_thread_status]).get_validity()
+        assert DriverHealthReport(True, 0, 2, [max_thread_status]).get_validity()
+        assert not DriverHealthReport(True, 0, 2, [max_thread_status + 1]).get_validity()
+
+        # thread_statuses correct length
+        assert not DriverHealthReport(False, 0, 2, []).get_validity()
+        assert DriverHealthReport(False, 0, 2, [ThreadStatus.NONE]).get_validity()
+        assert not DriverHealthReport(False, 0, 2, [ThreadStatus.NONE]*2).get_validity()