add NeXus models with convertion from XDI

.gitignore

@@ -15,3 +15,4 @@ __pycache__/
 *.egg-info/
 .eggs/
 /doc/_generated
+/doc/_static/example_nxxas_data.h5

README.md

@@ -10,4 +10,6 @@ Library for reading and writing XAS data in NeXus format.
         <img src="https://img.shields.io/badge/license-MIT-blue" /></a>
     <a href="https://github.com/psf/black" alt="Code Style">
         <img src="https://img.shields.io/badge/code%20style-black-000000.svg" /></a>
+    <a href="https://myhdf5.hdfgroup.org/view" alt="NeXus">
+        <img src="https://raw.githubusercontent.com/nexusformat/wiki/master/public/favicon.ico" /></a>
 </p>

doc/_ext/myhdf5_inline_role.py

@@ -0,0 +1,50 @@
+import re
+import os
+from docutils import nodes
+from pynxxas.io.convert import convert_files
+
+
+def setup(app):
+    app.add_role("myhdf5", myhdf5_role)
+    app.connect("html-page-context", inject_dynamic_url_js)
+    app.connect("builder-inited", generate_example_nxxas_data)
+
+
+def myhdf5_role(name, rawtext, text, lineno, inliner, options={}, content=[]):
+    matches = re.match(r"(\S+)\s*<([^<>]+)>", text)
+    display_text = matches.group(1)
+    filename = matches.group(2)
+
+    url_template = f"https://myhdf5.hdfgroup.org/view?url=placeholder{filename}"
+
+    link = f'<a class="myhdf5" href="{url_template}">{display_text}</a>'
+
+    node = nodes.raw("", link, format="html")
+    return [node], []
+
+
+def inject_dynamic_url_js(app, pagename, templatename, context, doctree):
+    if app.builder.name != "html" or doctree is None:
+        return
+
+    script = """
+    <script>
+    document.addEventListener("DOMContentLoaded", function() {
+        var links = document.querySelectorAll("a.myhdf5");
+        var currentURL = encodeURIComponent(window.location.href + "/");
+        links.forEach(function(link) {
+            var href = link.getAttribute("href");
+            link.setAttribute("href", href.replace("placeholder", currentURL));
+        });
+    });
+    </script>
+    """
+
+    context["body"] += script
+
+
+def generate_example_nxxas_data(app):
+    output_filename = os.path.join(app.srcdir, "_static", "example_nxxas_data.h5")
+    file_pattern1 = os.path.join(app.srcdir, "..", "xdi_files", "*")
+    file_pattern2 = os.path.join(app.srcdir, "..", "xas_beamline_data", "*")
+    convert_files([file_pattern1, file_pattern2], output_filename, "nexus")

doc/conf.py

@@ -4,8 +4,12 @@
 # -- Project information -----------------------------------------------------
 # https://www.sphinx-doc.org/en/master/usage/configuration.html#project-information
 
+import os
+import sys
 from pynxxas import __version__ as release
 
+sys.path.append(os.path.abspath("./_ext"))
+
 project = "pynxxas"
 version = ".".join(release.split(".")[:2])
 copyright = "2024-present, ESRF"
@@ -20,6 +24,7 @@
     "sphinx.ext.autosummary",
     "sphinx.ext.viewcode",
     "sphinx_autodoc_typehints",
+    "myhdf5_inline_role",
 ]
 templates_path = ["_templates"]
 exclude_patterns = ["build"]
@@ -39,7 +44,8 @@
 # https://www.sphinx-doc.org/en/master/usage/configuration.html#options-for-html-output
 
 html_theme = "pydata_sphinx_theme"
-html_static_path = []
+html_static_path = ["_static"]
+html_extra_path = ["_static"]
 html_theme_options = {
     "icon_links": [
         {

doc/howtoguides.rst

@@ -0,0 +1,7 @@
+How-to Guides
+=============
+
+.. toctree::
+
+    howtoguides/install
+    howtoguides/convert_files

doc/howtoguides/convert_files.rst

@@ -0,0 +1,8 @@
+Convert file formats
+====================
+
+Convert all files in the *xdi_files* and *xas_beamline_data* to *HDF5/NeXus* format
+
+.. code-block:: bash
+
+    nxxas-convert xdi_files/*.* xas_beamline_data/*.* ./converted/data.h5

doc/howtoguides/install.rst

@@ -0,0 +1,6 @@
+Install
+=======
+
+.. code-block:: bash
+
+    pip install pynxxas

doc/index.rst

@@ -3,7 +3,11 @@ pynxxas |version|
 
 Library for reading and writing XAS data in `NeXus format <https://www.nexusformat.org/>`_.
 
+An example HDF5 file can be found :myhdf5:`here <example_nxxas_data.h5>`.
+
 .. toctree::
     :hidden:
 
+    howtoguides
+    tutorials
     api

doc/tutorials.rst

@@ -0,0 +1,6 @@
+Tutorials
+=========
+
+.. toctree::
+
+    tutorials/models

doc/tutorials/models.rst

@@ -0,0 +1,34 @@
+Data models
+===========
+
+Data from different data formats are represented in memory as a *pydantic* models.
+You can convert between different models and save/load models from file.
+
+NeXus models
+------------
+
+Build an *NXxas* model instance in steps
+
+.. code-block:: python
+
+    from pynxxas.models import NxXasModel
+
+    nxxas_model = NxXasModel(element="Fe", absorption_edge="K", mode="transmission")
+    nxxas_model.energy = [7, 7.1], "keV"
+    nxxas_model.intensity = [10, 20]
+
+Create an *NXxas* model instance from a dictionary and convert back to a dictionary
+
+.. code-block:: python
+
+    data_in = {
+        "NX_class": "NXsubentry",
+        "mode": "transmission",
+        "element": "Fe",
+        "absorption_edge": "K",
+        "energy": [[7, 7.1], "keV"],
+        "intensity": [10, 20],
+    }
+
+    nxxas_model = NxXasModel(**data_in)
+    data_out = nxxas_model.model_dump()

setup.cfg

@@ -23,11 +23,13 @@ package_dir=
 packages=find:
 python_requires = >=3.8
 install_requires = 
+    typing_extensions; python_version < "3.9"
+    strenum; python_version < "3.11"
     numpy
     h5py
     pydantic >=2.6
     pint
-    typing_extensions; python_version < "3.9"
+    periodictable
 
 [options.packages.find]
 where=src

src/pynxxas/apps/__init__.py

@@ -0,0 +1,2 @@
+"""Command-Line Interface (CLI)
+"""

src/pynxxas/apps/nxxas_convert.py

@@ -1,33 +1,46 @@
 import sys
 import logging
 import argparse
-from glob import glob
 
-from ..io.xdi import read_xdi
+from .. import models
+from ..io.convert import convert_files
 
+logger = logging.getLogger(__name__)
 
-def main(argv=None):
+
+def main(argv=None) -> int:
     if argv is None:
         argv = sys.argv
 
     parser = argparse.ArgumentParser(
         prog="nxxas_convert", description="Convert data to NXxas format"
     )
 
-    parser.add_argument("--output", type=str, default=None, help="Path to HDF5 file")
     parser.add_argument(
-        "patterns",
+        "--output-format",
+        type=str,
+        default="nexus",
+        choices=list(models.MODELS),
+        help="Output format",
+    )
+
+    parser.add_argument(
+        "file_patterns",
         type=str,
-        nargs="+",
-        help="Glob file name patterns",
+        nargs="*",
+        help="Files to convert",
+    )
+
+    parser.add_argument(
+        "output_filename", type=str, help="Convert destination filename"
     )
 
     args = parser.parse_args(argv[1:])
     logging.basicConfig()
 
-    for pattern in args.patterns:
-        for filename in glob(pattern):
-            read_xdi(filename)
+    convert_files(
+        args.file_patterns, args.output_filename, args.output_format, interactive=True
+    )
 
 
 if __name__ == "__main__":

src/pynxxas/io/__init__.py

@@ -0,0 +1,31 @@
+"""File formats
+"""
+
+from typing import Generator
+
+import pydantic
+
+from .url_utils import UrlType
+from . import xdi
+from . import nexus
+from .. import models
+
+
+def load_models(url: UrlType) -> Generator[pydantic.BaseModel, None, None]:
+    if xdi.is_xdi_file(url):
+        yield from xdi.load_xdi_file(url)
+    elif nexus.is_nexus_file(url):
+        yield from nexus.load_nexus_file(url)
+    else:
+        raise NotImplementedError(f"File format not supported: {url}")
+
+
+def save_model(model_instance: pydantic.BaseModel, url: UrlType) -> None:
+    if isinstance(model_instance, models.NxXasModel):
+        nexus.save_nexus_file(model_instance, url)
+    elif isinstance(model_instance, models.XdiModel):
+        xdi.save_xdi_file(model_instance, url)
+    else:
+        raise NotImplementedError(
+            f"Saving of {type(model_instance).__name__} not implemented"
+        )

src/pynxxas/io/convert.py

@@ -0,0 +1,92 @@
+import logging
+import pathlib
+from glob import glob
+from contextlib import contextmanager
+from typing import Iterator, Generator
+
+import pydantic
+
+from .. import io
+from .. import models
+from ..models import convert
+
+logger = logging.getLogger(__name__)
+
+
+def convert_files(
+    file_patterns: Iterator[str],
+    output_filename: str,
+    output_format: str,
+    interactive: bool = False,
+) -> int:
+    model_type = models.MODELS[output_format]
+
+    output_filename = pathlib.Path(output_filename)
+    if output_filename.exists():
+        if interactive:
+            result = input(f"Overwrite {output_filename}? (y/[n])")
+            if not result.lower() in ("y", "yes"):
+                return 1
+        output_filename.unlink()
+    output_filename.parent.mkdir(parents=True, exist_ok=True)
+
+    state = {"return_code": 0, "scan_number": 0, "filename": None}
+    scan_number = 0
+    for model_in in _iter_load_models(file_patterns, state):
+        scan_number += 1
+        for model_out in _iter_convert_model(model_in, model_type, state):
+            if output_format == "nexus":
+                output_url = f"{output_filename}?path=/dataset{scan_number:02}"
+                if model_out.NX_class == "NXsubentry":
+                    breakpoint()
+                    output_url = f"{output_url}/{model_out.mode.replace(' ', '_')}"
+            else:
+                basename = f"{output_filename.stem}_{scan_number:02}"
+                if model_out.NX_class == "NXsubentry":
+                    basename = f"{basename}_{model_out.mode.replace(' ', '_')}"
+                output_url = output_filename.parent / basename + output_filename.suffix
+
+            with _handle_error("saving", state):
+                io.save_model(model_out, output_url)
+
+    return state["return_code"]
+
+
+def _iter_load_models(
+    file_patterns: Iterator[str], state: dict
+) -> Generator[pydantic.BaseModel, None, None]:
+    for file_pattern in file_patterns:
+        for filename in glob(file_pattern):
+            filename = pathlib.Path(filename).absolute()
+            state["filename"] = filename
+            it_model_in = io.load_models(filename)
+            while True:
+                with _handle_error("loading", state):
+                    try:
+                        yield next(it_model_in)
+                    except StopIteration:
+                        break
+
+
+def _iter_convert_model(
+    model_in: Iterator[pydantic.BaseModel], model_type: str, state: dict
+) -> Generator[pydantic.BaseModel, None, None]:
+    it_model_out = convert.convert_model(model_in, model_type)
+    while True:
+        with _handle_error("converting", state):
+            try:
+                yield next(it_model_out)
+            except StopIteration:
+                break
+
+
+@contextmanager
+def _handle_error(action: str, state: dict) -> Generator[None, None, None]:
+    try:
+        yield
+    except NotImplementedError as e:
+        state["return_code"] = 1
+        logger.warning("Error when %s '%s': %s", action, state["filename"], e)
+    except Exception:
+        state["return_code"] = 1
+        logger.error("Error when %s '%s'", action, state["filename"], exc_info=True)