72 lines
3.2 KiB
Python
72 lines
3.2 KiB
Python
"""Retained acquisitions and pending processor-history deletions."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from collections import deque
|
|
from itertools import chain
|
|
import time
|
|
from typing import TypeVar
|
|
|
|
from python_app.gui.settings_defaults import GUI_SAVE_HISTORY_LIMIT
|
|
from python_app.gui.runtime.history import record_result_history
|
|
from python_app.models.collection_identity import CollectionIdentity, collection_key, history_collection_removed
|
|
from python_app.models.dataset_model import ResultCollection, SweepCollection
|
|
|
|
_Collection = TypeVar("_Collection", SweepCollection, ResultCollection)
|
|
|
|
|
|
def _remove_acquisition(history: deque[_Collection], key: tuple[int, int]) -> None:
|
|
retained = [item for item in history if collection_key(item) != key]
|
|
history.clear()
|
|
history.extend(retained)
|
|
|
|
|
|
class HistoryController:
|
|
"""Own GUI history without opening rings, starting workers, or drawing plots."""
|
|
|
|
def __init__(self, limit: int = GUI_SAVE_HISTORY_LIMIT) -> None:
|
|
self.raw: deque[SweepCollection] = deque(maxlen=limit)
|
|
self.preprocessed: deque[SweepCollection] = deque(maxlen=limit)
|
|
self.results: deque[ResultCollection] = deque(maxlen=limit)
|
|
self.replay_results: list[ResultCollection] = []
|
|
self.run_signature: tuple[object, ...] | None = None
|
|
self.clear_before_ns = 0
|
|
self.removed: list[dict[str, int]] = []
|
|
|
|
def is_removed(self, collection: CollectionIdentity) -> bool:
|
|
return history_collection_removed(collection, self.clear_before_ns, self.removed)
|
|
|
|
def record_result(self, collection: ResultCollection) -> bool:
|
|
"""Replace a recomputed sweep or insert a new one in acquisition order."""
|
|
return not self.is_removed(collection) and record_result_history(self.results, collection)
|
|
|
|
def apply_deletion(self, command: str, target: dict[str, int] | None = None) -> None:
|
|
"""Stage deletions before publishing settings; a failed write must retain them."""
|
|
if command == "clear_all":
|
|
self.clear_before_ns = time.monotonic_ns()
|
|
self.removed.clear()
|
|
elif command == "remove_last" and target is not None:
|
|
self.removed.append(dict(target))
|
|
|
|
def delete(self, *, remove_last: bool) -> None:
|
|
"""Mutate retained stages and stage their deletion before any external action."""
|
|
self.replay_results.clear()
|
|
if not remove_last:
|
|
self.apply_deletion("clear_all")
|
|
self.raw.clear()
|
|
self.preprocessed.clear()
|
|
self.results.clear()
|
|
return
|
|
newest = max(chain(self.raw, self.preprocessed, self.results), key=lambda item: item.capture_start_ns, default=None)
|
|
if newest is None:
|
|
return
|
|
key = collection_key(newest)
|
|
self.apply_deletion("remove_last", {"collection_id": key[0], "capture_start_ns": key[1]})
|
|
_remove_acquisition(self.raw, key)
|
|
_remove_acquisition(self.preprocessed, key)
|
|
_remove_acquisition(self.results, key)
|
|
|
|
def bscan_source(self) -> list[ResultCollection]:
|
|
"""A completed replay contains only the re-fed sweeps with current settings."""
|
|
return list(self.replay_results or self.results)
|