Add the BIFROST elastic monitor Q-map

src/ess/livedata/config/instruments/bifrost/factories.py

@@ -9,6 +9,7 @@
 import scipp as sc
 
 from ess.livedata.config import Instrument
+from ess.livedata.config.value_log import ValueLog
 from ess.livedata.config.workflow_spec import Temporality
 
 from . import specs
@@ -19,8 +20,19 @@
     BifrostQMapParams,
     DetectorRatemeterParams,
     DetectorRatemeterRegionParams,
+    ElasticMonitorQMapParams,
 )
 
+
+class DetectorTankAngleLog(ValueLog):
+    """Per-binding Sciline key for the BIFROST detector-tank rotation readback.
+
+    The elastic monitor rides the tank, so its ``depends_on`` chain runs
+    through ``detector_tank_angle_r0`` and its position is only known once the
+    live readback arrives.
+    """
+
+
 # Q-vector basis for Q-map calculations
 _Q_VECTORS = {
     'Qx': sc.vector([1, 0, 0]),
@@ -45,6 +57,12 @@ def setup_factories(instrument: Instrument) -> None:
         CutAxis2,
         CutData,
     )
+    from ess.bifrost.single_crystal import BifrostBraggPeakMonitorWorkflow
+    from ess.bifrost.single_crystal.types import (
+        IntensityQparQperp,
+        QParallelBins,
+        QPerpendicularBins,
+    )
     from ess.reduce.nexus.types import (
         Filename,
         NeXusData,
@@ -54,6 +72,7 @@ def setup_factories(instrument: Instrument) -> None:
     from ess.reduce.unwrap import LookupTableFilename
     from ess.reduce.unwrap.types import LookupTableRelativeErrorThreshold
     from ess.spectroscopy.types import (
+        ElasticMonitor,
         InstrumentAngle,
         PreopenNeXusFile,
         ProtonCharge,
@@ -90,6 +109,26 @@ def setup_factories(instrument: Instrument) -> None:
     specs.detector_ratemeter_handle.skip_instrument_contexts()
     specs.unified_detector_view_handle.skip_instrument_contexts()
 
+    # The elastic monitor rides the detector tank, so the tank angle serves it
+    # twice: as geometry, patched into the monitor's ``depends_on`` chain, and as
+    # the a4 coordinate ``group_by_rotation`` bins on. A stream carries one
+    # context key per spec, so the chain patch is the binding and
+    # ``_instrument_angle_from_tank_log`` below derives the coordinate from it.
+    # Chain-patch bindings must live at instrument scope; the sample rotation is
+    # direct-bind and stays spec-scope.
+    instrument.add_context_binding(
+        stream_name='detector_tank_angle_r0',
+        dependent_sources={'elastic_monitor'},
+        workflow_key=DetectorTankAngleLog,
+    )
+    # The plain monitor histogram is counts-over-TOA and never resolves a
+    # position, so it must not wait on the tank readback.
+    specs.monitor_handle.skip_instrument_contexts()
+    specs.elastic_monitor_qmap_handle.add_context_binding(
+        stream_name='rotation_stage',
+        workflow_key=SampleAngle[SampleRun],
+    )
+
     # Create base reduction workflow
     (
         reduction_workflow,
@@ -219,6 +258,59 @@ def _custom_elastic_qmap_workflow(
         wf[CutAxis2] = axis2
         return _make_cut_stream_processor(wf)
 
+    # Elastic monitor Q-map workflow
+    @cache
+    def _init_elastic_monitor_qmap_workflow() -> sciline.Pipeline:
+        """Initialize the elastic monitor Q-map workflow.
+
+        Geometry comes from the geometry artifact rather than the McStas
+        simulation file the Q-cut workflows use: the tank angle is patched into
+        the monitor's ``depends_on`` chain at the path the f144 stream targets,
+        and only the artifact writes the chain entry at that path. The
+        simulation file keys the same transform one level up and carries a
+        720-sample rotation scan in it, which no live readback can replace.
+        """
+        fname = instrument.nexus_file
+        with snx.File(fname) as f:
+            monitor_names = list(f['entry/instrument'][snx.NXmonitor])
+        workflow = BifrostBraggPeakMonitorWorkflow()
+        workflow[Filename[SampleRun]] = fname
+        workflow[LookupTableFilename] = lookup_table_simulation()
+        workflow[LookupTableRelativeErrorThreshold] = {
+            'detector': float('inf'),
+            **{name: float('inf') for name in monitor_names},
+        }
+        workflow[PreopenNeXusFile] = PreopenNeXusFile(True)
+        return workflow
+
+    def _instrument_angle_from_tank_log(
+        log: DetectorTankAngleLog,
+    ) -> InstrumentAngle[SampleRun]:
+        """Reuse the chain-patched tank readback as the a4 grouping coordinate."""
+        return InstrumentAngle[SampleRun](log.values)
+
+    @specs.elastic_monitor_qmap_handle.attach_factory()
+    def _elastic_monitor_qmap_workflow(
+        params: ElasticMonitorQMapParams,
+    ) -> StreamProcessorWorkflow:
+        # The map is accumulated here rather than emitted as a scalar for the
+        # dashboard's correlation histogram to bin: the full wavelength band puts
+        # a distribution of Q in every update, not a scalar, and over a
+        # multi-day run a timeseries would grow without bound where a histogram
+        # does not.
+        wf = _init_elastic_monitor_qmap_workflow().copy()
+        wf.insert(_instrument_angle_from_tank_log)
+        wf[QParallelBins] = params.q_parallel_edges.get_edges().rename(Q='Q_parallel')
+        wf[QPerpendicularBins] = params.q_perpendicular_edges.get_edges().rename(
+            Q='Q_perpendicular'
+        )
+        return StreamProcessorWorkflow(
+            wf,
+            dynamic_keys={'elastic_monitor': NeXusData[ElasticMonitor, SampleRun]},
+            target_keys={'q_map': IntensityQparQperp[SampleRun]},
+            accumulators=(IntensityQparQperp[SampleRun],),
+        )
+
 
 def _transpose_with_coords(data: sc.DataArray, dims: tuple[str, ...]) -> sc.DataArray:
     """

src/ess/livedata/config/instruments/bifrost/specs.py

@@ -242,6 +242,35 @@ class BifrostCustomElasticQMapParams(pydantic.BaseModel):
     )
 
 
+class ElasticMonitorQMapParams(pydantic.BaseModel):
+    q_parallel_edges: QEdges = pydantic.Field(
+        default=QEdges(start=-QMAX_DEFAULT, stop=QMAX_DEFAULT, num_bins=QBIN_DEFAULT),
+        description="Bin edges for Q parallel to the beam (in 1/Å).",
+    )
+    q_perpendicular_edges: QEdges = pydantic.Field(
+        default=QEdges(start=-QMAX_DEFAULT, stop=QMAX_DEFAULT, num_bins=QBIN_DEFAULT),
+        description="Bin edges for Q perpendicular to the beam (in 1/Å).",
+    )
+
+
+def _make_2d_template() -> sc.DataArray:
+    """Create an empty 2D template for 2D output data."""
+    return sc.DataArray(sc.zeros(dims=['dim_0', 'dim_1'], shape=[0, 0], unit='counts'))
+
+
+class ElasticMonitorQMapOutputs(WorkflowOutputsBase):
+    """Outputs for the elastic monitor Q-map workflow."""
+
+    q_map: sc.DataArray = pydantic.Field(
+        default_factory=_make_2d_template,
+        title='Q Map',
+        description=(
+            'Elastic intensity accumulated over a sample rotation scan, binned in '
+            'Q perpendicular vs Q parallel.'
+        ),
+    )
+
+
 class QMapOutputs(WorkflowOutputsBase):
     """Outputs for Bifrost Q-map workflows."""
 
@@ -278,7 +307,9 @@ class QMapOutputs(WorkflowOutputsBase):
 instrument_registry.register(instrument)
 
 # Register monitor workflow spec (TOA-only, no TOF lookup tables)
-register_monitor_workflow_specs(instrument, monitors, params=TOAOnlyMonitorDataParams)
+monitor_handle = register_monitor_workflow_specs(
+    instrument, monitors, params=TOAOnlyMonitorDataParams
+)
 
 
 def _logical_view(obj: sc.Variable | sc.DataArray, source_name: str) -> sc.DataArray:
@@ -391,3 +422,22 @@ def _bifrost_spectrum_transform(
     params=BifrostCustomElasticQMapParams,
     outputs=QMapOutputs,
 )
+
+# cbm5, a single-pixel monitor riding the detector tank. Over a sample rotation
+# scan it maps elastic intensity in Q, to be compared against the expected
+# reciprocal lattice. Older material -- and upstream's
+# ``BifrostBraggPeakMonitorWorkflow`` -- calls it the Bragg peak monitor.
+# The title carries a source qualifier because the workflow chooser lists bare
+# titles within a group, next to the ``elastic_qmap`` detector workflow.
+elastic_monitor_qmap_handle = instrument.register_spec(
+    name='elastic_monitor_qmap',
+    version=1,
+    title='Elastic Q map (monitor)',
+    description=(
+        'Elastic intensity from the elastic monitor, accumulated over a sample '
+        'rotation scan and binned in Q perpendicular vs Q parallel.'
+    ),
+    source_names=['elastic_monitor'],
+    params=ElasticMonitorQMapParams,
+    outputs=ElasticMonitorQMapOutputs,
+)

src/ess/livedata/preprocessors/detector_data.py

@@ -78,6 +78,15 @@ def make_preprocessor(self, key: StreamId) -> Accumulator | None:
     'geometry-loki-2026-05-08.nxs': 'md5:4edc75ba015e7916dfc23e8d78f9cca6',
     'geometry-bifrost-2025-01-01.nxs': 'md5:ae3caa99dd56de9495b9321eea4e4fef',
     'geometry-bifrost-2026-06-08.nxs': 'md5:31d0aa10243e29a14aac0655454ff205',
+    # 'repaired' marks a hand-patched stopgap, to be superseded by a clean
+    # regeneration once the writer is fixed: the BIFROST writer attaches the
+    # event-mode monitors' geometry to `<name>_backup` / `<name>_da00` NXnote
+    # siblings, while the NXmonitor's own `depends_on` points at a
+    # `transformations` group it does not have. The repair copies those
+    # transformations back onto the monitor they belong to (six added datasets,
+    # values bit-identical to the source), which is what makes the elastic
+    # monitor's chain resolvable.
+    'geometry-bifrost-repaired-2026-08-11.nxs': 'md5:4db6124e1e8d85a2e9c92d110b717736',
     'geometry-odin-2025-09-25.nxs': 'md5:5615a6203813b4ab84a191f7478ceb3c',
     'geometry-tbl-2025-12-03.nxs': 'md5:040a70659155eb386245755455ee3e62',
     'geometry-tbl-2026-07-01.nxs': 'md5:81535e5468a6907e47b97c4cb8e1fd3c',

tests/config/motion_binding_test.py

@@ -39,11 +39,16 @@ def _chain_patch_inputs(instrument: Instrument) -> list[ContextBinding]:
 def _load_chain(artifact: str, source_name: str) -> TransformationChain | None:
     """Walk a source's depends_on chain via scippnexus.
 
-    Returns ``None`` for static components with no ``depends_on`` field.
+    Returns ``None`` for static components with no ``depends_on`` field and for
+    sources that are not NeXus groups at all — BIFROST's ``unified_detector`` is
+    a logical name covering 45 triplet groups.
     """
     parent_path = f'/entry/instrument/{source_name}'
     with snx.File(artifact, 'r') as f:
-        comp = f[parent_path]
+        try:
+            comp = f[parent_path]
+        except KeyError:
+            return None
         try:
             depends_on = comp['depends_on'][()]
         except KeyError:

tests/services/data_reduction_test.py

@@ -758,3 +758,51 @@ def test_message_with_bad_timestamp_is_ignored(
     app.step()
     assert len(sink.messages) == 1
     assert sink.messages[0].value.values.sum() == 2000
+
+
+@pytest.mark.slow
+def test_bifrost_elastic_monitor_qmap_maps_once_rotation_context_arrives() -> None:
+    """The elastic monitor Q-map runs off monitor events plus a3/a4 context.
+
+    Unlike the detector Q-maps this workflow's source is a monitor
+    (``elastic_monitor``, cbm5), so it exercises the monitor route into
+    ``data_reduction``. The tank angle arrives as a chain-patch binding — it
+    places the monitor *and* supplies the a4 grouping coordinate — while the
+    sample rotation is a direct spec-scope bind, so the gate must open on both
+    devices even though the instrument-scope binds name ``unified_detector``.
+    """
+    app = make_reduction_app(instrument='bifrost')
+    sink = app.sink
+    service = app.service
+    workflow_id, _ = _get_workflow_from_registry('bifrost', 'elastic_monitor_qmap')
+
+    workflow_config = workflow_spec.WorkflowConfig(
+        identifier=workflow_id, job_id=_job_id('elastic_monitor')
+    )
+    app.publish_config_message(workflow_config)
+    service.step()
+    sink.messages.clear()
+
+    # Monitor events before any rotation context: gate holds, no crash.
+    app.publish_monitor_events(size=2000, time=2)
+    service.step()
+    assert len(sink.messages) == 0
+
+    for device in ('detector_tank_angle_r0', 'rotation_stage'):
+        for substream, value in (
+            ('target_value', 0.0),
+            ('idle_flag', 1),
+            ('value', 0.0),
+        ):
+            app.publish_log_message(
+                source_name=f'{device}/{substream}', time=4, value=value
+            )
+    service.step()
+
+    sink.messages.clear()
+    app.publish_monitor_events(size=2000, time=5)
+    service.step()
+    assert len(sink.messages) >= 1
+    result = sink.messages[-1].value
+    assert result.dims == ('Q_perpendicular', 'Q_parallel')
+    assert result.data.sum().value > 0