diff options
Diffstat (limited to 'daq_system/core/acquisition.py')
| -rw-r--r-- | daq_system/core/acquisition.py | 214 |
1 files changed, 214 insertions, 0 deletions
diff --git a/daq_system/core/acquisition.py b/daq_system/core/acquisition.py new file mode 100644 index 0000000..7e43529 --- /dev/null +++ b/daq_system/core/acquisition.py @@ -0,0 +1,214 @@ +""" +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 |