path: root/core/acquisition.py

"""
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