From 2ed9d37da7b27d25173535550fb92702225ac14e Mon Sep 17 00:00:00 2001 From: Christian Kolset Date: Mon, 13 Apr 2026 15:22:56 -0600 Subject: Removed QtPy5 snippet and fixed directories --- core/__init__.py | 3 + core/__pycache__/__init__.cpython-314.pyc | Bin 0 -> 277 bytes core/__pycache__/acquisition.cpython-314.pyc | Bin 0 -> 14572 bytes core/acquisition.py | 225 +++++++++++++++++++++++++++ 4 files changed, 228 insertions(+) create mode 100644 core/__init__.py create mode 100644 core/__pycache__/__init__.cpython-314.pyc create mode 100644 core/__pycache__/acquisition.cpython-314.pyc create mode 100644 core/acquisition.py (limited to 'core') diff --git a/core/__init__.py b/core/__init__.py new file mode 100644 index 0000000..06fabd5 --- /dev/null +++ b/core/__init__.py @@ -0,0 +1,3 @@ +# core/__init__.py +from core.acquisition import AcquisitionEngine, ChannelBuffer +__all__ = ["AcquisitionEngine", "ChannelBuffer"] diff --git a/core/__pycache__/__init__.cpython-314.pyc b/core/__pycache__/__init__.cpython-314.pyc new file mode 100644 index 0000000..2a7cb7e Binary files /dev/null and b/core/__pycache__/__init__.cpython-314.pyc differ diff --git a/core/__pycache__/acquisition.cpython-314.pyc b/core/__pycache__/acquisition.cpython-314.pyc new file mode 100644 index 0000000..d84b66b Binary files /dev/null and b/core/__pycache__/acquisition.cpython-314.pyc differ diff --git a/core/acquisition.py b/core/acquisition.py new file mode 100644 index 0000000..640e6d5 --- /dev/null +++ b/core/acquisition.py @@ -0,0 +1,225 @@ +""" +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 -- cgit v1.2.3