Buffer history of results in DataService

src/ess/livedata/dashboard/correlation_histogram.py

@@ -392,11 +392,25 @@ def add_correlation_processor(
         items: dict[ResultKey, sc.DataArray],
     ) -> None:
         """Add a correlation histogram processor with DataService subscription."""
+        from .extractors import FullHistoryExtractor
+
         self._processors.append(processor)
 
-        # Create subscriber that merges data and sends to processor
+        # Create subscriber that merges data and sends to processor.
+        # Use FullHistoryExtractor to get complete timeseries history needed for
+        # correlation histogram computation.
+        # TODO We should update the plotter to operate more efficiently by simply
+        # subscribing to the changes. This will likely require a new extractor type as
+        # well as changes in the plotter, so we defer this for now.
         assembler = MergingStreamAssembler(set(items))
-        subscriber = DataSubscriber(assembler, processor)
+        extractors = {key: FullHistoryExtractor() for key in items}
+
+        # Create factory that sends initial data to processor and returns it
+        def processor_pipe_factory(data: dict[ResultKey, sc.DataArray]):
+            processor.send(data)
+            return processor
+
+        subscriber = DataSubscriber(assembler, processor_pipe_factory, extractors)
         self._data_service.register_subscriber(subscriber)
 
     def get_timeseries(self) -> list[ResultKey]:
@@ -412,7 +426,23 @@ def create_2d_config(self) -> CorrelationHistogramConfigurationAdapter:
 
 
 def _is_timeseries(da: sc.DataArray) -> bool:
-    return da.dims == ('time',) and 'time' in da.coords
+    """Check if data represents a timeseries.
+
+    When DataService uses LatestValueExtractor (default), it returns the latest value
+    from a timeseries buffer as a 0D scalar with a time coordinate. This function
+    identifies such values as originating from a timeseries.
+
+    Parameters
+    ----------
+    da:
+        DataArray to check.
+
+    Returns
+    -------
+    :
+        True if the data is a 0D scalar with a time coordinate.
+    """
+    return da.ndim == 0 and 'time' in da.coords
 
 
 class CorrelationHistogramProcessor:

src/ess/livedata/dashboard/data_service.py

@@ -70,10 +70,7 @@ def __init__(
         self._default_extractor = LatestValueExtractor()
         self._subscribers: list[DataServiceSubscriber[K]] = []
         self._update_callbacks: list[Callable[[set[K]], None]] = []
-        self._key_change_subscribers: list[Callable[[set[K], set[K]], None]] = []
         self._pending_updates: set[K] = set()
-        self._pending_key_additions: set[K] = set()
-        self._pending_key_removals: set[K] = set()
         self._transaction_depth = 0
 
     @contextmanager
@@ -137,14 +134,11 @@ def _build_subscriber_data(
             Dictionary mapping keys to extracted data (None values filtered out).
         """
         subscriber_data = {}
-        extractors = subscriber.extractors
 
-        for key in subscriber.keys:
-            extractor = extractors[key]
+        for key, extractor in subscriber.extractors.items():
             buffered_data = self._buffer_manager.get_buffered_data(key)
             if buffered_data is not None:
-                data = extractor.extract(buffered_data)
-                subscriber_data[key] = data
+                subscriber_data[key] = extractor.extract(buffered_data)
 
         return subscriber_data
 
@@ -186,20 +180,6 @@ def register_update_callback(self, callback: Callable[[set[K]], None]) -> None:
         """
         self._update_callbacks.append(callback)
 
-    def subscribe_to_changed_keys(
-        self, subscriber: Callable[[set[K], set[K]], None]
-    ) -> None:
-        """
-        Register a subscriber for key change updates (additions/removals).
-
-        Parameters
-        ----------
-        subscriber:
-            A callable that accepts two sets: added_keys and removed_keys.
-        """
-        self._key_change_subscribers.append(subscriber)
-        subscriber(set(self._buffer_manager.keys()), set())
-
     def _notify_subscribers(self, updated_keys: set[K]) -> None:
         """
         Notify relevant subscribers about data updates.
@@ -213,23 +193,12 @@ def _notify_subscribers(self, updated_keys: set[K]) -> None:
         for subscriber in self._subscribers:
             if updated_keys & subscriber.keys:
                 subscriber_data = self._build_subscriber_data(subscriber)
-                if subscriber_data:
-                    subscriber.trigger(subscriber_data)
+                subscriber.trigger(subscriber_data)
 
         # Notify update callbacks with just key names
         for callback in self._update_callbacks:
             callback(updated_keys)
 
-    def _notify_key_change_subscribers(self) -> None:
-        """Notify subscribers about key changes (additions/removals)."""
-        if not self._pending_key_additions and not self._pending_key_removals:
-            return
-
-        for subscriber in self._key_change_subscribers:
-            subscriber(
-                self._pending_key_additions.copy(), self._pending_key_removals.copy()
-            )
-
     def __getitem__(self, key: K) -> V:
         """Get the latest value for a key."""
         buffered_data = self._buffer_manager.get_buffered_data(key)
@@ -240,7 +209,6 @@ def __getitem__(self, key: K) -> V:
     def __setitem__(self, key: K, value: V) -> None:
         """Set a value, storing it in a buffer."""
         if key not in self._buffer_manager:
-            self._pending_key_additions.add(key)
             extractors = self._get_extractors(key)
             self._buffer_manager.create_buffer(key, extractors)
         self._buffer_manager.update_buffer(key, value)
@@ -249,7 +217,6 @@ def __setitem__(self, key: K, value: V) -> None:
 
     def __delitem__(self, key: K) -> None:
         """Delete a key and its buffer."""
-        self._pending_key_removals.add(key)
         self._buffer_manager.delete_buffer(key)
         self._pending_updates.add(key)
         self._notify_if_not_in_transaction()
@@ -273,6 +240,3 @@ def _notify(self) -> None:
             pending = set(self._pending_updates)
             self._pending_updates.clear()
             self._notify_subscribers(pending)
-        self._notify_key_change_subscribers()
-        self._pending_key_additions.clear()
-        self._pending_key_removals.clear()