""" core/acquisition.py Background acquisition engine. Polls all connected devices, buffers data, fires Qt signals, writes CSV logs, and checks alarm thresholds. """ import csv import os import threading import time from collections import deque from datetime import datetime from typing import Dict, List, Optional, Tuple from PyQt6.QtCore import QObject, pyqtSignal from devices.base_device import BaseDevice, DeviceStatus MAX_BUFFER = 20_000 # samples per channel class ChannelBuffer: """Circular time-series buffer for one channel.""" 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 window(self, seconds: float) -> Tuple[List[float], List[float]]: """Return the last `seconds` worth of data.""" 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:] def all(self) -> Tuple[List[float], List[float]]: return list(self.times), list(self.values) def latest(self) -> Optional[float]: return self.values[-1] if self.values else None def clear(self): self.times.clear() self.values.clear() def __len__(self): return len(self.times) class AcquisitionEngine(QObject): """ Thread-safe DAQ polling engine. 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_started(filepath) log_stopped(filepath) """ new_data = pyqtSignal(str, str, float, float) alarm_triggered = pyqtSignal(str, str, float, str) device_status_changed = pyqtSignal(str, str) log_started = pyqtSignal(str) log_stopped = pyqtSignal(str) def __init__(self, poll_interval_ms: int = 100): super().__init__() self._interval = 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._t0 = 0.0 # Logging self._logging = False self._log_path = "" self._csv_file = None self._csv_writer = None # Alarm dedup self._alarm_state: Dict[str, bool] = {} # ── Device management ──────────────────────────────────────────────── def add_device(self, dev: BaseDevice): self._devices.append(dev) self._buffers[dev.info.device_id] = { ch.channel_id: ChannelBuffer() for ch in dev.info.channels } def remove_device(self, device_id: str): 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): if self._running: return self._t0 = time.time() self._running = True self._thread = threading.Thread(target=self._loop, daemon=True, name="DAQ-Acq") self._thread.start() def stop(self): self._running = False if self._thread: self._thread.join(timeout=3.0) self.stop_logging() # ── Logging ────────────────────────────────────────────────────────── def start_logging(self, filepath: str = "") -> str: if not filepath: os.makedirs("logs", exist_ok=True) ts = datetime.now().strftime("%Y%m%d_%H%M%S") filepath = f"logs/daq_{ts}.csv" self._log_path = filepath self._csv_file = open(filepath, "w", newline="") headers = ["elapsed_s"] for dev in self._devices: for ch in dev.info.channels: headers.append(f"{dev.info.device_id}/{ch.channel_id}[{ch.unit}]") self._csv_writer = csv.writer(self._csv_file) self._csv_writer.writerow(headers) self._logging = True self.log_started.emit(filepath) return filepath def stop_logging(self): if not self._logging: return self._logging = False path = self._log_path try: if self._csv_file: self._csv_file.flush() self._csv_file.close() except Exception: pass self._csv_file = None self._csv_writer = None self.log_stopped.emit(path) # ── Acquisition loop ────────────────────────────────────────────────── def _loop(self): while self._running: t_start = time.time() elapsed = t_start - self._t0 log_row = [f"{elapsed:.4f}"] for dev in list(self._devices): active = dev.status in (DeviceStatus.CONNECTED, DeviceStatus.SIMULATED) try: readings = dev.read_channels() if active else {} except Exception as e: print(f"[Acq] {dev.info.device_id} read error: {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(elapsed, float(val)) # Signal self.new_data.emit(dev.info.device_id, ch.channel_id, elapsed, float(val)) log_row.append(f"{val:.5f}") # Alarms self._check_alarm(dev.info.device_id, ch, float(val)) if self._logging and self._csv_writer: try: self._csv_writer.writerow(log_row) except Exception: pass # Sleep remainder of interval dt = time.time() - t_start sleep = self._interval - dt if sleep > 0: time.sleep(sleep) # ── Alarm logic ─────────────────────────────────────────────────────── def _check_alarm(self, device_id: str, ch, val: float): lo_key = f"{device_id}.{ch.channel_id}.lo" hi_key = f"{device_id}.{ch.channel_id}.hi" if ch.alarm_low is not None: if val < ch.alarm_low: if not self._alarm_state.get(lo_key): self._alarm_state[lo_key] = True self.alarm_triggered.emit(device_id, ch.channel_id, val, "low") else: self._alarm_state[lo_key] = False if ch.alarm_high is not None: if val > ch.alarm_high: if not self._alarm_state.get(hi_key): self._alarm_state[hi_key] = True self.alarm_triggered.emit(device_id, ch.channel_id, val, "high") else: self._alarm_state[hi_key] = False