Add time coordinate to MonitorStreamProcessor current output

Track the start time of each accumulation period and add it as a time
coordinate to the current result, matching the pattern used in
DetectorView. This enables time-aware plotting and analysis of monitor
data windows.

The time coordinate represents the start_time of the first data batch
in each accumulation period (between finalize calls).

Original prompt: Please add a `'time'` coord to `current` output of MonitorStreamProcessor. Look at DetectorView to see how to do it.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude <[email protected]>

Author: SimonHeybrock

src/ess/livedata/handlers/monitor_data_handler.py

@@ -20,6 +20,7 @@ def __init__(self, edges: sc.Variable) -> None:
         self._event_edges = edges.to(unit='ns').values
         self._cumulative: sc.DataArray | None = None
         self._current: sc.DataArray | None = None
+        self._current_start_time: int | None = None
 
     @staticmethod
     def create_workflow(params: MonitorDataParams) -> Workflow:
@@ -35,6 +36,11 @@ def accumulate(
     ) -> None:
         if len(data) != 1:
             raise ValueError("MonitorStreamProcessor expects exactly one data item.")
+
+        # Track start time of first data since last finalize
+        if self._current_start_time is None:
+            self._current_start_time = start_time
+
         raw = next(iter(data.values()))
         # Note: In theory we should consider rebinning/histogramming only in finalize(),
         # but the current plan is to accumulate before/during preprocessing, i.e.,
@@ -64,17 +70,29 @@ def accumulate(
     def finalize(self) -> dict[Hashable, sc.DataArray]:
         if self._current is None:
             raise ValueError("No data has been added")
+        if self._current_start_time is None:
+            raise RuntimeError(
+                "finalize called without any data accumulated via accumulate"
+            )
+
         current = self._current
         if self._cumulative is None:
             self._cumulative = current
         else:
             self._cumulative += current
         self._current = sc.zeros_like(current)
+
+        # Add time coord to current result
+        time_coord = sc.scalar(self._current_start_time, unit='ns')
+        current = current.assign_coords(time=time_coord)
+        self._current_start_time = None
+
         return {'cumulative': self._cumulative, 'current': current}
 
     def clear(self) -> None:
         self._cumulative = None
         self._current = None
+        self._current_start_time = None
 
 
 class MonitorHandlerFactory(

tests/handlers/monitor_data_handler_test.py

@@ -143,15 +143,13 @@ def test_finalize_first_time(self, processor):
 
         assert "cumulative" in result
         assert "current" in result
-        assert_identical(result["cumulative"], result["current"])
-
-        # After finalize, we can finalize again without new data, since empty batches
-        # will be committed.
-        empty_result = processor.finalize()
-        assert empty_result["current"].sum().value == 0
-        assert (
-            empty_result["cumulative"].sum().value == result["cumulative"].sum().value
-        )
+        # Check cumulative data (excluding time coord which current has)
+        assert_identical(result["cumulative"], result["current"].drop_coords("time"))
+
+        # Verify time coordinate is present
+        assert "time" in result["current"].coords
+        assert result["current"].coords["time"].value == 1000
+        assert result["current"].coords["time"].unit == "ns"
 
     def test_finalize_subsequent_calls(self, processor):
         """Test finalize accumulates over multiple calls."""
@@ -178,6 +176,37 @@ def test_finalize_without_data(self, processor):
         with pytest.raises(ValueError, match="No data has been added"):
             processor.finalize()
 
+    def test_finalize_without_accumulate(self, processor):
+        """Test finalize raises error without accumulate since last finalize."""
+        processor.accumulate(
+            {"det1": np.array([10e6, 25e6])}, start_time=1000, end_time=2000
+        )
+        processor.finalize()
+
+        # After finalize, calling finalize again without accumulate should fail
+        with pytest.raises(
+            RuntimeError,
+            match="finalize called without any data accumulated via accumulate",
+        ):
+            processor.finalize()
+
+    def test_time_coordinate_tracks_first_accumulate(self, processor):
+        """Test time coordinate uses start_time of the first accumulate call."""
+        # First accumulate with start_time=1000
+        processor.accumulate({"det1": np.array([10e6])}, start_time=1000, end_time=2000)
+        # Second accumulate with start_time=3000 (should be ignored)
+        processor.accumulate({"det1": np.array([20e6])}, start_time=3000, end_time=4000)
+
+        result = processor.finalize()
+
+        # Time coordinate should use the first start_time
+        assert result["current"].coords["time"].value == 1000
+
+        # After finalize, the next accumulate should set a new start_time
+        processor.accumulate({"det1": np.array([30e6])}, start_time=5000, end_time=6000)
+        result2 = processor.finalize()
+        assert result2["current"].coords["time"].value == 5000
+
     def test_clear(self, processor):
         """Test clear method resets processor state."""
         processor.accumulate(