orange-hdf5: Update .metadata handling to match latest proposed

Author: stuart-cls

src/orangecontrib/protospec/data.py

@@ -1,4 +1,5 @@
 import json
+import pickle
 
 from os import path
 
@@ -15,6 +16,7 @@
     Table,
 )
 from Orange.data.io import FileFormat
+from Orange.data.io_base import PICKLE_PROTOCOL
 
 from Orange.version import short_version as ORANGE_VERSION  # noqa N812
 from orangecontrib.spectroscopy.io import HDF5MetaReader
@@ -38,7 +40,7 @@ def read_domain(sub):
                 if f'{sub}_args' in d
                 else ['{}'] * len(subdomain)
             )
-            for attr, args in zip(subdomain, subdomain_args, strict=False):
+            for attr, args in zip(subdomain, subdomain_args):  # noqa B905
                 yield attr[0], attr[1], json.loads(args)
 
         def make_var(name, header, args):
@@ -141,6 +143,48 @@ def parse(attr):
                     f.create_dataset(f'metas/{i}', data=col_data, dtype=col_type)
         cls.write_table_metadata(filename, data)
 
+    @classmethod
+    def write_table_metadata(cls, filename, data):
+        dump_dict = {}
+        for key, value in data.attributes.items():
+            if isinstance(value, str):
+                dump_dict[key] = value
+            else:
+                try:
+                    dump_dict[key] = json.dumps(value)
+                except TypeError:
+                    # value is not JSON serializable, fall back to pickle
+                    dump_dict[key] = pickle.dumps(value, protocol=PICKLE_PROTOCOL).hex()
+
+        with h5py.File(filename, 'r+') as f:
+            metadata_group = f.require_group('metadata')
+            str_dtype = h5py.string_dtype()
+            for key, value in dump_dict.items():
+                metadata_group.create_dataset(key, data=value, dtype=str_dtype)
+
+    @classmethod
+    def set_table_metadata(cls, filename, data):
+        with h5py.File(filename, 'r') as f:
+            if 'metadata' in f:
+                metadata_group = f['metadata']
+                for key in metadata_group:
+                    value = metadata_group[key][()]
+                    if isinstance(value, bytes):
+                        value = value.decode('utf-8')
+                    if value.startswith('{') or value.startswith('['):
+                        try:
+                            value = json.loads(value)
+                        except json.JSONDecodeError:
+                            pass
+                    elif value.startswith(f"80{PICKLE_PROTOCOL:02x}"):
+                        try:
+                            value = pickle.loads(bytes.fromhex(value))
+                        except (pickle.UnpicklingError, ValueError):
+                            pass
+                    data.attributes[key] = value
+            else:
+                super().set_table_metadata(filename, data)
+
 
 class IRisF1HDF5Reader(FileFormat):
     """Reader for IRsweep IRis-F1 HDF5 _processed_data files"""

src/orangecontrib/protospec/tests/test_io.py

@@ -1,4 +1,6 @@
+import os
 import unittest
+from tempfile import NamedTemporaryFile
 
 import numpy as np
 from Orange.data import (
@@ -41,3 +43,50 @@ def test_roundtrip_hdf5(self):
             np.testing.assert_equal(data.metas[:2], self.data.metas[:2])
             self.assertEqual(data.metas[2, 0], "")
             np.testing.assert_equal(data.domain, self.data.domain)
+
+
+class Unserializable:
+    def __init__(self, name):
+        self.name = name
+
+
+class TestWriterMetadata(unittest.TestCase):
+    def setUp(self):
+        TestHDF5.setUp(self)
+        self.data.attributes.update(
+            {
+                "Name": "Test dataset",
+                "Description": "This is a test dataset.",
+                "Author": "Unit Tester",
+                "Year": "2024",
+                "Reference": "None",
+            }
+        )
+
+    def test_metadata_hdf5(self):
+        data = self.data.copy()
+        data.attributes["CustomAttr"] = {"key1": "value1", "key2": 2}
+        with NamedTemporaryFile(suffix=".hdf5", delete=False) as f:
+            fname = f.name
+        try:
+            HDF5Reader.write(fname, data)
+            table = HDF5Reader(fname).read()
+            self.assertEqual(table.attributes, data.attributes)
+        finally:
+            os.remove(fname)
+
+    def test_metadata_hdf5_pickle(self):
+        data = self.data.copy()
+        data.attributes["Unserializable"] = Unserializable(name="test")
+        with NamedTemporaryFile(suffix=".hdf5", delete=False) as f:
+            fname = f.name
+        try:
+            HDF5Reader.write(fname, data)
+            table = HDF5Reader(fname).read()
+            for key, value in table.attributes.items():
+                if isinstance(value, Unserializable):
+                    self.assertIsInstance(data.attributes[key], Unserializable)
+                else:
+                    self.assertEqual(value, data.attributes[key])
+        finally:
+            os.remove(fname)