Use ScippNeutron's metadata utilities

src/ess/reduce/nexus/_nexus_loader.py

@@ -8,14 +8,16 @@
 from contextlib import AbstractContextManager, contextmanager, nullcontext
 from dataclasses import dataclass
 from math import prod
-from typing import cast
+from typing import TypeVar, cast
 
 import scipp as sc
 import scippnexus as snx
 
 from ..logging import get_logger
 from .types import (
+    Beamline,
     FilePath,
+    Measurement,
     NeXusAllLocationSpec,
     NeXusEntryName,
     NeXusFile,
@@ -27,6 +29,8 @@
 class NoNewDefinitionsType: ...
 
 
+_Model = TypeVar('_Model', Beamline, Measurement)
+
 NoNewDefinitions = NoNewDefinitionsType()
 
 
@@ -66,6 +70,18 @@ def load_all_components(
     return components
 
 
+def load_metadata(
+    file_path: FilePath | NeXusFile | NeXusGroup,
+    model: type[_Model],
+    *,
+    entry_name: NeXusEntryName | None = None,
+    definitions: Mapping | NoNewDefinitionsType = NoNewDefinitions,
+) -> _Model:
+    with _open_nexus_file(file_path, definitions=definitions) as f:
+        entry = _unique_child_group(f, snx.NXentry, entry_name)
+        return model.from_nexus_entry(entry)
+
+
 def compute_component_position(dg: sc.DataGroup) -> sc.DataGroup:
     # In some downstream packages we use some of the Nexus components which attempt
     # to compute positions without having actual Nexus data defining depends_on chains.

src/ess/reduce/nexus/types.py

@@ -7,6 +7,7 @@
 import sciline
 import scipp as sc
 import scippnexus as snx
+from scippneutron import meta as scn_meta
 
 FilePath = NewType('FilePath', Path)
 """Full path to a NeXus file on disk."""
@@ -177,6 +178,9 @@ class TransmissionRun(Generic[ScatteringRunType]):
 UniqueComponent = TypeVar('UniqueComponent', snx.NXsample, snx.NXsource)
 """Components that can be identified by their type as there will only be one."""
 
+Beamline = scn_meta.Beamline
+Measurement = scn_meta.Measurement
+
 
 class NeXusName(sciline.Scope[Component, str], str):
     """Name of a component in a NeXus file."""

src/ess/reduce/nexus/workflow.py

@@ -19,6 +19,7 @@
 from .types import (
     AllNeXusComponents,
     Analyzers,
+    Beamline,
     CalibratedBeamline,
     CalibratedDetector,
     CalibratedMonitor,
@@ -29,6 +30,7 @@
     DetectorPositionOffset,
     Filename,
     GravityVector,
+    Measurement,
     MonitorData,
     MonitorPositionOffset,
     MonitorType,
@@ -45,6 +47,7 @@
     Position,
     PreopenNeXusFile,
     RunType,
+    SampleRun,
     TimeInterval,
     UniqueComponent,
 )
@@ -586,6 +589,18 @@ def _add_variances(da: sc.DataArray) -> sc.DataArray:
     return out
 
 
+def load_beamline_metadata_from_nexus(file_spec: NeXusFileSpec[SampleRun]) -> Beamline:
+    """Load beamline metadata from a sample NeXus file."""
+    return nexus.load_metadata(file_spec.value, Beamline)
+
+
+def load_measurement_metadata_from_nexus(
+    file_spec: NeXusFileSpec[SampleRun],
+) -> Measurement:
+    """Load measurement metadata from a sample NeXus file."""
+    return nexus.load_metadata(file_spec.value, Measurement)
+
+
 definitions = snx.base_definitions()
 definitions["NXdetector"] = _StrippedDetector
 definitions["NXmonitor"] = _StrippedMonitor
@@ -631,6 +646,11 @@ def _add_variances(da: sc.DataArray) -> sc.DataArray:
 
 _analyzer_providers = (parse_analyzers,)
 
+_metadata_providers = (
+    load_beamline_metadata_from_nexus,
+    load_measurement_metadata_from_nexus,
+)
+
 
 def LoadMonitorWorkflow() -> sciline.Pipeline:
     """Generic workflow for loading monitor data from a NeXus file."""
@@ -689,6 +709,7 @@ def GenericNeXusWorkflow(
             *_detector_providers,
             *_chopper_providers,
             *_analyzer_providers,
+            *_metadata_providers,
         )
     )
     wf[DetectorBankSizes] = DetectorBankSizes({})

tests/nexus/workflow_test.py

@@ -1,5 +1,7 @@
 # SPDX-License-Identifier: BSD-3-Clause
 # Copyright (c) 2024 Scipp contributors (https://github.com/scipp)
+from datetime import datetime, timezone
+
 import pytest
 import scipp as sc
 import scippnexus as snx
@@ -10,10 +12,12 @@
 from ess.reduce.nexus.types import (
     Analyzers,
     BackgroundRun,
+    Beamline,
     Choppers,
     DetectorData,
     EmptyBeamRun,
     Filename,
+    Measurement,
     Monitor1,
     Monitor2,
     Monitor3,
@@ -578,6 +582,32 @@ def test_generic_nexus_workflow_load_analyzers() -> None:
     assert analyzer['usage'] == 'Bragg'
 
 
+def test_generic_nexus_workflow_load_beamline_metadata() -> None:
+    wf = GenericNeXusWorkflow()
+    wf[Filename[SampleRun]] = data.bifrost_simulated_elastic()
+    beamline = wf.compute(Beamline)
+
+    assert beamline.name == 'BIFROST'
+    assert beamline.facility == 'ESS'
+    assert beamline.site is None
+
+
+def test_generic_nexus_workflow_load_measurement_metadata() -> None:
+    wf = GenericNeXusWorkflow()
+    wf[Filename[SampleRun]] = data.loki_tutorial_sample_run_60250()
+    wf[Filename[BackgroundRun]] = data.loki_tutorial_background_run_60248()
+    measurement = wf.compute(Measurement)
+
+    assert measurement.title == 'My experiment'
+    assert measurement.experiment_id == 'p1234'
+    assert measurement.start_time == datetime(
+        2022, 2, 28, 21, 15, 0, tzinfo=timezone.utc
+    )
+    assert measurement.end_time == datetime(2032, 2, 29, 9, 15, 0, tzinfo=timezone.utc)
+    assert measurement.run_number is None
+    assert measurement.experiment_doi is None
+
+
 def test_generic_nexus_workflow_includes_only_given_run_and_monitor_types() -> None:
     wf = GenericNeXusWorkflow(run_types=[SampleRun], monitor_types=[Monitor1, Monitor3])
     graph = wf.underlying_graph