hdf5_policy.py: Update meta-data

Author: christoph2

.bumpversion.cfg

@@ -1,5 +1,5 @@
 [bumpversion]
-current_version = 0.25.4
+current_version = 0.25.5
 commit = True
 tag = False
 

pyproject.toml

@@ -38,7 +38,7 @@ dynamic = ["license", "readme", "authors", "requires-python", "description", "cl
 [tool.poetry]
 authors = ["Christoph Schueler <[email protected]>"]
 name = "pyxcp"
-version = "0.25.4"
+version = "0.25.5"
 readme = "docs/README.rst"
 description = "Universal Calibration Protocol for Python"
 keywords = ["automotive", "ecu", "xcp", "asam", "autosar"]

pyxcp/__init__.py

@@ -17,4 +17,4 @@
 
 # if you update this manually, do not forget to update
 # .bumpversion.cfg and pyproject.toml.
-__version__ = "0.25.4"
+__version__ = "0.25.5"

pyxcp/transport/hdf5_policy.py

@@ -2,7 +2,7 @@
 
 import datetime
 from pathlib import Path
-from typing import List, Dict, Any
+from typing import List, Dict
 
 import h5py
 import numpy as np
@@ -26,6 +26,21 @@
     "BF16": np.float16,
 }
 
+MAP_TO_ASAM_HO = {
+    "U8": "A_UINT8",
+    "I8": "A_INT8",
+    "U16": "A_UINT16",
+    "I16": "A_INT16",
+    "U32": "A_UINT32",
+    "I32": "A_INT32",
+    "U64": "A_UINT64",
+    "I64": "A_INT64",
+    "F32": "A_FLOAT32",
+    "F64": "A_FLOAT64",
+    "F16": "A_FLOAT16",
+    "BF16": "A_FLOAT16",
+}
+
 
 class BufferedDataset:
     def __init__(self, dataset: h5py.Dataset):
@@ -40,7 +55,7 @@ def add_sample(self, sample: int | float):
     def flush(self):
         batch = np.array(self.buffer)
         self.dataset.resize((self.dataset.shape[0] + len(batch),))
-        self.dataset[-len(batch):] = batch
+        self.dataset[-len(batch) :] = batch
         self.buffer.clear()
         self.dataset.flush()
 
@@ -50,10 +65,10 @@ def __len__(self):
 
 class DatasetGroup:
     def __init__(
-            self,
-            ts0_ds: BufferedDataset,
-            ts1_ds: BufferedDataset,
-            datasets: List[BufferedDataset],
+        self,
+        ts0_ds: BufferedDataset,
+        ts1_ds: BufferedDataset,
+        datasets: List[BufferedDataset],
     ):
         self.ts0_ds = ts0_ds
         self.ts1_ds = ts1_ds
@@ -73,13 +88,16 @@ def finalize(self):
 
 
 def create_timestamp_column(hdf_file: h5py.File, group_name: str, num: int) -> h5py.Dataset:
-    return hdf_file.create_dataset(
+    result = hdf_file.create_dataset(
         f"/{group_name}/timestamp{num}",
         shape=(0,),
         maxshape=(None,),
         dtype=np.uint64,
         chunks=True,
     )
+    result.attrs["asam_data_type"] = "A_UINT64"
+    result.attrs["resolution"] = ("1 nanosecond",)
+    return result
 
 
 class Hdf5OnlinePolicy(DaqOnlinePolicy):
@@ -106,20 +124,29 @@ def initialize(self):
         for num, daq_list in enumerate(self.daq_lists):
             if daq_list.stim:
                 continue
-            self.hdf.create_group(daq_list.name)
+            grp = self.hdf.create_group(daq_list.name)
+            grp.attrs["event_num"] = daq_list.event_num
+            grp.attrs["enable_timestamps"] = daq_list.enable_timestamps
+            grp.attrs["prescaler"] = daq_list.prescaler
+            grp.attrs["priority"] = daq_list.priority
+            grp.attrs["direction"] = "STIM" if daq_list.stim else "DAQ"
             ts0 = BufferedDataset(create_timestamp_column(self.hdf, daq_list.name, 0))
             ts1 = BufferedDataset(create_timestamp_column(self.hdf, daq_list.name, 1))
             meas_map = {m.name: m for m in self.daq_lists[num].measurements}
             dsets = []
             for name, _ in daq_list.headers:
                 meas = meas_map[name]
                 dataset = self.hdf.create_dataset(
-                    f"/{daq_list.name}/{meas.name}",
+                    f"/{daq_list.name}/{meas.name}/raw",
                     shape=(0,),
                     maxshape=(None,),
                     dtype=MAP_TO_NP[meas.data_type],
                     chunks=(1024,),
                 )
+                sub_group = dataset.parent
+                sub_group.attrs["asam_data_type"] = MAP_TO_ASAM_HO.get(meas.data_type, "n/a")
+                dataset.attrs["ecu_address"] = meas.address
+                dataset.attrs["ecu_address_extension"] = meas.ext
                 dsets.append(BufferedDataset(dataset))
             self.datasets[num] = DatasetGroup(ts0_ds=ts0, ts1_ds=ts1, datasets=dsets)
         self.hdf.flush()