NXxas pydantic model

.github/workflows/ci.yml

@@ -35,7 +35,7 @@ jobs:
           pip install .[test]
 
       - name: Run unit tests
-        run: pytest -v .
+        run: pytest -v -W error .
 
       - name: Install doc dependencies
         run: pip install .[doc]

doc/howtoguides.rst

@@ -0,0 +1,6 @@
+How-to Guides
+=============
+
+.. toctree::
+
+    howtoguides/nxdl_validate

doc/howtoguides/nxdl_validate.rst

@@ -0,0 +1,26 @@
+Validate NXDL definitions
+=========================
+
+Validate all NXDL definitions from the official NeXus repository
+
+.. code:: bash
+
+    nxdl_validate
+
+Validate one or more specific NXDL definitions from the official NeXus repository
+
+.. code:: bash
+
+    nxdl_validate NXentry NXtomo
+
+Validate from another NeXus repository
+
+.. code:: bash
+
+    nxdl_validate --url https://github.com/XraySpectroscopy/nexus_definitions.git NXxas_new
+
+Validate from a local NeXus repository
+
+.. code:: bash
+
+    nxdl_validate --dir /home/${USER}/projects/definitions NXsample NXfluo NXmx

doc/index.rst

@@ -3,7 +3,9 @@ pynxxas |version|
 
 Library for reading and writing XAS data in `NeXus format <https://www.nexusformat.org/>`_.
 
+
 .. toctree::
     :hidden:
 
+    howtoguides
     api

setup.cfg

@@ -23,6 +23,9 @@ package_dir=
 packages=find:
 python_requires = >=3.8
 install_requires = 
+    gitpython
+    xmlschema
+    pydantic >=2
 
 [options.packages.find]
 where=src
@@ -40,6 +43,11 @@ doc =
     sphinx-autodoc-typehints >=1.16
     pydata-sphinx-theme < 0.15
 
+[options.entry_points]
+console_scripts =
+    nxdl_validate=pynxxas.apps.nxdl_validate:main
+    nx_validate=pynxxas.apps.nx_validate:main
+
 # E501 (line too long) ignored for now
 # E203 and W503 incompatible with black formatting (https://black.readthedocs.io/en/stable/compatible_configs.html#flake8)
 [flake8]

src/pynxxas/apps/nx_validate.py

@@ -0,0 +1,40 @@
+import sys
+import logging
+import argparse
+
+from .. import nexus
+
+
+def main(argv=None):
+    if argv is None:
+        argv = sys.argv
+
+    parser = argparse.ArgumentParser(
+        prog="nx_validate", description="Validate NeXus instances"
+    )
+
+    parser.add_argument("--url", type=str, default=None, help="NXDL repository URL")
+    parser.add_argument(
+        "--dir", type=str, default=None, help="Local directory of the NXDL repository"
+    )
+    parser.add_argument(
+        "instances",
+        type=str,
+        nargs="+",
+        help="NeXus instances to validate (e.g. HDF5 files)",
+    )
+
+    args = parser.parse_args(argv[1:])
+    logging.basicConfig()
+
+    instances = args.instances
+    repo_options = {"localdir": args.dir, "url": args.url, "reset": not args.dir}
+
+    for instance in instances:
+        # TODO: Load instance content and validate with model
+        name = instance
+        _ = nexus.load_model(name, **repo_options)
+
+
+if __name__ == "__main__":
+    sys.exit(main())

src/pynxxas/apps/nxdl_validate.py

@@ -0,0 +1,38 @@
+import sys
+import logging
+import argparse
+
+from .. import nxdl
+
+
+def main(argv=None):
+    if argv is None:
+        argv = sys.argv
+
+    parser = argparse.ArgumentParser(
+        prog="nxdl_validate", description="Validate NXDL definitions"
+    )
+
+    parser.add_argument("--url", type=str, default=None, help="NXDL repository URL")
+    parser.add_argument(
+        "--dir", type=str, default=None, help="Local directory of the NXDL repository"
+    )
+    parser.add_argument(
+        "definitions", type=str, nargs="*", help="NXDL definitions to validate"
+    )
+
+    args = parser.parse_args(argv[1:])
+    logging.basicConfig()
+
+    definitions = args.definitions
+    repo_options = {"localdir": args.dir, "url": args.url, "reset": not args.dir}
+
+    if not definitions:
+        definitions = nxdl.get_nxdl_definition_names(**repo_options)
+
+    for name in definitions:
+        _ = nxdl.load_definition(name, **repo_options)
+
+
+if __name__ == "__main__":
+    sys.exit(main())

src/pynxxas/nexus/__init__.py

@@ -0,0 +1 @@
+from .models import load_model  # noqa F401

src/pynxxas/nexus/models.py

@@ -0,0 +1,92 @@
+"""NeXus model"""
+
+import pydantic
+from typing import Union
+
+from ..nxdl import models as nxdl_models
+from ..nxdl import load_definition as load_nxdl_definition
+
+
+def load_model(name: str, **repo_options) -> pydantic.BaseModel:
+    return _NexusModelCreator(repo_options).create(name)
+
+
+class _NexusModelCreator:
+    def __init__(self, repo_options: dict) -> None:
+        self._repo_options = repo_options
+        self._definitions = dict()
+
+    def create(self, name: str):
+        self._create_group(self._get_definition(name))
+
+    def _create_group(
+        self,
+        nxclass: Union[nxdl_models.Definition, nxdl_models.Group],
+        *args,
+        indent=0,
+    ):
+        if isinstance(nxclass, nxdl_models.Definition):
+            nxclass_parent = self._get_definition(nxclass.extends)
+            assert nxclass_parent.type == "group"
+        else:
+            nxclass_parent = self._get_definition(nxclass.type)
+
+        # nxclass inherits from nxclass_parent while overwriting attributes,
+        # fields and groups.
+
+        # TODO: use pydantic.create_model to create a merged model
+        #       from nxclass and nxclass_parent
+        # Circular sub-groups: NXgeometry <-> NXtranslation
+
+        self._print(indent, nxclass.name, nxclass.type, *args)
+        indent += 1
+        if indent > 5:
+            return  # circular dependencies
+
+        base_attributes = {attr.name: attr for attr in nxclass_parent.attribute}
+        overwrite_attributes = {attr.name: attr for attr in nxclass.attribute}
+        attributes = {**base_attributes, **overwrite_attributes}
+
+        base_fields = {field.name: field for field in nxclass_parent.field}
+        overwrite_fields = {field.name: field for field in nxclass.field}
+        fields = {**base_fields, **overwrite_fields}
+
+        base_groups = {group.name: group for group in nxclass_parent.group}
+        overwrite_groups = {group.name: group for group in nxclass.group}
+        groups = {**base_groups, **overwrite_groups}
+
+        for attr in attributes.values():
+            occurs = _occurs(1, 1, attr.optional, attr.recommended)
+            self._print(indent, attr.name, attr.type, occurs)
+
+        for field in fields.values():
+            occurs = _occurs(
+                field.minOccurs, field.maxOccurs, field.optional, field.recommended
+            )
+            self._print(indent, field.name, field.type, occurs)
+
+        for group in groups.values():
+            occurs = _occurs(
+                group.minOccurs, group.maxOccurs, group.optional, group.recommended
+            )
+            self._create_group(group, occurs, indent=indent)
+
+    def _print(self, indent: int, *args):
+        print(" " * indent, *args)
+
+    def _get_definition(self, name: str) -> nxdl_models.Definition:
+        definition = self._definitions.get(name)
+        if definition:
+            return definition
+        definition = load_nxdl_definition(name, **self._repo_options)
+        self._definitions[name] = definition
+        return definition
+
+
+def _occurs(minOccurs, maxOccurs, optional, recommended):
+    optional = optional or recommended
+    if optional:
+        minOccurs = 0
+    if not isinstance(maxOccurs, int):
+        maxOccurs = float("inf")
+    return f"[{minOccurs}, {maxOccurs}]"

src/pynxxas/nxdl/__init__.py

@@ -0,0 +1,14 @@
+import pydantic
+from .repo import get_nxdl_definition_names  # noqa F401
+from .repo import get_nxdl_definition as _get_nxdl_definition
+from .models import Definition as _Definition
+
+
+def load_definition(name: str, **repo_options) -> _Definition:
+    """Every NeXus definition is defined in an XML file using the
+    NXDL schema with a 'definition' root element."""
+    xml_content = _get_nxdl_definition(name, **repo_options)
+    try:
+        return _Definition(**xml_content)
+    except pydantic.ValidationError as e:
+        raise ValueError(f"NeXus definition '{name}' is invalid") from e