""" core/acquisition.py Background acquisition engine. Polls all connected devices at their configured sample rates and emits data via Qt signals. """ import time import threading import csv import os from collections import deque from datetime import datetime from typing import Callable, Dict, List, Optional, Tuple from PyQt6.QtCore import QObject, pyqtSignal from devices.base_device import BaseDevice, DeviceStatus MAX_BUFFER = 10_000 # points per channel class ChannelBuffer: """Ring buffer for one channel's time-series data.""" def __init__(self, maxlen: int = MAX_BUFFER): self.times: deque = deque(maxlen=maxlen) self.values: deque = deque(maxlen=maxlen) def append(self, t: float, v: float): self.times.append(t) self.values.append(v) def latest(self, n: int = 1) -> Tuple[List[float], List[float]]: ts = list(self.times)[-n:] vs = list(self.values)[-n:] return ts, vs def all(self) -> Tuple[List[float], List[float]]: return list(self.times), list(self.values) def clear(self): self.times.clear() self.values.clear() def __len__(self): return len(self.times) class AcquisitionEngine(QObject): """ Runs a background polling thread for all registered devices. Signals: new_data(device_id, channel_id, timestamp, value) alarm_triggered(device_id, channel_id, value, kind) kind = 'low'|'high' device_status_changed(device_id, status_str) log_written(filepath) """ new_data = pyqtSignal(str, str, float, float) alarm_triggered = pyqtSignal(str, str, float, str) device_status_changed = pyqtSignal(str, str) log_written = pyqtSignal(str) def __init__(self, poll_interval_ms: int = 100): super().__init__() self.poll_interval_s = poll_interval_ms / 1000.0 self._devices: List[BaseDevice] = [] self._buffers: Dict[str, Dict[str, ChannelBuffer]] = {} self._running = False self._thread: Optional[threading.Thread] = None self._start_time = 0.0 # Logging self._logging = False self._log_file: Optional[str] = None self._csv_writer = None self._csv_handle = None # Alarm state (prevent repeated triggers) self._alarm_active: Dict[str, bool] = {} # ------------------------------------------------------------------ # # Device management # # ------------------------------------------------------------------ # def add_device(self, device: BaseDevice) -> None: self._devices.append(device) self._buffers[device.info.device_id] = { ch.channel_id: ChannelBuffer() for ch in device.info.channels } def remove_device(self, device_id: str) -> None: self._devices = [d for d in self._devices if d.info.device_id != device_id] self._buffers.pop(device_id, None) def get_buffer(self, device_id: str, channel_id: str) -> Optional[ChannelBuffer]: return self._buffers.get(device_id, {}).get(channel_id) # ------------------------------------------------------------------ # # Start / Stop # # ------------------------------------------------------------------ # def start(self) -> None: if self._running: return self._running = True self._start_time = time.time() self._thread = threading.Thread(target=self._loop, daemon=True) self._thread.start() def stop(self) -> None: self._running = False if self._thread: self._thread.join(timeout=2.0) self.stop_logging() # ------------------------------------------------------------------ # # Logging # # ------------------------------------------------------------------ # def start_logging(self, filepath: Optional[str] = None) -> str: if filepath is None: ts = datetime.now().strftime("%Y%m%d_%H%M%S") os.makedirs("logs", exist_ok=True) filepath = f"logs/daq_{ts}.csv" self._log_file = filepath self._csv_handle = open(filepath, "w", newline="") # Build header headers = ["timestamp"] for dev in self._devices: for ch in dev.info.channels: headers.append(f"{dev.info.device_id}.{ch.channel_id}") self._csv_writer = csv.writer(self._csv_handle) self._csv_writer.writerow(headers) self._logging = True return filepath def stop_logging(self) -> None: self._logging = False if self._csv_handle: try: self._csv_handle.close() except Exception: pass self._csv_handle = None self._csv_writer = None # ------------------------------------------------------------------ # # Background loop # # ------------------------------------------------------------------ # def _loop(self) -> None: while self._running: t0 = time.time() timestamp = t0 - self._start_time log_row = [f"{timestamp:.3f}"] for dev in self._devices: if dev.status not in (DeviceStatus.CONNECTED, DeviceStatus.SIMULATED): for ch in dev.info.channels: log_row.append("") continue try: readings = dev.read_channels() except Exception as e: print(f"[Acq] Error reading {dev.info.device_id}: {e}") readings = {} for ch in dev.info.channels: val = readings.get(ch.channel_id) if val is None: log_row.append("") continue # Buffer buf = self._buffers.get(dev.info.device_id, {}).get(ch.channel_id) if buf is not None: buf.append(timestamp, val) # Signal (emit on main thread via Qt queued connection) self.new_data.emit(dev.info.device_id, ch.channel_id, timestamp, val) log_row.append(f"{val:.4f}") # Alarms self._check_alarm(dev.info.device_id, ch, val) if self._logging and self._csv_writer: try: self._csv_writer.writerow(log_row) except Exception: pass elapsed = time.time() - t0 sleep_t = self.poll_interval_s - elapsed if sleep_t > 0: time.sleep(sleep_t) def _check_alarm(self, device_id: str, ch, val: float) -> None: key_lo = f"{device_id}.{ch.channel_id}.low" key_hi = f"{device_id}.{ch.channel_id}.high" if ch.alarm_low is not None: if val < ch.alarm_low and not self._alarm_active.get(key_lo): self._alarm_active[key_lo] = True self.alarm_triggered.emit(device_id, ch.channel_id, val, "low") elif val >= ch.alarm_low: self._alarm_active[key_lo] = False if ch.alarm_high is not None: if val > ch.alarm_high and not self._alarm_active.get(key_hi): self._alarm_active[key_hi] = True self.alarm_triggered.emit(device_id, ch.channel_id, val, "high") elif val <= ch.alarm_high: self._alarm_active[key_hi] = False