summaryrefslogtreecommitdiff
path: root/core
diff options
context:
space:
mode:
authorChristian Kolset <ckolset@colostate.edu>2026-08-03 15:32:56 -0600
committerChristian Kolset <ckolset@colostate.edu>2026-08-03 15:32:56 -0600
commit9bcad5d3af258eb72bc9c8ac87e3f1713e1b8b80 (patch)
treebd2b07ae968e3b422c536fab21f57a5b660d4b98 /core
parent68a0c5ba06f471d1d63aedfb61d708faa4b76565 (diff)
Fix acquisition pipeline stutter with many channels
Batches per-tick device readings into one signal emit instead of one per channel, evaluates derived channels once per tick instead of once per raw sample, and makes ChannelBuffer.window() cost scale with the window size instead of total buffer history. Strip chart X-range now scrolls on an independent ~30fps timer instead of only snapping when new data lands, which was the main visible cause of the stutter. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Diffstat (limited to 'core')
-rw-r--r--core/acquisition.py41
-rw-r--r--core/signal_processor.py18
2 files changed, 43 insertions, 16 deletions
diff --git a/core/acquisition.py b/core/acquisition.py
index a984c71..4fffb8b 100644
--- a/core/acquisition.py
+++ b/core/acquisition.py
@@ -33,13 +33,22 @@ class ChannelBuffer:
self.values.append(v)
def window(self, seconds: float) -> Tuple[List[float], List[float]]:
+ """Last `seconds` of data. Walks in from the newest sample only —
+ cost is proportional to the window size, not the full buffer
+ (which can hold far more history than is ever displayed)."""
if not self.times:
return [], []
cutoff = self.times[-1] - seconds
- ts = list(self.times)
- vs = list(self.values)
- idx = next((i for i, t in enumerate(ts) if t >= cutoff), 0)
- return ts[idx:], vs[idx:]
+ out_t: List[float] = []
+ out_v: List[float] = []
+ for t, v in zip(reversed(self.times), reversed(self.values)):
+ if t < cutoff:
+ break
+ out_t.append(t)
+ out_v.append(v)
+ out_t.reverse()
+ out_v.reverse()
+ return out_t, out_v
def all(self) -> Tuple[List[float], List[float]]:
return list(self.times), list(self.values)
@@ -61,13 +70,14 @@ class AcquisitionEngine(QObject):
Signals
-------
- new_data(device_id, channel_id, timestamp, value)
+ new_data(timestamp, readings) # readings: list[(device_id, channel_id, value)] —
+ # one batched emit per poll tick, not one per channel
device_status_changed(device_id, status_str)
log_started(filepath)
log_stopped(filepath)
"""
- new_data = pyqtSignal(str, str, float, float)
+ new_data = pyqtSignal(float, list)
device_status_changed = pyqtSignal(str, str)
log_started = pyqtSignal(str)
log_stopped = pyqtSignal(str)
@@ -109,6 +119,14 @@ class AcquisitionEngine(QObject):
# ── Start / stop ─────────────────────────────────────────────────────
+ def is_running(self) -> bool:
+ return self._running
+
+ def elapsed_now(self) -> float:
+ """Wall-clock elapsed since start() — lets the UI scroll the strip
+ chart smoothly between poll ticks instead of only on data arrival."""
+ return time.time() - self._t0 if self._running else 0.0
+
def start(self):
if self._running:
return
@@ -165,6 +183,7 @@ class AcquisitionEngine(QObject):
t_start = time.time()
elapsed = t_start - self._t0
log_row = [f"{elapsed:.4f}"]
+ batch: List[Tuple[str, str, float]] = []
for dev in list(self._devices):
active = dev.status in (DeviceStatus.CONNECTED, DeviceStatus.SIMULATED)
@@ -179,11 +198,15 @@ class AcquisitionEngine(QObject):
if val is None or not ch.enabled:
log_row.append("")
continue
+ fval = float(val)
buf = self._buffers.get(dev.info.device_id, {}).get(ch.channel_id)
if buf is not None:
- buf.append(elapsed, float(val))
- self.new_data.emit(dev.info.device_id, ch.channel_id, elapsed, float(val))
- log_row.append(f"{val:.5f}")
+ buf.append(elapsed, fval)
+ batch.append((dev.info.device_id, ch.channel_id, fval))
+ log_row.append(f"{fval:.5f}")
+
+ if batch:
+ self.new_data.emit(elapsed, batch)
if self._logging and self._csv_writer:
try:
diff --git a/core/signal_processor.py b/core/signal_processor.py
index bfb7ba3..8ab3937 100644
--- a/core/signal_processor.py
+++ b/core/signal_processor.py
@@ -380,7 +380,7 @@ class SignalProcessor(QObject):
Applies filter pipelines to raw channel data, then evaluates all
derived channels and emits processed_data for everything.
- Connect: engine.new_data → processor.on_raw_data
+ Connect: engine.new_data → processor.on_raw_batch
Connect: processor.processed_data → chart.on_new_data
"""
@@ -468,9 +468,16 @@ class SignalProcessor(QObject):
# ── Main data path ────────────────────────────────────────────────────
- def on_raw_data(self, device_id: str, channel_id: str,
- timestamp: float, value: float):
- """Slot: receive raw data, apply filters, emit processed, update derived."""
+ def on_raw_batch(self, timestamp: float, readings: List[Tuple[str, str, float]]):
+ """Slot: receive one tick's worth of raw readings, apply filters and
+ emit processed for each, then evaluate derived channels once for the
+ whole batch — not once per channel."""
+ for device_id, channel_id, value in readings:
+ self._process_one(device_id, channel_id, timestamp, value)
+ self._evaluate_derived(timestamp)
+
+ def _process_one(self, device_id: str, channel_id: str,
+ timestamp: float, value: float):
key = (device_id, channel_id)
# Apply filter pipeline
@@ -485,9 +492,6 @@ class SignalProcessor(QObject):
# Emit processed physical channel
self.processed_data.emit(device_id, channel_id, timestamp, processed)
- # Evaluate all derived channels whose sources include this channel
- self._evaluate_derived(timestamp)
-
def _evaluate_derived(self, timestamp: float):
with self._lock:
derived = list(self._derived)