A support for parsing compressed data products in fprime-dp

src/fprime_gds/common/dp/decoder.py

@@ -19,6 +19,9 @@
 from pathlib import Path
 from typing import Dict, List, Any, Optional
 import dataclasses
+import re
+from io import BytesIO
+import zlib
 
 from fprime_gds.common.dp.common import (
     ChecksumConfig,
@@ -85,7 +88,7 @@ class DataProductDecoder:
         - both these assumptions can be resolved by loading dictionaries (see executables/data_products.py)
     """
 
-    def __init__(self, dictionaries: Dictionaries, binary_file_path: str, output_json_path: Optional[str] = None):
+    def __init__(self, dictionaries: Dictionaries, binary_file_path: str, output_json_path: Optional[str] = None, disable_decompression: bool = False):
         """Initialize the decoder.
 
         Args:
@@ -96,6 +99,7 @@ def __init__(self, dictionaries: Dictionaries, binary_file_path: str, output_jso
         """
         self.dictionaries = dictionaries
         self.binary_file_path = binary_file_path
+        self.disable_decompression = disable_decompression
         if output_json_path is None:
             # Generate default output path if not provided as same path with .json extension
             self.output_json_path = str(Path(binary_file_path).with_suffix('.json'))
@@ -145,7 +149,7 @@ def decode_record(self, file_handle, record_id: int) -> Dict[str, Any]:
         Raises:
             RecordNotFoundError: If record ID not found
         """
-        
+
         # Query ConfigManager for record definition
         record_template: DpRecordTemplate = self.dictionaries.dp_record_id.get(record_id)
 
@@ -197,6 +201,52 @@ def read_element(element_type):
 
         return record
 
+    def decode_records(self, r_io, data_size) -> List[Any]:
+        records = list()
+        position_at_start = r_io.tell()
+        while (r_io.tell() - position_at_start) < data_size:
+            # Read record ID
+            record_id_bin = r_io.read(ConfigManager().get_type("FwDpIdType").getSize())
+            if len(record_id_bin) == 0:
+                # No bytes read. Off nominal behavior
+                raise DataProductError(f"End of file reached while processing data product")
+
+            record_id_obj = ConfigManager().get_type("FwDpIdType")()
+            record_id_obj.deserialize(record_id_bin, 0)
+            record_id = record_id_obj.val
+
+            # decode the record
+            record = self.decode_record(r_io, record_id)
+            records.append(record)
+
+        return records
+
+    def is_compression_record(self, record):
+        return re.search("dpCompressProc.CompressionRecord$", record["Record"]["record_name"]) is not None
+
+    def decompress_records(self, records):
+        uncomp_bytes = bytearray()
+        CompressionMetadata = ConfigManager().get_type("Svc.CompressionMetadata")
+
+        for record in records:
+            if not self.is_compression_record(record):
+                # All records must be compressed, otherwise bail
+                return None
+
+            record_io = BytesIO(bytes(record["Data"]["values"]))
+
+            record_meta = CompressionMetadata()
+            record_meta_data = record_io.read(record_meta.getMaxSize())
+            record_meta.deserialize(record_meta_data, 0)
+            if record_meta.val['algorithm'] == 'UNCOMPRESSED':
+                uncomp_bytes.extend(record_io.read())
+            elif record_meta.val['algorithm'] == 'ZLIB_DEFLATE':
+                uncomp_bytes.extend(zlib.decompress(record_io.read()))
+            else:
+                raise DataProductError(f"Compression algorithm {record_meta.val['algorithm']} unsupported")
+
+        return uncomp_bytes
+
     def decode(self) -> List[Dict[str, Any]]:
         """decode the entire data product file.
 
@@ -223,16 +273,7 @@ def decode(self) -> List[Dict[str, Any]]:
             #####################
             data_size = header_json['DataSize']["value"]
             position_at_start = f.tell()
-            while (f.tell() - position_at_start) < data_size:
-                # Read record ID
-                record_id_bin = f.read(ConfigManager().get_type("FwDpIdType").getSize())
-                record_id_obj = ConfigManager().get_type("FwDpIdType")()
-                record_id_obj.deserialize(record_id_bin, 0)
-                record_id = record_id_obj.val
-
-                # decode the record
-                record = self.decode_record(f, record_id)
-                results["Records"].append(record)
+            results["Records"] = self.decode_records(f, data_size)
 
             #####################
             # Validate checksum #
@@ -250,6 +291,16 @@ def decode(self) -> List[Dict[str, Any]]:
             if computed_crc != dp_crc.val:
                 raise CRCError("Data", dp_crc.val, computed_crc)
 
+        if not self.disable_decompression and len(results["Records"]) > 0 and self.is_compression_record(results["Records"][0]):
+
+            # Compressed records. Decompress and re-process
+            uncomp_bytes = self.decompress_records(results["Records"])
+            if uncomp_bytes is not None:
+                uncomp_io = BytesIO(uncomp_bytes)
+                uncomp_records = self.decode_records(uncomp_io, len(uncomp_bytes))
+                if uncomp_records is not None:
+                    results["Records"] = uncomp_records
+
         return results
 
     def process(self):

src/fprime_gds/executables/data_products.py

@@ -13,6 +13,7 @@ def main():
     decode_parser.add_argument("-b", "--bin-file", required=True, help="Path to input data product binary file (.fdp)")
     decode_parser.add_argument("-d", "--dictionary", required=True, help="Path to F Prime JSON Dictionary")
     decode_parser.add_argument("-o", "--output", required=False, help="Path to output JSON file (defaults to <binFilename>.json)")
+    decode_parser.add_argument("-z", "--disable-decompression", action='store_true', help="Disable automatic decompression of data products")
 
     validate_parser = subcommands_parser.add_parser('validate', help='Validate a data product')
     validate_parser.add_argument("-b", "--bin-file", required=True, help="Path to input data product binary file (.fdp)")
@@ -29,7 +30,7 @@ def main():
 
     if args.command == "decode":
         assert args.dictionaries is not None, "Dictionaries must be loaded"
-        DataProductDecoder(args.dictionaries, args.bin_file, args.output).process()
+        DataProductDecoder(args.dictionaries, args.bin_file, args.output, args.disable_decompression).process()
 
     elif args.command == "validate":
         success = DataProductValidator(

test/fprime_gds/common/dp/test_decoder.py

@@ -23,6 +23,7 @@
 # Path to test data directory
 TEST_DATA_DIR = Path(__file__).parent / "test_dp_data"
 DICTIONARY_PATH = TEST_DATA_DIR / "dictionary.json"
+DICTIONARY_REF_PATH = TEST_DATA_DIR / "dictionary_ref.json"
 
 
 @pytest.fixture
@@ -35,6 +36,16 @@ def load_dictionary():
     yield dictionaries
     globals_cleanup()
 
+@pytest.fixture
+def load_dictionary_ref():
+    """Fixture to load the test dictionary into ConfigManager before tests.
+    Also uses the globals_cleanup utility to reset global state and not interfere 
+    with other tests."""
+    globals_cleanup()
+    dictionaries = Dictionaries.load_dictionaries_into_config(str(DICTIONARY_REF_PATH))
+    yield dictionaries
+    globals_cleanup()
+
 
 class TestDataProductDecoderBasicTypes:
     """Test decoding of basic primitive data types."""
@@ -468,6 +479,50 @@ def test_decode_and_verify_json_structure(self, load_dictionary, tmp_path):
             assert "Record" in record
             assert "Data" in record
 
+class TestDataProductDecoderCompressedProducts:
+    """Test decoding of compressed products."""
+
+    def test_decode_compressed(self, load_dictionary_ref, tmp_path):
+        """Test that compressed records can be decoded"""
+        decoder = DataProductDecoder(
+            load_dictionary_ref,
+            str(TEST_DATA_DIR / "DpDemoCompressed.fdp")
+        )
+        result = decoder.decode()
+
+        # Should contain 14 records from DpDemo
+        records = result["Records"]
+        assert len(records) == 14
+
+    def test_decode_compressed_raw(self, load_dictionary_ref, tmp_path):
+        """Test that RAW compressed records can be generated"""
+        decoder = DataProductDecoder(
+            load_dictionary_ref,
+            str(TEST_DATA_DIR / "DpDemoCompressed.fdp"),
+            disable_decompression=True
+        )
+        result = decoder.decode()
+
+        # Product should contain one compressed record segment
+        records = result["Records"]
+        assert len(records) == 1
+        assert result["Records"][0]["Record"]["record_name"] == "DpCompression.dpCompressProc.CompressionRecord"
+
+    def test_decode_compressed_compare(self, load_dictionary_ref, tmp_path):
+        """Test that compressed records can be identical to uncompressed records"""
+        decoder_compressed = DataProductDecoder(
+            load_dictionary_ref,
+            str(TEST_DATA_DIR / "DpDemoCompressed.fdp")
+        )
+        result_compressed = decoder_compressed.decode()
+
+        decoder_uncompressed = DataProductDecoder(
+            load_dictionary_ref,
+            str(TEST_DATA_DIR / "DpDemoUncompressed.fdp")
+        )
+        result_uncompressed = decoder_uncompressed.decode()
+
+        assert result_uncompressed["Records"] == result_compressed["Records"]
 
 if __name__ == "__main__":
     pytest.main([__file__, "-v"])

test/fprime_gds/common/dp/test_dp_data/dictionary.json

@@ -1423,6 +1423,43 @@
         "tSub" : 0
       },
       "annotation" : "Data structure representing a data product."
+    },
+    {
+      "kind" : "struct",
+      "qualifiedName" : "Svc.CompressionMetadata",
+      "members" : {
+        "algorithm" : {
+          "type" : {
+            "name" : "Svc.CompressionAlgorithm",
+            "kind" : "qualifiedIdentifier"
+          },
+          "index" : 0
+        }
+      },
+      "default" : {
+        "algorithm" : "Svc.CompressionAlgorithm.UNCOMPRESSED"
+      }
+    },
+    {
+      "kind" : "enum",
+      "qualifiedName" : "Svc.CompressionAlgorithm",
+      "representationType" : {
+        "name" : "U8",
+        "kind" : "integer",
+        "size" : 8,
+        "signed" : false
+      },
+      "enumeratedConstants" : [
+        {
+          "name" : "UNCOMPRESSED",
+          "value" : 0
+        },
+        {
+          "name" : "ZLIB_DEFLATE",
+          "value" : 1
+        }
+      ],
+      "default" : "Svc.CompressionAlgorithm.UNCOMPRESSED"
     }
   ],
   "commands" : [
@@ -10103,6 +10140,18 @@
       "array" : false,
       "id" : 5389,
       "annotation" : "Record 14"
+    },
+    {
+      "name" : "DpCompression.dpCompressProc.CompressionRecord",
+      "type" : {
+        "name" : "U8",
+        "kind" : "integer",
+        "size" : 8,
+        "signed" : false
+      },
+      "array" : true,
+      "id" : 68157440,
+      "annotation" : "Record ID used to mark compressed records in a data product"
     }
   ],
   "containers" : [