A quick append mode for already existing hdf5 files

pynxtools/dataconverter/convert.py

@@ -63,13 +63,14 @@ def get_names_of_all_readers() -> List[str]:
     return all_readers
 
 
-# pylint: disable=too-many-arguments
+# pylint: disable=too-many-arguments, too-many-locals
 def convert(input_file: Tuple[str],
             reader: str,
             nxdl: str,
             output: str,
             generate_template: bool = False,
             fair: bool = False,
+            append: bool = False,
             **kwargs):
     """The conversion routine that takes the input parameters and calls the necessary functions."""
     # Reading in the NXDL and generating a template
@@ -124,7 +125,7 @@ def convert(input_file: Tuple[str],
             continue
         logger.warning("The path, %s, is being written but has no documentation.", path)
 
-    Writer(data=data, nxdl_path=nxdl_path, output_path=output).write()
+    Writer(data=data, nxdl_path=nxdl_path, output_path=output, append=append).write()
 
     logger.info("The output file generated: %s", output)
 
@@ -179,13 +180,20 @@ def parse_params_file(params_file):
     default=None,
     help='Allows to pass a .yaml file with all the parameters the converter supports.'
 )
+@click.option(
+    '--append',
+    is_flag=True,
+    default=False,
+    help="Appends to the given <output> file if it already exists."
+)
 def convert_cli(input_file: Tuple[str],
                 reader: str,
                 nxdl: str,
                 output: str,
                 generate_template: bool,
                 fair: bool,
-                params_file: str):
+                params_file: str,
+                append: bool):
     """The CLI entrypoint for the convert function"""
     if params_file:
         try:
@@ -201,7 +209,7 @@ def convert_cli(input_file: Tuple[str],
             sys.tracebacklimit = 0
             raise IOError("\nError: Please supply an NXDL file with the option:"
                           " --nxdl <path to NXDL>")
-        convert(input_file, reader, nxdl, output, generate_template, fair)
+        convert(input_file, reader, nxdl, output, generate_template, fair, append)
 
 
 if __name__ == '__main__':

pynxtools/dataconverter/writer.py

@@ -183,12 +183,16 @@ class Writer:
         nxs_namespace (str): The namespace used in the NXDL tags. Helps search for XML children.
     """
 
-    def __init__(self, data: dict = None, nxdl_path: str = None, output_path: str = None):
+    def __init__(self,
+                 data: dict = None,
+                 nxdl_path: str = None,
+                 output_path: str = None,
+                 append: bool = False):
         """Constructs the necessary objects required by the Writer class."""
         self.data = data
         self.nxdl_path = nxdl_path
         self.output_path = output_path
-        self.output_nexus = h5py.File(self.output_path, "w")
+        self.output_nexus = h5py.File(self.output_path, "a" if append else "w")
         self.nxdl_data = ET.parse(self.nxdl_path).getroot()
         self.nxs_namespace = get_namespace(self.nxdl_data)
 

tests/dataconverter/test_writer.py

@@ -41,6 +41,23 @@ def fixture_writer(filled_test_data, tmp_path):
     del writer
 
 
+@pytest.mark.usefixtures("filled_test_data")
+@pytest.fixture(name="writer_append")
+def fixture_writer_append(filled_test_data, tmp_path):
+    """pytest fixture to setup Writer object with append mode."""
+    with h5py.File(os.path.join(tmp_path, "append.nxs"), "w") as append_file:
+        append_file["/already/existing_value"] = 1
+
+    writer = Writer(
+        filled_test_data,
+        os.path.join("tests", "data", "dataconverter", "NXtest.nxdl.xml"),
+        os.path.join(tmp_path, "append.nxs"),
+        append=True
+    )
+    yield writer
+    del writer
+
+
 def test_init(writer):
     """Test to verify Writer's initialization works."""
     assert isinstance(writer, Writer)
@@ -81,3 +98,12 @@ def test_wrong_dict_provided_in_template(filled_test_data, tmp_path):
                                        "A dictionary was provided to the template but it didn't "
                                        "fall into any of the know cases of handling dictionaries"
                                        ". This occured for: ext_link")
+
+
+def test_append(writer_append):
+    """Test whether append is correctly working for the writer."""
+    # TODO: Should already existing fields be overwritten or not. Proposal: Ask everytime (y/n)
+    writer_append.write()
+    with h5py.File(writer_append.output_path, "r") as append_file:
+        assert append_file["/already/existing_value"][()] == 1
+        assert append_file["/my_entry/definition"].asstr()[...] == "NXtest"  # pylint: disable=no-member