Add templating support to NXWriter()

CHANGES.rst

@@ -28,7 +28,7 @@ describe the future plans.
 1.6.18
 ******
 
-release expected by 2023-12-01
+release expected by 2023-12-15
 
 New Features
 ------------
@@ -37,6 +37,7 @@ New Features
   * DG-645 digital delay/pulse generator
   * Measurement Computing USB CTR08 High-Speed Counter/Timer
 * Add subnet check for APSU beamlines.
+* Add template support for writing NeXus/HDF5 files.
 * New lineup2() plan can be used in console, notebooks, and queueserver.
 
 Maintenance

apstools/callbacks/nexus_writer.py

@@ -9,6 +9,7 @@
 """
 
 import datetime
+import json
 import logging
 import pathlib
 import time
@@ -94,6 +95,7 @@ def my_plan(dets, n=5):
        ~create_NX_group
        ~get_sample_title
        ~get_stream_link
+       ~resolve_class_path
        ~wait_writer
        ~wait_writer_plan_stub
        ~write_data
@@ -108,6 +110,7 @@ def my_plan(dets, n=5):
        ~write_slits
        ~write_source
        ~write_streams
+       ~write_templates
        ~write_user
 
     New with apstools release 1.3.0.
@@ -123,6 +126,9 @@ def my_plan(dets, n=5):
     nxdata_signal_axes = None  # name of dataset for X axis on plot
     root = None  # instance of h5py.File
 
+    template_key = "nxwriter_template"
+    """The template (dict) is written as a JSON string to this metadata key."""
+
     _external_file_read_timeout = 20
     _external_file_read_retry_delay = 0.5
     _writer_active = False
@@ -246,6 +252,34 @@ def h5string(self, text):
         text = text or ""
         return text.encode("utf8")
 
+    def resolve_class_path(self, class_path):
+        """
+        Parse the class path, make any groups, return the HDF5 address.
+
+        New with apstools release 1.6.18.
+        """
+        addr = ""
+        for level in class_path.split("/"):
+            if ":" in level:
+                group_name, nx_class = level.split(":")
+                if not nx_class.startswith("NX"):
+                    raise ValueError(f"nx_class must start with 'NX'.  Received {nx_class=!r}")
+                if group_name not in self.root[addr]:
+                    # fmt: off
+                    logger.info(
+                        "make HDF5 group with @NX_class=%r at address '%s/%s'",
+                        nx_class, addr, group_name
+                    )
+                    # fmt: on
+                    self.create_NX_group(self.root[addr], level)
+                addr += f"/{group_name}"
+            else:
+                addr += f"/{level}"
+            addr = addr.replace("//", "/")
+
+        logger.debug("HDF5 address=%r", addr)
+        return addr
+
     def wait_writer(self):
         """
         Wait for the writer to finish.  For interactive use (Not in a plan).
@@ -414,6 +448,11 @@ def write_entry(self):
         nxentry["plan_name"] = self.root["/entry/instrument/bluesky/metadata/plan_name"]
         nxentry["entry_identifier"] = self.root["/entry/instrument/bluesky/uid"]
 
+        try:
+            self.write_templates()
+        except Exception as exc:
+            logger.warning("Problem writing template(s): %s", exc)
+
         return nxentry
 
     def write_instrument(self, parent):
@@ -753,6 +792,53 @@ def write_streams(self, parent):
 
         return bluesky
 
+    def write_templates(self):
+        """
+        Process any link templates provided as run metadata.
+
+        New in v1.6.18.
+        """
+        addr = f"/entry/instrument/bluesky/metadata/{self.template_key}"
+        if addr not in self.root:
+            return
+        templates = json.loads(self.root[addr][()])
+
+        for source, target in templates:
+            if "/@" in source:
+                p = source.rfind("/")
+                if source.find("/@") != p:  # more than one match
+                    raise ValueError(f"Only one attribute can be named.  Received: {source!r}")
+                h5addr = self.resolve_class_path(source[:p])
+                attr = source[p + 2 :]
+                logger.debug("Set attribute: group=%r attr=%r value=%r", h5addr, attr, target)
+                if h5addr in self.root:
+                    self.root[h5addr].attrs[attr] = target
+                else:
+                    logger.warning("group %r not in root %r", h5addr, self.root.name)
+            elif source.endswith("="):
+                p = source.rfind("/")
+                h5addr = self.resolve_class_path(source[:p])
+                field = source.split("/")[-1].rstrip("=")
+                if h5addr in self.root:
+                    if isinstance(target, (int, float)):
+                        target = [target]
+                    ds = self.root[h5addr].create_dataset(field, data=target)
+                    ds.attrs["target"] = ds.name
+                    # fmt: off
+                    logger.info(
+                        "Set constant field: group=%r field=%r value=%r",
+                        h5addr, field, ds[()]
+                    )
+                    # fmt: on
+                else:
+                    logger.warning("group %r not in root %r", h5addr, self.root.name)
+            elif source in self.root:
+                h5addr = self.resolve_class_path(target)
+                self.root[h5addr] = self.root[source]
+                logger.debug("Template: Linked %r to %r", source, h5addr)
+            else:
+                logger.warning("Not handled: source=%r target=%r", source, target)
+
     def write_user(self, parent):
         """
         group: /entry/contact:NXuser

apstools/callbacks/tests/test_nxwriter.py

@@ -1,3 +1,4 @@
+import json
 import pathlib
 import tempfile
 
@@ -14,12 +15,12 @@
 from ophyd.areadetector.plugins import HDF5Plugin_V34 as HDF5Plugin
 from ophyd.areadetector.plugins import ImagePlugin_V34 as ImagePlugin
 
-from ...devices import ensure_AD_plugin_primed
 from ...devices import CamMixin_V34 as CamMixin
 from ...devices import SingleTrigger_V34 as SingleTrigger
-from .. import NXWriter
-from ...tests import IOC_GP
+from ...devices import ensure_AD_plugin_primed
 from ...tests import IOC_AD
+from ...tests import IOC_GP
+from .. import NXWriter
 
 MOTOR_PV = f"{IOC_GP}m1"
 IMAGE_DIR = "adsimdet/%Y/%m/%d"
@@ -153,3 +154,58 @@ def test_NXWriter_with_RunEngine(camera, motor):
         assert signal in nxdata
         frames = nxdata[signal]
         assert frames.shape == (npoints, 1024, 1024)
+
+
+def test_NXWriter_templates(camera, motor):
+    test_file = pathlib.Path(tempfile.mkdtemp()) / "nxwriter.h5"
+    catalog = databroker.temp().v2
+
+    nxwriter = NXWriter()
+    nxwriter.file_name = str(test_file)
+    assert isinstance(nxwriter.file_name, pathlib.Path)
+    nxwriter.warn_on_missing_content = False
+
+    RE = RunEngine()
+    RE.subscribe(catalog.v1.insert)
+    RE.subscribe(nxwriter.receiver)
+
+    templates = [
+        ["/entry/_TEST:NXdata/array=", [1, 2, 3]],
+        ["/entry/_TEST/@signal", "array"],
+        ["/entry/_TEST/array", "/entry/_TEST/d123"],
+        ["/entry/_TEST/d123", "/entry/_TEST/note:NXnote/x"],
+    ]
+    md = {
+        "title": "NeXus/HDF5 template support",
+        nxwriter.template_key: json.dumps(templates),
+    }
+    npoints = 3
+    uids = RE(bp.scan([camera], motor, -0.1, 0, npoints, md=md))
+    assert isinstance(uids, (list, tuple))
+    assert len(uids) == 1
+    assert uids[-1] in catalog
+
+    nxwriter.wait_writer()
+    # time.sleep(1)  # wait just a bit longer
+
+    assert test_file.exists()
+    with h5py.File(test_file, "r") as root:
+        default = root.attrs.get("default", "entry")
+        assert default in root
+        nxentry = root[default]
+
+        assert "_TEST" in nxentry, f"{test_file=} {list(nxentry)=}"
+        nxdata = nxentry["_TEST"]
+
+        signal = nxdata.attrs.get("signal")
+        assert signal in nxdata
+        assert nxdata[signal].shape == (3,)
+
+        assert "d123" in nxdata
+        assert nxdata["d123"].attrs["target"] == nxdata[signal].attrs["target"]
+
+        assert "note" in nxdata
+        nxnote = nxdata["note"]
+
+        assert "x" in nxnote
+        assert nxnote["x"].attrs["target"] == nxdata[signal].attrs["target"]

docs/source/api/_filewriters.rst

@@ -188,6 +188,90 @@ Notes:
 1. ``detectors[0]`` will be used as the ``/entry/data@signal`` attribute
 2. the *complete* list in ``positioners`` will be used as the ``/entry/data@axes`` attribute
 
+.. index:: templates
+
+.. _filewriters.templates:
+
+Templates
+~~~~~~~~~~~~~~~~~~~~~~~~~
+
+Use templates to build NeXus base classes or application definitions with data
+already collected by the nxwriter and saved in other places in the NeXus/HDF5
+file (such as `/entry/instrument/bluesky`).  Templates are used to:
+
+* make links from existing fields or groups to new locations
+* create new groups as directed
+* create constants for attributes or fields
+
+A template is a ``[source, target]`` list, where source is a string and target
+varies depending on the type of template.  A list of templates is stored as a
+JSON string in the run's metadata under a pre-arranged key.
+
+For reasons of expediency, always use absolute HDF5 addresses. Relative
+addresses are not supported now.
+
+A *link* template makes a pointer from an existing field or group to another
+location. In a link template, the source is an existing HDF5 address.  The
+target is a NeXus class path.  If the path contains a group which is not yet
+defined, the addtional component names the NeXus class to be used. See the
+examples below.
+
+A *constant* template defines a new NeXus field.  The source string, which can be
+a class path (see above), ends with an ``=``.  The target can be a text, a
+number, or an array.  Anything that can be converted into a JSON document and
+then written to an HDF5 dataset.
+
+An *attribute* template adds a new attribute to a field or group.  Use the `@`
+symbol in the source string as shown in the examples below.  The target is the
+value of the attribute.
+
+EXAMPLE:
+
+.. code-block:: python
+    :linenos:
+
+    template_examples = [
+        # *constant* template
+        # define a constant array in a new NXdata group
+        ["/entry/example:NXdata/array=", [1, 2, 3]],
+
+        # *attribute* template
+        # set the example group "signal" attribute
+        #   (new array is the default plottable data)
+        ["/entry/example/@signal", "array"],
+
+        # *link* template
+        # link the new array into a new NXnote group as field "x"
+        ["/entry/example/array", "/entry/example/note:NXnote/x"],
+    ]
+
+    md = {
+        "title": "NeXus/HDF5 template support example",
+
+        # encode the Python dictionary as a JSON string
+        nxwriter.template_key: json.dumps(template_examples),
+    }
+
+The templates in this example add this structure to the ``/entry`` group in the
+HDF5 file:
+
+.. code-block:: text
+    :linenos:
+
+    /entry:NXentry
+        @NX_class = "NXentry"
+        ...
+        example:NXdata
+            @NX_class = "NXdata"
+            @signal = "array"
+            @target = "/entry/example"
+            array:NX_INT64[3] = [1, 2, 3]
+                @target = "/entry/example/array"
+            note:NXnote
+                @NX_class = "NXnote"
+                @target = "/entry/example/note"
+                x --> /entry/example/array
+
 NXWriterAPS
 ^^^^^^^^^^^