Buffer history of results in DataService

src/ess/livedata/config/workflows.py

@@ -76,7 +76,10 @@ def is_empty(self) -> bool:
     def _get_value(self) -> sc.DataArray:
         if self._to_nxlog is None:
             raise ValueError("No data accumulated")
-        return self._to_nxlog.get()
+        # Return latest value. Will be aggregated into a timeseries in frontend (if a
+        # plot requests it). This accumulator may be fully replaced once it is clear how
+        # we want to handle obtaining the full history (e.g., after frontend restarts).
+        return self._to_nxlog.get()[-1]
 
     def _do_push(self, value: sc.DataArray) -> None:
         if self._to_nxlog is None:

src/ess/livedata/core/job.py

@@ -184,7 +184,11 @@ def add(self, data: JobData) -> JobReply:
                     remapped_aux_data[field_name] = value
 
             # Pass data to workflow with field names (not stream names)
-            self._processor.accumulate({**data.primary_data, **remapped_aux_data})
+            self._processor.accumulate(
+                {**data.primary_data, **remapped_aux_data},
+                start_time=data.start_time,
+                end_time=data.end_time,
+            )
             if data.is_active():
                 if self._start_time is None:
                     self._start_time = data.start_time

src/ess/livedata/dashboard/data_service.py

@@ -2,28 +2,74 @@
 # Copyright (c) 2025 Scipp contributors (https://github.com/scipp)
 from __future__ import annotations
 
-from collections import UserDict
-from collections.abc import Callable, Hashable
+from abc import ABC, abstractmethod
+from collections.abc import Callable, Hashable, Iterator, Mapping, MutableMapping
 from contextlib import contextmanager
-from typing import TypeVar
+from typing import Any, Generic, TypeVar
 
-from .data_subscriber import DataSubscriber
+from .extractors import LatestValueExtractor, UpdateExtractor
+from .temporal_buffer_manager import TemporalBufferManager
 
 K = TypeVar('K', bound=Hashable)
 V = TypeVar('V')
 
 
-class DataService(UserDict[K, V]):
+class DataServiceSubscriber(ABC, Generic[K]):
+    """Base class for data service subscribers with cached keys and extractors."""
+
+    def __init__(self) -> None:
+        """Initialize subscriber and cache keys from extractors."""
+        # Cache keys from extractors to avoid repeated computation
+        self._keys = set(self.extractors.keys())
+
+    @property
+    def keys(self) -> set[K]:
+        """Return the set of data keys this subscriber depends on."""
+        return self._keys
+
+    @property
+    @abstractmethod
+    def extractors(self) -> Mapping[K, UpdateExtractor]:
+        """
+        Return extractors for obtaining data views.
+
+        Returns a mapping from key to the extractor to use for that key.
+        """
+
+    @abstractmethod
+    def trigger(self, store: dict[K, Any]) -> None:
+        """Trigger the subscriber with updated data."""
+
+
+class DataService(MutableMapping[K, V]):
     """
     A service for managing and retrieving data and derived data.
 
     New data is set from upstream Kafka topics. Subscribers are typically plots that
     provide a live view of the data.
+
+    Uses buffers internally for storage, but presents a dict-like interface
+    that returns the latest value for each key.
     """
 
-    def __init__(self) -> None:
-        super().__init__()
-        self._subscribers: list[DataSubscriber[K]] = []
+    def __init__(
+        self,
+        buffer_manager: TemporalBufferManager | None = None,
+    ) -> None:
+        """
+        Initialize DataService.
+
+        Parameters
+        ----------
+        buffer_manager:
+            Manager for buffer sizing. If None, creates a new TemporalBufferManager.
+        """
+        if buffer_manager is None:
+            buffer_manager = TemporalBufferManager()
+        self._buffer_manager = buffer_manager
+        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()
@@ -48,17 +94,98 @@ def transaction(self):
     def _in_transaction(self) -> bool:
         return self._transaction_depth > 0
 
-    def register_subscriber(self, subscriber: DataSubscriber[K]) -> None:
+    def _get_extractors(self, key: K) -> list[UpdateExtractor]:
         """
-        Register a subscriber for updates.
+        Collect extractors for a key from all subscribers.
+
+        Examines all subscribers that need this key.
+
+        Parameters
+        ----------
+        key:
+            The key to collect extractors for.
+
+        Returns
+        -------
+        :
+            List of extractors from all subscribers for this key.
+        """
+        extractors = []
+
+        for subscriber in self._subscribers:
+            subscriber_extractors = subscriber.extractors
+            if key in subscriber_extractors:
+                extractor = subscriber_extractors[key]
+                extractors.append(extractor)
+
+        return extractors
+
+    def _build_subscriber_data(
+        self, subscriber: DataServiceSubscriber[K]
+    ) -> dict[K, Any]:
+        """
+        Extract data for a subscriber based on its extractors.
+
+        Parameters
+        ----------
+        subscriber:
+            The subscriber to extract data for.
+
+        Returns
+        -------
+        :
+            Dictionary mapping keys to extracted data (None values filtered out).
+        """
+        subscriber_data = {}
+        extractors = subscriber.extractors
+
+        for key in subscriber.keys:
+            extractor = extractors[key]
+            buffered_data = self._buffer_manager.get_buffered_data(key)
+            if buffered_data is not None:
+                data = extractor.extract(buffered_data)
+                subscriber_data[key] = data
+
+        return subscriber_data
+
+    def register_subscriber(self, subscriber: DataServiceSubscriber[K]) -> None:
+        """
+        Register a subscriber for updates with extractor-based data access.
+
+        Triggers the subscriber immediately with existing data using its extractors.
 
         Parameters
         ----------
         subscriber:
-            The subscriber to register. Must implement the DataSubscriber interface.
+            The subscriber to register.
         """
         self._subscribers.append(subscriber)
 
+        # Add extractors for keys this subscriber needs
+        for key in subscriber.keys:
+            if key in self._buffer_manager:
+                extractor = subscriber.extractors[key]
+                self._buffer_manager.add_extractor(key, extractor)
+
+        # Trigger immediately with existing data using subscriber's extractors
+        existing_data = self._build_subscriber_data(subscriber)
+        subscriber.trigger(existing_data)
+
+    def register_update_callback(self, callback: Callable[[set[K]], None]) -> None:
+        """
+        Register a callback for key update notifications.
+
+        Callback receives only the set of updated key names, not the data.
+        Use this for infrastructure that needs to know what changed but will
+        query data itself.
+
+        Parameters
+        ----------
+        callback:
+            Callable that accepts a set of updated keys.
+        """
+        self._update_callbacks.append(callback)
+
     def subscribe_to_changed_keys(
         self, subscriber: Callable[[set[K], set[K]], None]
     ) -> None:
@@ -71,7 +198,7 @@ def subscribe_to_changed_keys(
             A callable that accepts two sets: added_keys and removed_keys.
         """
         self._key_change_subscribers.append(subscriber)
-        subscriber(set(self.data.keys()), set())
+        subscriber(set(self._buffer_manager.keys()), set())
 
     def _notify_subscribers(self, updated_keys: set[K]) -> None:
         """
@@ -82,16 +209,16 @@ def _notify_subscribers(self, updated_keys: set[K]) -> None:
         updated_keys
             The set of data keys that were updated.
         """
+        # Notify extractor-based subscribers
         for subscriber in self._subscribers:
-            if not isinstance(subscriber, DataSubscriber):
-                subscriber(updated_keys)
-                continue
             if updated_keys & subscriber.keys:
-                # Pass only the data that the subscriber is interested in
-                subscriber_data = {
-                    key: self.data[key] for key in subscriber.keys if key in self.data
-                }
-                subscriber.trigger(subscriber_data)
+                subscriber_data = self._build_subscriber_data(subscriber)
+                if 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)."""
@@ -103,19 +230,38 @@ def _notify_key_change_subscribers(self) -> None:
                 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)
+        if buffered_data is None:
+            raise KeyError(key)
+        return self._default_extractor.extract(buffered_data)
+
     def __setitem__(self, key: K, value: V) -> None:
-        if key not in self.data:
+        """Set a value, storing it in a buffer."""
+        if key not in self._buffer_manager:
             self._pending_key_additions.add(key)
-        super().__setitem__(key, value)
+            extractors = self._get_extractors(key)
+            self._buffer_manager.create_buffer(key, extractors)
+        self._buffer_manager.update_buffer(key, value)
         self._pending_updates.add(key)
         self._notify_if_not_in_transaction()
 
     def __delitem__(self, key: K) -> None:
+        """Delete a key and its buffer."""
         self._pending_key_removals.add(key)
-        super().__delitem__(key)
+        self._buffer_manager.delete_buffer(key)
         self._pending_updates.add(key)
         self._notify_if_not_in_transaction()
 
+    def __iter__(self) -> Iterator[K]:
+        """Iterate over keys."""
+        return iter(self._buffer_manager)
+
+    def __len__(self) -> int:
+        """Return the number of keys."""
+        return len(self._buffer_manager)
+
     def _notify_if_not_in_transaction(self) -> None:
         """Notify subscribers if not in a transaction."""
         if not self._in_transaction:

src/ess/livedata/dashboard/data_subscriber.py

@@ -3,10 +3,12 @@
 from __future__ import annotations
 
 from abc import ABC, abstractmethod
-from collections.abc import Hashable
+from collections.abc import Callable, Hashable, Mapping
 from typing import Any, Generic, Protocol, TypeVar
 
 from ess.livedata.config.workflow_spec import ResultKey
+from ess.livedata.dashboard.data_service import DataServiceSubscriber
+from ess.livedata.dashboard.extractors import UpdateExtractor
 
 
 class PipeBase(Protocol):
@@ -40,6 +42,7 @@ def __init__(self, data: Any) -> None:
 
 
 Key = TypeVar('Key', bound=Hashable)
+P = TypeVar('P', bound=PipeBase)
 
 
 class StreamAssembler(ABC, Generic[Key]):
@@ -85,27 +88,45 @@ def assemble(self, data: dict[Key, Any]) -> Any:
         """
 
 
-class DataSubscriber(Generic[Key]):
+class DataSubscriber(DataServiceSubscriber[Key], Generic[Key, P]):
     """Unified subscriber that uses a StreamAssembler to process data."""
 
-    def __init__(self, assembler: StreamAssembler[Key], pipe: PipeBase) -> None:
+    def __init__(
+        self,
+        assembler: StreamAssembler[Key],
+        pipe_factory: Callable[[dict[Key, Any]], P],
+        extractors: Mapping[Key, UpdateExtractor],
+    ) -> None:
         """
-        Initialize the subscriber with an assembler and pipe.
+        Initialize the subscriber with an assembler and pipe factory.
 
         Parameters
         ----------
         assembler:
             The assembler responsible for processing the data.
-        pipe:
-            The pipe to send assembled data to.
+        pipe_factory:
+            Factory function to create the pipe on first trigger.
+        extractors:
+            Mapping from keys to their UpdateExtractor instances.
         """
         self._assembler = assembler
-        self._pipe = pipe
+        self._pipe_factory = pipe_factory
+        self._pipe: P | None = None
+        self._extractors = extractors
+        # Initialize parent class to cache keys
+        super().__init__()
 
     @property
-    def keys(self) -> set[Key]:
-        """Return the set of data keys this subscriber depends on."""
-        return self._assembler.keys
+    def extractors(self) -> Mapping[Key, UpdateExtractor]:
+        """Return extractors for obtaining data views."""
+        return self._extractors
+
+    @property
+    def pipe(self) -> P:
+        """Return the pipe (must be created by first trigger)."""
+        if self._pipe is None:
+            raise RuntimeError("Pipe not yet initialized - subscriber not triggered")
+        return self._pipe
 
     def trigger(self, store: dict[Key, Any]) -> None:
         """
@@ -118,7 +139,13 @@ def trigger(self, store: dict[Key, Any]) -> None:
         """
         data = {key: store[key] for key in self.keys if key in store}
         assembled_data = self._assembler.assemble(data)
-        self._pipe.send(assembled_data)
+
+        if self._pipe is None:
+            # First trigger - create pipe with correctly extracted data
+            self._pipe = self._pipe_factory(assembled_data)
+        else:
+            # Subsequent triggers - send to existing pipe
+            self._pipe.send(assembled_data)
 
 
 class MergingStreamAssembler(StreamAssembler):