Add load_field and load_group to allow loading custom entries using GenericNeXusWorkflow

src/ess/reduce/nexus/workflow.py

@@ -683,23 +683,21 @@ def LoadMonitorWorkflow(
     wf = sciline.Pipeline(
         (*_common_providers, *_monitor_providers),
         constraints=_gather_constraints(
-            run_types=run_types, monitor_types=monitor_types
+            run_types=run_types, monitor_types=monitor_types, component_types=[]
         ),
     )
     wf[PreopenNeXusFile] = PreopenNeXusFile(False)
     return wf
 
 
 def LoadDetectorWorkflow(
-    *,
-    run_types: Iterable[sciline.typing.Key],
-    monitor_types: Iterable[sciline.typing.Key],
+    *, run_types: Iterable[sciline.typing.Key]
 ) -> sciline.Pipeline:
     """Generic workflow for loading detector data from a NeXus file."""
     wf = sciline.Pipeline(
         (*_common_providers, *_detector_providers),
         constraints=_gather_constraints(
-            run_types=run_types, monitor_types=monitor_types
+            run_types=run_types, monitor_types=[], component_types=[]
         ),
     )
     wf[DetectorBankSizes] = DetectorBankSizes({})
@@ -711,6 +709,7 @@ def GenericNeXusWorkflow(
     *,
     run_types: Iterable[sciline.typing.Key],
     monitor_types: Iterable[sciline.typing.Key],
+    component_types: Iterable[sciline.typing.Key] | None = None,
 ) -> sciline.Pipeline:
     """
     Generic workflow for loading detector and monitor data from a NeXus file.
@@ -735,6 +734,9 @@ def GenericNeXusWorkflow(
         List of monitor types to include in the workflow.
         Constrains the possible values of :class:`ess.reduce.nexus.types.MonitorType`
         and :class:`ess.reduce.nexus.types.Component`.
+    component_types:
+        Additional component types to include in the workflow.
+        Constrains the possible values of :class:`ess.reduce.nexus.types.Component`.
 
     Returns
     -------
@@ -750,7 +752,9 @@ def GenericNeXusWorkflow(
             *_metadata_providers,
         ),
         constraints=_gather_constraints(
-            run_types=run_types, monitor_types=monitor_types
+            run_types=run_types,
+            monitor_types=monitor_types,
+            component_types=[] if component_types is None else component_types,
         ),
     )
     wf[DetectorBankSizes] = DetectorBankSizes({})
@@ -763,11 +767,12 @@ def _gather_constraints(
     *,
     run_types: Iterable[sciline.typing.Key],
     monitor_types: Iterable[sciline.typing.Key],
+    component_types: Iterable[sciline.typing.Key],
 ) -> dict[TypeVar, Iterable[type]]:
     mon = tuple(iter(monitor_types))
     constraints = {
         RunType: run_types,
         MonitorType: mon,
-        Component: (*COMPONENT_CONSTRAINTS, *mon),
+        Component: (*COMPONENT_CONSTRAINTS, *mon, *component_types),
     }
     return constraints

tests/nexus/workflow_test.py

@@ -4,11 +4,12 @@
 from pathlib import Path
 
 import pytest
+import sciline as sl
 import scipp as sc
 import scippnexus as snx
 from scipp.testing import assert_identical
 
-from ess.reduce.nexus import compute_component_position, workflow
+from ess.reduce.nexus import compute_component_position, load_component, workflow
 from ess.reduce.nexus.types import (
     BackgroundRun,
     Beamline,
@@ -776,3 +777,53 @@ def assert_not_contains_type_arg(node: object, excluded: set[type]) -> None:
     assert not any(
         arg in excluded for arg in getattr(node, "__args__", ())
     ), f"Node {node} contains one of {excluded!r}"
+
+
+def test_generic_nexus_workflow_load_custom_component_user_affiliation(
+    loki_tutorial_sample_run_60250: Path,
+) -> None:
+    # Load a component in one of the top-level entries
+
+    class UserAffiliation(sl.Scope[RunType, str], str):
+        """User affiliation."""
+
+    def load_user_affiliation(
+        location: NeXusComponentLocationSpec[UserAffiliation, RunType],
+    ) -> UserAffiliation[RunType]:
+        return UserAffiliation[RunType](
+            load_component(location, nx_class=snx.NXuser)['affiliation']
+        )
+
+    wf = GenericNeXusWorkflow(
+        run_types=[SampleRun], monitor_types=[], component_types=[UserAffiliation]
+    )
+    wf.insert(load_user_affiliation)
+    wf[Filename[SampleRun]] = loki_tutorial_sample_run_60250
+    wf[NeXusName[UserAffiliation]] = '/entry/user_0'
+    affiliation = wf.compute(UserAffiliation[SampleRun])
+    assert affiliation == 'ESS'
+
+
+def test_generic_nexus_workflow_load_custom_component_source_name(
+    loki_tutorial_sample_run_60250: Path,
+) -> None:
+    # Load a component inside the instrument entry
+
+    class SourceName(sl.Scope[RunType, str], str):
+        """Source name."""
+
+    def load_source_name(
+        location: NeXusComponentLocationSpec[SourceName, RunType],
+    ) -> SourceName[RunType]:
+        return SourceName[RunType](
+            load_component(location, nx_class=snx.NXsource)['name']
+        )
+
+    wf = GenericNeXusWorkflow(
+        run_types=[SampleRun], monitor_types=[], component_types=[SourceName]
+    )
+    wf.insert(load_source_name)
+    wf[Filename[SampleRun]] = loki_tutorial_sample_run_60250
+    wf[NeXusName[SourceName]] = '/entry/instrument/source'
+    source_name = wf.compute(SourceName[SampleRun])
+    assert source_name == 'moderator'