use hints for NXdata fields, log only once

Author: lukaspie

pyproject.toml

@@ -112,7 +112,6 @@ validate_nexus = "pynxtools.dataconverter.validate_file:validate_cli"
 
 [tool.setuptools.package-data]
 pynxtools = ["definitions/**/*.xml", "definitions/**/*.xsd"]
-"pynxtools.dataconverter.units" = ["*.txt"]
 "pynxtools.dataconverter.readers.hall" = ["enum_map.json"]
 "pynxtools.dataconverter.readers.rii_database.formula_parser" = ["dispersion_function_grammar.lark"]
 

src/pynxtools/dataconverter/helpers.py

@@ -84,7 +84,11 @@ class Collector:
     """A class to collect data and return it in a dictionary format."""
 
     def __init__(self):
-        self.data = set()
+        self.data: dict[str, set] = {
+            "warning_and_error": set(),
+            "info": set(),
+        }
+
         self.logging = True
 
     def _log(self, path: str, log_type: ValidationProblem, value: Optional[Any], *args):
@@ -223,26 +227,36 @@ def collect_and_log(
             "NX_ANY",
         ):
             return
-        if self.logging and path + str(log_type) + str(value) not in self.data:
-            self._log(path, log_type, value, *args, **kwargs)
+
+        message: str = path + str(log_type) + str(value)
+
         # info messages should not fail validation
-        if log_type not in (
+        if log_type in (
             ValidationProblem.UnitWithoutDocumentation,
             ValidationProblem.OpenEnumWithNewItem,
         ):
-            self.data.add(path + str(log_type) + str(value))
+            if self.logging and message not in self.data["info"]:
+                self._log(path, log_type, value, *args, **kwargs)
+            self.data["info"].add(message)
+        else:
+            if self.logging and message not in self.data["warning_and_error"]:
+                self._log(path, log_type, value, *args, **kwargs)
+            self.data["warning_and_error"].add(message)
 
     def has_validation_problems(self) -> bool:
         """Returns True if there were any validation problems."""
-        return len(self.data) > 0
+        return len(self.data["warning_and_error"]) > 0
 
     def get(self):
         """Returns the set of problematic paths."""
-        return self.data
+        return self.data["warning_and_error"]
 
     def clear(self):
         """Clears the collected data."""
-        self.data = set()
+        self.data: dict[str, set] = {
+            "warning_and_error": set(),
+            "info": set(),
+        }
 
 
 collector = Collector()

src/pynxtools/dataconverter/validate_file.py

@@ -95,12 +95,12 @@ def validate(file: str, ignore_undocumented: bool = False):
 
             if is_valid:
                 logger.info(
-                    f"The entry `{entry}` in file `{file}` is a valid file"
+                    f"The entry `{entry}` in file `{file}` is valid"
                     f" according to the `{nxdl}` application definition.",
                 )
             else:
                 logger.info(
-                    f"Invalid: The entry `{entry}` in file `{file}` is NOT a valid file"
+                    f"Invalid: The entry `{entry}` in file `{file}` is NOT valid"
                     f" according to the `{nxdl}` application definition.",
                 )
 

src/pynxtools/dataconverter/validation.py

@@ -200,6 +200,7 @@ def validate_hdf_group_against(
     def best_namefit_of(
         name: str,
         nodes: Iterable[NexusNode],
+        hint: Optional[Literal["axis", "signal"]] = None,
     ) -> Optional[NexusNode]:
         """
         Get the best namefit of `name` in `nodes`.
@@ -218,6 +219,8 @@ def best_namefit_of(
         best_match = None
         best_score = -1
 
+        hint_map: dict[str, str] = {"DATA": "signal", "AXISNAME": "axis"}
+
         for node in nodes:
             if not node.variadic:
                 if name == node.name:
@@ -227,22 +230,28 @@ def best_namefit_of(
                 name_partial = node.name_type == "partial"
                 score = get_nx_namefit(name, node.name, name_any, name_partial)
                 if score > best_score:
+                    if hint and hint_map.get(node.name) != hint:
+                        continue
                     best_match = node
                     best_score = score
 
+        # if hint:
+        #     print(hint, node, name)
+
         return best_match
 
     # Only cache based on path. That way we retain the nx_class information
     # in the tree
     # Allow for 10000 cache entries. This should be enough for most cases
     @cached(
         cache=LRUCache(maxsize=10000),
-        key=lambda path, node_type=None, nx_class=None: hashkey(path),
+        key=lambda path, node_type=None, nx_class=None, hint=None: hashkey(path),
     )
     def find_node_for(
         path: str,
         node_type: Optional[Literal["group", "field", "attribute"]] = None,
         nx_class: Optional[str] = None,
+        hint: Optional[Literal["axis", "signal"]] = None,
     ) -> Optional[NexusNode]:
         """
         Find the NexusNode for a given HDF5 path, optionally constrained by node type and NX_class.
@@ -264,7 +273,7 @@ def find_node_for(
 
         *prev_path, last_elem = path.rsplit("/", 1)
 
-        node = find_node_for(prev_path[0]) if prev_path else tree
+        node = find_node_for(prev_path[0], hint=hint) if prev_path else tree
         current = copy.copy(node)
 
         if node is None:
@@ -276,7 +285,7 @@ def find_node_for(
                 nx_class=nx_class, node_type=node_type
             )
         ]
-        node = best_namefit_of(last_elem, children_to_check)
+        node = best_namefit_of(last_elem, children_to_check, hint)
 
         if node is None:
             # Check that there is no other node with the same name, but a different type
@@ -563,21 +572,25 @@ def check_reserved_prefix(
 
         return
 
-    def handle_group(path: str, data: h5py.Group):
+    def handle_group(path: str, group: h5py.Group):
         """
         Handle validation logic for HDF5 groups.
 
         Args:
             path (str): Relative HDF5 path to the group.
-            data (h5py.Group): The group object.
+            group (h5py.Group): The group object.
         """
         full_path = f"{entry_name}/{path}"
 
         check_reserved_prefix(full_path, appdef_node.name, "group")
 
+        if not group.attrs.get("NX_class"):
+            # We ignore additional groups that don't have an NX_class
+            return
+
         try:
             node = find_node_for(
-                path, node_type="group", nx_class=data.attrs.get("NX_class")
+                path, node_type="group", nx_class=group.attrs.get("NX_class")
             )
         except TypeError:
             return
@@ -596,22 +609,22 @@ def handle_group(path: str, data: h5py.Group):
             return
 
         if node.nx_class == "NXdata":
-            handle_nxdata(path, data)
+            handle_nxdata(path, group)
         if node.nx_class == "NXcollection":
             return
 
-    def handle_nxdata(path: str, data: h5py.Group):
+    def handle_nxdata(path: str, group: h5py.Group):
         """
         Handle validation of NXdata groups, including signal, axes, and auxiliary signals.
 
         Args:
             path (str): HDF5 path to the NXdata group.
-            data (h5py.Group): The NXdata group object.
+            group (h5py.Group): The NXdata group object.
         """
         full_path = f"{entry_name}/{path}"
 
         def check_nxdata():
-            data_field = data.get(signal)
+            data_field = group.get(signal)
 
             if data_field is None:
                 collector.collect_and_log(
@@ -620,23 +633,20 @@ def check_nxdata():
                     None,
                 )
             else:
-                handle_field(
-                    f"{path}/{signal}",
-                    data_field,
-                )
+                handle_field(f"{path}/{signal}", data_field, hint="signal")
 
             # check NXdata attributes
             attrs = ("signal", "auxiliary_signals", "axes")
-            data_attrs = {k: data.attrs[k] for k in attrs if k in data.attrs}
+            data_attrs = {k: group.attrs[k] for k in attrs if k in group.attrs}
 
             handle_attributes(path, data_attrs)
 
             for i, axis in enumerate(axes):
                 if axis == ".":
                     continue
-                index = data.get(f"{axis}_indices", i)
+                index = group.get(f"{axis}_indices", i)
 
-                axis_field = data.get(axis)
+                axis_field = group.get(axis)
 
                 if axis_field is None:
                     collector.collect_and_log(
@@ -646,10 +656,7 @@ def check_nxdata():
                     )
                     break
                 else:
-                    handle_field(
-                        f"{path}/{axis}",
-                        data_field,
-                    )
+                    handle_field(f"{path}/{axis}", axis_field, hint="axis")
                 if np.shape(data_field)[index] != len(axis_field):
                     collector.collect_and_log(
                         f"{path}/{axis}",
@@ -658,45 +665,57 @@ def check_nxdata():
                         index,
                     )
 
-        signal = data.attrs.get("signal")
-        aux_signals = data.attrs.get("auxiliary_signals", [])
-        axes = data.attrs.get("axes", [])
+        signal = group.attrs.get("signal")
+        aux_signals = group.attrs.get("auxiliary_signals", [])
+        axes = group.attrs.get("axes", [])
 
         if isinstance(axes, str):
             axes = [axes]
 
-        if signal is not None:
-            check_nxdata()
-
         indices = map(lambda x: f"{x}_indices", axes)
         errors = map(lambda x: f"{x}_errors", [signal, *aux_signals, *axes])
 
-    def handle_field(path: str, data: h5py.Dataset):
+        # TODO: check that the indices match
+        # TODO: check that the errors have the same dim as the fields
+
+        if signal is not None:
+            check_nxdata()
+
+    def handle_field(
+        path: str,
+        dataset: h5py.Dataset,
+        hint: Optional[Literal["axis", "signal"]] = None,
+    ):
         """
         Validate a NeXus field (dataset) within the HDF5 structure.
 
         Args:
             path (str): Path to the dataset.
             data (h5py.Dataset): Dataset object.
+            hint (str):
+                If the field is in an NXdata group, this is used to figure out
+                if it is an AXISNAME or a DATA.
         """
+
         full_path = f"{entry_name}/{path}"
         check_reserved_prefix(full_path, appdef_node.name, "field")
         try:
-            node = find_node_for(path, node_type="field")
+            node = find_node_for(path, node_type="field", hint=hint)
         except TypeError:
             return
+
         if node is None:
             key_path = path.replace("@", "")
+            parent_node = None
             while "/" in key_path:
                 key_path = key_path.rsplit("/", 1)[0]  # Remove last segment
-                parent_node = find_node_for(path, node_type="field")
-                if parent_node is None:
-                    parent_node = find_node_for(key_path, node_type="group")
-                if (
-                    parent_node
-                    and parent_node.type == "group"
-                    and parent_node.nx_class == "NXcollection"
-                ):
+                parent_data = data.get(key_path)
+                nx_class = (
+                    parent_data.attrs.get("NX_class")
+                    if parent_data is not None
+                    else None
+                )
+                if nx_class == "NXcollection":
                     # Collection found for parents, mark as documented
                     return
 
@@ -714,10 +733,14 @@ def handle_field(path: str, data: h5py.Dataset):
             return
 
         is_valid_data_field(
-            clean_str_attr(data[()]), node.dtype, node.items, node.open_enum, full_path
+            clean_str_attr(dataset[()]),
+            node.dtype,
+            node.items,
+            node.open_enum,
+            full_path,
         )
 
-        units = data.attrs.get("units")
+        units = dataset.attrs.get("units")
         units_path = f"{full_path}/@units"
         if node.unit is not None:
             remove_from_req_entities(f"{path}/@units")
@@ -730,7 +753,7 @@ def handle_field(path: str, data: h5py.Dataset):
                     return
             # Special case: NX_TRANSFORMATION unit depends on `@transformation_type` attribute
             if (
-                transformation_type := data.attrs.get("transformation_type")
+                transformation_type := dataset.attrs.get("transformation_type")
             ) is not None:
                 hints = {"transformation_type": transformation_type}
             else:
@@ -766,26 +789,26 @@ def handle_attributes(path: str, attrs: h5py.AttributeManager):
                 node = find_node_for(f"{path}/{attr_name}", node_type="attribute")
             except TypeError:
                 return
-            if node is None:
-                key_path = full_path.replace("@", "")
-                found_collection = False
-                while "/" in key_path:
-                    key_path = key_path.rsplit("/", 1)[0]  # Remove last segment
-                    parent_node = find_node_for(path, node_type="field")
-                    if parent_node is None:
-                        parent_node = find_node_for(key_path, node_type="group")
 
-                    if (
-                        parent_node
-                        and parent_node.type == "group"
-                        and parent_node.nx_class == "NXcollection"
-                    ):
-                        # Collection found for parents, mark as documented
-                        found_collection = True
-                        break
+            key_path = f"{path}/{attr_name}"
+            parent_node = None
+            found_collection = False
+            while "/" in key_path:
+                key_path = key_path.rsplit("/", 1)[0]  # Remove last segment
+                parent_data = data.get(key_path)
+                nx_class = (
+                    parent_data.attrs.get("NX_class")
+                    if parent_data is not None
+                    else None
+                )
+                if nx_class == "NXcollection":
+                    # Collection found for parents, mark as documented
+                    found_collection = True
+                    break
+            if found_collection:
+                continue  # This continues the outer attr_name loop
 
-                if found_collection:
-                    continue  # This continues the outer attr_name loop
+            if node is None:
                 if not ignore_undocumented:
                     collector.collect_and_log(
                         full_path,

src/pynxtools/testing/nexus_conversion.py

@@ -164,7 +164,7 @@ def convert_to_nexus(
 
         assert test_output == []
 
-        # Validate created file using the verify_nexus functionality
+        # Validate created file using the validate_nexus functionality
         validate(self.created_nexus, ignore_undocumented=ignore_undocumented)
 
         if NOMAD_AVAILABLE: