add NeXus models with convertion from XDI

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

@@ -6,4 +6,7 @@ Library for reading and writing XAS data in `NeXus format <https://www.nexusform
 .. toctree::
     :hidden:
 
+    install
+    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,60 @@
 import sys
+import pathlib
 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=pathlib.Path, 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)
+    output_filename = args.output_filename
+    model_type = models.MODELS[args.output_format]
+
+    if output_filename.exists():
+        result = input(f"Overwrite {output_filename}? (y/[n])")
+        if not result.lower() in ("y", "yes"):
+            return 0
+        output_filename.unlink()
+    output_filename.parent.mkdir(parents=True, exist_ok=True)
+
+    filenames = list()
+    for pattern in args.file_patterns:
+        filenames.extend(glob(pattern))
+
+    convert_files(filenames, model_type, output_filename, args.output_format)
 
 
 if __name__ == "__main__":

src/pynxxas/io/__init__.py

@@ -0,0 +1,30 @@
+"""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)
+    if nexus.is_nexus_file(url):
+        yield from nexus.load_nexus_file(url)
+    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,73 @@
+import logging
+from contextlib import contextmanager
+from typing import Iterator, Generator
+
+import pydantic
+
+from .. import io
+from ..models import convert
+
+logger = logging.getLogger(__name__)
+
+
+def convert_files(
+    filenames: Iterator[str], model_type: str, output_filename: str, output_format: str
+) -> int:
+    state = {"return_code": 0, "scan_number": 0, "filename": None}
+    scan_number = 0
+    for model_in in _iter_load_models(filenames, 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(
+    filenames: Iterator[str], state: dict
+) -> Generator[pydantic.BaseModel, None, None]:
+    for filename in filenames:
+        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)

src/pynxxas/io/hdf5_utils.py

@@ -0,0 +1,41 @@
+import os
+from typing import Optional, Union
+
+import h5py
+
+
+def create_hdf5_link(
+    h5group: h5py.Group,
+    target_name: str,
+    target_filename: Optional[str],
+    absolute: bool = False,
+) -> Union[h5py.SoftLink, h5py.ExternalLink]:
+    """Create HDF5 soft link (supports relative down paths) or external link (supports relative paths)."""
+    this_name = h5group.name
+    this_filename = h5group.file.filename
+
+    target_filename = target_filename or this_filename
+
+    if os.path.isabs(target_filename):
+        rel_target_filename = os.path.relpath(target_filename, this_filename)
+    else:
+        rel_target_filename = target_filename
+        target_filename = os.path.abs(os.path.join(this_filename, target_filename))
+
+    if "." not in target_name:
+        rel_target_name = os.path.relpath(target_name, this_name)
+    else:
+        rel_target_name = target_name
+        target_name = os.path.abspath(os.path.join(this_name, target_name))
+
+    # Internal link
+    if rel_target_filename == ".":
+        if absolute or ".." in rel_target_name:
+            # h5py.SoftLink does not support relative links upwards
+            return h5py.SoftLink(target_name)
+        return h5py.SoftLink(rel_target_name)
+
+    # External link
+    if absolute:
+        return h5py.ExternalLink(target_filename, target_name)
+    return h5py.ExternalLink(rel_target_filename, target_name)