Add ROI histograms

src/beamlime/config/raw_detectors/dream.py

@@ -8,7 +8,7 @@
 import scipp as sc
 from ess.reduce.live import raw
 
-_res_scale = 8
+_res_scale = 12
 
 dream_detectors_config = {
     'dashboard': {'nrow': 3, 'ncol': 2},
@@ -25,6 +25,11 @@
             'projection': 'xy_plane',
             'gridspec': (0, 1),
         },
+        'High-Res': {
+            'detector_name': 'high_resolution_detector',
+            'resolution': {'y': 20 * _res_scale, 'x': 20 * _res_scale},
+            'projection': 'xy_plane',
+        },
         # We use the arc length instead of phi as it makes it easier to get a correct
         # aspect ratio for the plot if both axes have the same unit.
         'mantle_projection': {

src/beamlime/core/handler.py

@@ -178,6 +178,7 @@ def __init__(
             default={'value': 1.0, 'unit': 's'},
             convert=lambda raw: int(sc.scalar(**raw).to(unit='ns', copy=False).value),
         )
+        self._logger.info('Setup handler with %s accumulators', len(accumulators))
 
     def handle(self, messages: list[Message[T]]) -> list[Message[V]]:
         # Note an issue here: We preprocess all messages, with a range of timestamps.

src/beamlime/handlers/accumulators.py

@@ -1,12 +1,15 @@
 # SPDX-License-Identifier: BSD-3-Clause
-# Copyright (c) 2024 Scipp contributors (https://github.com/scipp)
+# Copyright (c) 2025 Scipp contributors (https://github.com/scipp)
 from __future__ import annotations
 
 from collections.abc import Sequence
 from dataclasses import dataclass
+from typing import Any
 
 import numpy as np
 import scipp as sc
+from ess.reduce.live.roi import ROIFilter
+from scipp._scipp.core import _bins_no_validate
 from streaming_data_types import eventdata_ev44
 
 from ..core.handler import Accumulator, Config, ConfigValueAccessor
@@ -55,6 +58,17 @@ def from_ev44(ev44: eventdata_ev44.EventData) -> DetectorEvents:
         )
 
 
+class NullAccumulator(Accumulator[Any, None]):
+    def add(self, timestamp: int, data: Any) -> None:
+        pass
+
+    def get(self) -> None:
+        return None
+
+    def clear(self) -> None:
+        pass
+
+
 class Cumulative(Accumulator[sc.DataArray, sc.DataArray]):
     def __init__(self, config: Config, clear_on_get: bool = False):
         self._config = config
@@ -117,34 +131,118 @@ def _cleanup(self) -> None:
         self._chunks = [c for c in self._chunks if latest - c.timestamp <= max_age]
 
 
-class PixelIDMerger(Accumulator[DetectorEvents, np.array]):
-    def __init__(self, config: Config):
+class GroupIntoPixels(Accumulator[DetectorEvents, sc.DataArray]):
+    def __init__(self, config: Config, detector_number: sc.Variable):
         self._config = config
-        self._chunks: list[np.ndarray] = []
-        # TODO This will be set via a config option, in other words, the control topic
-        # by the user. This mechanism is currently not implemented.
-        self._toa_range: tuple[float, float] | None = None
-
-    def _filter_by_toa(self, toa: np.ndarray, pixel_id: np.ndarray) -> np.ndarray:
-        if self._toa_range is None:
-            return pixel_id
-        else:
-            start, end = self._toa_range
-            mask = (toa >= start) & (toa < end)
-            return pixel_id[mask]
+        self._chunks: list[DetectorEvents] = []
+        self._toa_unit = 'ns'
+        self._sizes = detector_number.sizes
+        self._dim = 'detector_number'
+        self._groups = detector_number.flatten(to=self._dim)
 
     def add(self, timestamp: int, data: DetectorEvents) -> None:
         # timestamp in function signature is required for compliance with Accumulator
         # interface.
         _ = timestamp
-        ids = self._filter_by_toa(data.time_of_arrival, data.pixel_id)
-        self._chunks.append(ids)
+        # We could easily support other units, but ev44 is always in ns so this should
+        # never happen.
+        if data.unit != self._toa_unit:
+            raise ValueError(f"Expected unit '{self._toa_unit}', got '{data.unit}'")
+        self._chunks.append(data)
 
-    def get(self) -> np.array:
-        # Could optimize the concatenate by reusing a buffer.
-        events = np.concatenate(self._chunks or [[]])
+    def get(self) -> sc.DataArray:
+        # Could optimize the concatenate by reusing a buffer (directly write to it in
+        # self.add).
+        pixel_ids = np.concatenate([c.pixel_id for c in self._chunks])
+        time_of_arrival = np.concatenate([c.time_of_arrival for c in self._chunks])
+        da = sc.DataArray(
+            data=sc.array(dims=['event'], values=time_of_arrival, unit=self._toa_unit),
+            coords={self._dim: sc.array(dims=['event'], values=pixel_ids, unit=None)},
+        )
+        self._chunks.clear()
+        return da.group(self._groups).fold(dim=self._dim, sizes=self._sizes)
+
+    def clear(self) -> None:
+        self._chunks.clear()
+
+
+class ROIBasedTOAHistogram(Accumulator[sc.DataArray, sc.DataArray]):
+    def __init__(self, config: Config, roi_filter: ROIFilter):
+        self._config = config
+        self._roi_filter = roi_filter
+        self._chunks: list[sc.DataArray] = []
+
+        # Note: Currently we are using the same ROI config values for all detector
+        # handlers ands views. This is for demo purposes and will be replaced by a more
+        # flexible configuration in the future.
+        self._roi_x = ConfigValueAccessor(
+            config=config,
+            key='roi_x',
+            default={'min': 0.0, 'max': 0.0},
+            convert=lambda raw: (raw['min'], raw['max']),
+        )
+        self._roi_y = ConfigValueAccessor(
+            config=config,
+            key='roi_y',
+            default={'min': 0.0, 'max': 0.0},
+            convert=lambda raw: (raw['min'], raw['max']),
+        )
+        self._nbin = -1
+        self._edges: sc.Variable | None = None
+        self._edges_ns: sc.Variable | None = None
+        self._current_roi = None
+
+    def _check_for_config_updates(self) -> None:
+        nbin = self._config.get('time_of_arrival_bins', 100)
+        if self._edges is None or nbin != self._nbin:
+            self._nbin = nbin
+            self._edges = sc.linspace(
+                'time_of_arrival', 0.0, 1000 / 14, num=nbin, unit='ms'
+            )
+            self._edges_ns = self._edges.to(unit='ns')
+            self.clear()
+
+        # Access to protected variables should hopefully be avoided by changing the
+        # config values to send indices instead of percentages, once we have per-view
+        # configuration.
+        y, x = self._roi_filter._indices.dims
+        sizes = self._roi_filter._indices.sizes
+        y_min, y_max = self._roi_y()
+        x_min, x_max = self._roi_x()
+        # Convert fraction to indices
+        y_indices = (int(y_min * (sizes[y] - 1)), int(y_max * (sizes[y] - 1)))
+        x_indices = (int(x_min * (sizes[x] - 1)), int(x_max * (sizes[x] - 1)))
+        new_roi = {y: y_indices, x: x_indices}
+
+        if new_roi != self._current_roi:
+            self._current_roi = new_roi
+            self._roi_filter.set_roi_from_intervals(sc.DataGroup(new_roi))
+            self.clear()  # Clear accumulated data when ROI changes
+
+    def _add_weights(self, data: sc.DataArray) -> None:
+        constituents = data.bins.constituents
+        content = constituents['data']
+        content.coords['time_of_arrival'] = content.data
+        content.data = sc.ones(
+            dims=content.dims, shape=content.shape, dtype='float32', unit='counts'
+        )
+        data.data = _bins_no_validate(**constituents)
+
+    def add(self, timestamp: int, data: sc.DataArray) -> None:
+        self._check_for_config_updates()
+        # Note that the preprocessor does *not* add weights of 1 (unlike NeXus loaders).
+        # Instead, the data column of the content corresponds to the time of arrival.
+        filtered, scale = self._roi_filter.apply(data)
+        self._add_weights(filtered)
+        filtered *= scale
+        chunk = filtered.hist(time_of_arrival=self._edges_ns, dim=filtered.dim)
+        self._chunks.append(chunk)
+
+    def get(self) -> sc.DataArray:
+        da = sc.reduce(self._chunks).sum()
         self._chunks.clear()
-        return events
+        da.coords['time_of_arrival'] = self._edges
+        return da
 
     def clear(self) -> None:
         self._chunks.clear()

src/beamlime/handlers/detector_data_handler.py

@@ -7,7 +7,6 @@
 import re
 from typing import Any
 
-import numpy as np
 import scipp as sc
 from ess.reduce.live import raw
 
@@ -25,7 +24,12 @@
     PeriodicAccumulatingHandler,
 )
 from ..core.message import MessageKey
-from .accumulators import DetectorEvents, PixelIDMerger
+from .accumulators import (
+    DetectorEvents,
+    GroupIntoPixels,
+    NullAccumulator,
+    ROIBasedTOAHistogram,
+)
 
 detector_registry = {
     'dummy': dummy_detectors_config,
@@ -104,7 +108,19 @@ def make_handler(self, key: MessageKey) -> Handler[DetectorEvents, sc.DataArray]
             f'sliding_{name}': DetectorCounts(config=self._config, detector_view=view)
             for name, view in views.items()
         }
-        preprocessor = PixelIDMerger(config=self._config)
+        detector_number: sc.Variable | None = None
+        for name, view in views.items():
+            accumulators[f'{name}_ROI'] = ROIBasedTOAHistogram(
+                config=self._config, roi_filter=view.make_roi_filter()
+            )
+            detector_number = view.detector_number
+        if detector_number is None:
+            preprocessor = NullAccumulator()
+        else:
+            preprocessor = GroupIntoPixels(
+                config=self._config, detector_number=detector_number
+            )
+
         return PeriodicAccumulatingHandler(
             logger=self._logger,
             config=self._config,
@@ -113,16 +129,16 @@ def make_handler(self, key: MessageKey) -> Handler[DetectorEvents, sc.DataArray]
         )
 
 
-class DetectorCounts(Accumulator[np.ndarray, sc.DataArray]):
+class DetectorCounts(Accumulator[sc.DataArray, sc.DataArray]):
     """Accumulator for detector counts, based on a rolling detector view."""
 
     def __init__(self, config: Config, detector_view: raw.RollingDetectorView):
         self._config = config
         self._det = detector_view
 
-    def add(self, timestamp: int, data: np.ndarray) -> None:
+    def add(self, timestamp: int, data: sc.DataArray) -> None:
         _ = timestamp
-        self._det.add_counts(data)
+        self._det.add_events(data)
 
     def get(self) -> sc.DataArray:
         return self._det.get()

src/beamlime/services/dashboard.py

@@ -139,6 +139,24 @@ def _setup_layout(self) -> None:
                 marks={i: str(i) for i in range(0, 1001, 100)},
                 disabled=True,
             ),
+            html.Label('ROI X-axis (%)'),
+            dcc.RangeSlider(
+                id='roi-x',
+                min=0,
+                max=100,
+                step=1,
+                value=[45, 55],
+                marks={i: str(i) for i in range(0, 101, 20)},
+            ),
+            html.Label('ROI Y-axis (%)'),
+            dcc.RangeSlider(
+                id='roi-y',
+                min=0,
+                max=100,
+                step=1,
+                value=[45, 55],
+                marks={i: str(i) for i in range(0, 101, 20)},
+            ),
             html.Button('Clear', id='clear-button', n_clicks=0),
         ]
         self._app.layout = html.Div(
@@ -179,6 +197,45 @@ def _setup_callbacks(self) -> None:
             Output('clear-button', 'n_clicks'), Input('clear-button', 'n_clicks')
         )(self.clear_data)
 
+        self._app.callback(
+            Output('roi-x', 'value'),
+            Output('roi-y', 'value'),
+            Input('roi-x', 'value'),
+            Input('roi-y', 'value'),
+        )(self.update_roi)
+
+    def update_roi(self, roi_x, roi_y):
+        if roi_x is not None:
+            self._config_service.update_config(
+                'roi_x', {'min': roi_x[0] / 100, 'max': roi_x[1] / 100}
+            )
+        if roi_y is not None:
+            self._config_service.update_config(
+                'roi_y', {'min': roi_y[0] / 100, 'max': roi_y[1] / 100}
+            )
+
+        # Update ROI rectangles in all 2D detector plots
+        for fig in self._detector_plots.values():
+            if hasattr(fig.data[0], 'z'):  # Check if it's a 2D plot
+                x_range = [fig.data[0].x[0], fig.data[0].x[-1]]
+                y_range = [fig.data[0].y[0], fig.data[0].y[-1]]
+                x_min = x_range[0] + (x_range[1] - x_range[0]) * roi_x[0] / 100
+                x_max = x_range[0] + (x_range[1] - x_range[0]) * roi_x[1] / 100
+                y_min = y_range[0] + (y_range[1] - y_range[0]) * roi_y[0] / 100
+                y_max = y_range[0] + (y_range[1] - y_range[0]) * roi_y[1] / 100
+
+                fig.update_shapes(
+                    {
+                        'x0': x_min,
+                        'x1': x_max,
+                        'y0': y_min,
+                        'y1': y_max,
+                        'visible': True,
+                    }
+                )
+
+        return roi_x, roi_y
+
     @staticmethod
     def create_monitor_plot(key: str, data: sc.DataArray) -> go.Figure:
         fig = go.Figure()
@@ -207,19 +264,41 @@ def create_detector_plot(key: str, data: sc.DataArray) -> go.Figure:
             y=[],  # Will be filled with coordinate values
             colorscale='Viridis',
         )
+        # Add ROI rectangle (initially hidden)
+        fig.add_shape(
+            type="rect",
+            x0=0,
+            y0=0,
+            x1=1,
+            y1=1,
+            line={'color': 'red', 'width': 2},
+            fillcolor="red",
+            opacity=0.2,
+            visible=False,
+            name="ROI",
+        )
 
         def maybe_unit(dim: str) -> str:
             unit = data.coords[dim].unit
             return f' [{unit}]' if unit is not None else ''
 
-        fig.update_layout(
-            title=key,
-            width=500,
-            height=400,
-            xaxis_title=f'{x_dim}{maybe_unit(x_dim)}',
-            yaxis_title=f'{y_dim}{maybe_unit(y_dim)}',
-            uirevision=key,
-        )
+        size = 800
+        opts = {
+            'title': key,
+            'xaxis_title': f'{x_dim}{maybe_unit(x_dim)}',
+            'yaxis_title': f'{y_dim}{maybe_unit(y_dim)}',
+            'uirevision': key,
+            'showlegend': False,
+        }
+        y_size, x_size = data.shape
+        if y_size < x_size:
+            fig.update_layout(width=size, **opts)
+            fig.update_yaxes(scaleanchor="x", scaleratio=1, constrain="domain")
+            fig.update_xaxes(constrain="domain")
+        else:
+            fig.update_layout(height=size, **opts)
+            fig.update_xaxes(scaleanchor="y", scaleratio=1, constrain="domain")
+            fig.update_yaxes(constrain="domain")
         return fig
 
     def update_plots(self, n: int | None):

src/beamlime/services/fake_detectors.py

@@ -34,6 +34,7 @@
         'mantle_detector': (229377, 720896),
         'endcap_backward_detector': (71618, 229376),
         'endcap_forward_detector': (1, 71680),
+        'high_resolution_detector': (1122337, 1523680),  # Note: Not consecutive!
     },
     'loki': {
         'loki_detector_0': (1, 802816),