path: root/core/acquisition.py

"""
core/acquisition.py

Background acquisition engine.
Polls all connected devices, buffers data, fires Qt signals,
and writes CSV logs.
"""

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


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]]:
        """Last `seconds` of data. Walks in from the newest sample only —
        cost is proportional to the window size, not the full buffer
        (which can hold far more history than is ever displayed)."""
        if not self.times:
            return [], []
        cutoff = self.times[-1] - seconds
        out_t: List[float] = []
        out_v: List[float] = []
        for t, v in zip(reversed(self.times), reversed(self.values)):
            if t < cutoff:
                break
            out_t.append(t)
            out_v.append(v)
        out_t.reverse()
        out_v.reverse()
        return out_t, out_v

    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(timestamp, readings)  # readings: list[(device_id, channel_id, value)] —
                                   # one batched emit per poll tick, not one per channel
    device_status_changed(device_id, status_str)
    channel_unit_changed(device_id, channel_id, new_unit)  # e.g. Mark-10 gauge
                                   # switched lb/kgF/N/ozF on its own front panel
    log_started(filepath)
    log_stopped(filepath)
    """

    new_data              = pyqtSignal(float, list)
    device_status_changed = pyqtSignal(str, str)
    channel_unit_changed  = pyqtSignal(str, 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._last_units: Dict[Tuple[str, str], str]        = {}
        self._last_status: Dict[str, str]                   = {}
        self._running  = False
        self._thread:  Optional[threading.Thread]          = None
        self._t0       = 0.0

        self._logging    = False
        self._log_path   = ""
        self._csv_file   = None
        self._csv_writer = None

    # ── 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
        }
        for ch in dev.info.channels:
            self._last_units[(dev.info.device_id, ch.channel_id)] = ch.unit

    def on_device_reconfigured(self, device_id: str):
        """Reconcile buffers with a device's current channel list — reconfigure()
        can add/remove channels (e.g. Signals count changed), and buffers keyed
        by the old channel list would silently drop data for anything new."""
        dev = next((d for d in self._devices if d.info.device_id == device_id), None)
        if dev is None:
            return
        bufs = self._buffers.setdefault(device_id, {})
        valid_ids = {ch.channel_id for ch in dev.info.channels}
        for stale_id in [cid for cid in bufs if cid not in valid_ids]:
            bufs.pop(stale_id, None)
        for ch in dev.info.channels:
            if ch.channel_id not in bufs:
                bufs[ch.channel_id] = ChannelBuffer()
            self._last_units[(device_id, ch.channel_id)] = ch.unit
        for key in [k for k in self._last_units if k[0] == device_id and k[1] not in valid_ids]:
            self._last_units.pop(key, None)

    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)
        self._last_status.pop(device_id, None)
        for key in [k for k in self._last_units if k[0] == device_id]:
            self._last_units.pop(key, None)

    def get_buffer(self, device_id: str, channel_id: str) -> Optional[ChannelBuffer]:
        return self._buffers.get(device_id, {}).get(channel_id)

    def clear_history(self):
        for ch_map in self._buffers.values():
            for buf in ch_map.values():
                buf.clear()

    # ── Start / stop ─────────────────────────────────────────────────────

    def is_running(self) -> bool:
        return self._running

    def elapsed_now(self) -> float:
        """Wall-clock elapsed since start() — lets the UI scroll the strip
        chart smoothly between poll ticks instead of only on data arrival."""
        return time.time() - self._t0 if self._running else 0.0

    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 or filepath.endswith(os.sep) or filepath.endswith("/"):
            dir_ = filepath if filepath else "logs"
            os.makedirs(dir_, exist_ok=True)
            ts = datetime.now().strftime("%Y%m%d_%H%M%S")
            filepath = os.path.join(dir_, f"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}"]
            batch: List[Tuple[str, str, float]] = []

            for dev in list(self._devices):
                if self._last_status.get(dev.info.device_id) != dev.status.value:
                    self._last_status[dev.info.device_id] = dev.status.value
                    self.device_status_changed.emit(dev.info.device_id, dev.status.value)

                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:
                    key = (dev.info.device_id, ch.channel_id)
                    if self._last_units.get(key) != ch.unit:
                        self._last_units[key] = ch.unit
                        self.channel_unit_changed.emit(dev.info.device_id, ch.channel_id, ch.unit)

                    val = readings.get(ch.channel_id)
                    if val is None or not ch.enabled:
                        log_row.append("")
                        continue
                    fval = float(val)
                    buf = self._buffers.get(dev.info.device_id, {}).get(ch.channel_id)
                    if buf is not None:
                        buf.append(elapsed, fval)
                    batch.append((dev.info.device_id, ch.channel_id, fval))
                    log_row.append(f"{fval:.5f}")

            if batch:
                self.new_data.emit(elapsed, batch)

            if self._logging and self._csv_writer:
                try:
                    self._csv_writer.writerow(log_row)
                except Exception:
                    pass

            dt    = time.time() - t_start
            sleep = self._interval - dt
            if sleep > 0:
                time.sleep(sleep)