198 lines
8.5 KiB
Python
198 lines
8.5 KiB
Python
"""Re-feed the GUI's retained sweeps to recompute the full stopped B-scan.
|
|
|
|
The processor retains a shorter history than the GUI. Replay preserves acquisition
|
|
keys, bounds bursts by both ring capacities, and bypasses recording and live metrics.
|
|
Only the caller decides when the processor's input ring is safe to write.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from collections.abc import Callable
|
|
import time
|
|
from typing import Protocol
|
|
|
|
from PyQt6.QtCore import QTimer
|
|
|
|
from python_app.gui.controllers.history_controller import HistoryController
|
|
from python_app.models.collection_identity import collection_key
|
|
from python_app.models.dataset_model import ResultCollection
|
|
from python_app.models.run_config_schema import RingEndpointModel
|
|
from python_app.orchestration.shm.decoder import PREPROC_MAGIC
|
|
from python_app.orchestration.shm.ring_reader import ShmRingReader
|
|
from python_app.orchestration.shm.ring_writer import ShmRingWriter
|
|
from python_app.storage.npz.serialize import serialize_trace_collection
|
|
|
|
_REPLAY_CHUNK_SIZE = 12
|
|
_REPLAY_CHUNK_TIMEOUT_S = 2.0
|
|
_REPROCESS_DEBOUNCE_MS = 300
|
|
|
|
|
|
class WarningSink(Protocol):
|
|
def __call__(
|
|
self, text: str, *, details: str | None = None, once_key: str | None = None,
|
|
) -> None: ...
|
|
|
|
|
|
class BscanReplay:
|
|
def __init__(
|
|
self, history: HistoryController, on_timeout: Callable[[], None],
|
|
warning: WarningSink, debug: Callable[[str], None],
|
|
) -> None:
|
|
self.history = history
|
|
self._warning = warning
|
|
self._debug = debug
|
|
self._writer: ShmRingWriter | None = None
|
|
self._generation = 0
|
|
self.active = False
|
|
self.timer = QTimer()
|
|
self.timer.setSingleShot(True)
|
|
self.timer.timeout.connect(on_timeout)
|
|
|
|
def schedule(self, enabled: bool) -> None:
|
|
if enabled:
|
|
if self.active:
|
|
self.discard()
|
|
self.timer.start(_REPROCESS_DEBOUNCE_MS)
|
|
|
|
def run(
|
|
self, *, window: int, revision: int, ring: RingEndpointModel | None,
|
|
reader: ShmRingReader | None, pump_events: Callable[[float], None],
|
|
continue_replay: Callable[[], bool],
|
|
) -> bool:
|
|
if self.active or reader is None:
|
|
return False
|
|
generation = self._generation
|
|
if not self._continues(generation, continue_replay):
|
|
return False
|
|
tail = list(self.history.preprocessed)[-window:]
|
|
if not tail:
|
|
return False
|
|
writer = self._ensure_writer(ring)
|
|
if writer is None or ring is None:
|
|
return False
|
|
|
|
keys = [collection_key(collection) for collection in tail]
|
|
result_keys = {collection_key(collection) for collection in self.history.results}
|
|
self._debug(
|
|
f"B-scan replay starting: window={window}, preprocessed history="
|
|
f"{len(self.history.preprocessed)}, result history={len(self.history.results)}, "
|
|
f"to re-send={len(tail)}, of those already in result history="
|
|
f"{sum(key in result_keys for key in keys)}."
|
|
)
|
|
replayed: dict[tuple[int, int], ResultCollection] = {}
|
|
replayed_keys = set(keys)
|
|
sent = 0
|
|
chunk_size = min(_REPLAY_CHUNK_SIZE, ring.capacity, reader.capacity)
|
|
# The caller suspends result polling. Event pumping may still change settings
|
|
# or close the pipeline, so each boundary checks this session remains valid.
|
|
self.active = True
|
|
try:
|
|
for start in range(0, len(tail), chunk_size):
|
|
chunk = tail[start:start + chunk_size]
|
|
pushed = 0
|
|
for collection in chunk:
|
|
if not self._continues(generation, continue_replay):
|
|
return False
|
|
payload = serialize_trace_collection(collection, PREPROC_MAGIC)
|
|
if not writer.push(payload):
|
|
self._warning(
|
|
"B-scan replay stopped: a preprocessed sweep does not fit the ring slot.",
|
|
details=f"payload={len(payload)} bytes, slot={writer.slot_size_bytes} bytes",
|
|
once_key="bscan_replay_payload_too_large",
|
|
)
|
|
break
|
|
pushed += 1
|
|
sent += pushed
|
|
waiting_for = keys[start:start + pushed]
|
|
complete = self._collect(
|
|
reader, waiting_for, replayed_keys, replayed, pump_events, generation, continue_replay, revision,
|
|
)
|
|
if not self._continues(generation, continue_replay):
|
|
return False
|
|
if not complete:
|
|
self._debug(f"B-scan replay chunk at offset {start} came back incomplete.")
|
|
if pushed != len(chunk):
|
|
break
|
|
finally:
|
|
self.active = False
|
|
|
|
# Tap and results rings drop independently: only these replayed keys belong in the
|
|
# refreshed image. One image never mixes settings, so unreturned sweeps are left out.
|
|
self.history.replay_results[:] = [replayed[key] for key in keys if key in replayed]
|
|
received = sum(key in replayed for key in keys[:sent])
|
|
if received < sent:
|
|
self._warning(
|
|
f"B-scan replay recovered {received} of {sent} re-sent sweeps; "
|
|
"the image leaves the others out.",
|
|
once_key=f"bscan_replay_incomplete_{sent}_{received}",
|
|
)
|
|
self._debug(f"B-scan replay finished: sent={sent}, recovered={received}.")
|
|
return True
|
|
|
|
def _collect(
|
|
self, reader: ShmRingReader, waiting_for: list[tuple[int, int]],
|
|
replayed_keys: set[tuple[int, int]], into: dict[tuple[int, int], ResultCollection],
|
|
pump_events: Callable[[float], None], generation: int, continue_replay: Callable[[], bool],
|
|
revision: int,
|
|
) -> bool:
|
|
deadline = time.monotonic() + _REPLAY_CHUNK_TIMEOUT_S
|
|
while any(key not in into for key in waiting_for) and time.monotonic() < deadline:
|
|
if not self._continues(generation, continue_replay):
|
|
return False
|
|
collection = reader.pop_result_collection()
|
|
if collection is None:
|
|
pump_events(0.005)
|
|
if not self._continues(generation, continue_replay):
|
|
return False
|
|
continue
|
|
if self.history.is_removed(collection):
|
|
continue
|
|
self.history.record_result(collection)
|
|
key = collection_key(collection)
|
|
# An old processor replay can return this exact sweep before the fresh
|
|
# result. Its identity alone does not acknowledge the current request.
|
|
if key in replayed_keys and collection.settings_revision == revision:
|
|
into[key] = collection
|
|
return all(key in into for key in waiting_for)
|
|
|
|
def _continues(self, generation: int, continue_replay: Callable[[], bool]) -> bool:
|
|
if generation != self._generation:
|
|
return False
|
|
if not continue_replay():
|
|
self.discard()
|
|
return False
|
|
return True
|
|
|
|
def _ensure_writer(self, ring: RingEndpointModel | None) -> ShmRingWriter | None:
|
|
if self._writer is not None:
|
|
return self._writer
|
|
# Geometry must come from the active run. Guessing from an editable profile
|
|
# could recreate the segment underneath the processor.
|
|
if ring is None:
|
|
self._warning(
|
|
"Cannot recompute the full B-scan: this GUI session did not start the "
|
|
"pipeline, so the processor's ring geometry is unknown.",
|
|
once_key="bscan_replay_no_active_run_config",
|
|
)
|
|
return None
|
|
try:
|
|
self._writer = ShmRingWriter(ring.name, ring.capacity, ring.slot_size_bytes)
|
|
except Exception as exc: # noqa: BLE001 - report boundary failures through the GUI
|
|
self._warning("Failed to open the B-scan replay ring writer", details=str(exc))
|
|
return self._writer
|
|
|
|
def discard(self) -> None:
|
|
self._generation += 1
|
|
self.timer.stop()
|
|
self.history.replay_results.clear()
|
|
|
|
def close(self) -> None:
|
|
self._generation += 1
|
|
self.timer.stop()
|
|
writer, self._writer = self._writer, None
|
|
if writer is not None:
|
|
try:
|
|
writer.close()
|
|
except Exception as exc: # noqa: BLE001 - cleanup must leave no attached writer
|
|
self._warning("Failed to close the B-scan replay ring writer", details=str(exc))
|