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 SignalBuffer:
    """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, signal_id, value)] —
                                   # one batched emit per poll tick, not one per channel
    device_status_changed(device_id, status_str)
    channel_unit_changed(device_id, signal_id, new_unit)  # e.g. Mark-10 gauge
                                   # switched lb/kgF/N/ozF on its own front panel
    log_started(filepath)
    log_stopped(filepath)
    run_state_changed(running)   # emitted from start()/stop()
    """

    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)
    run_state_changed     = pyqtSignal(bool)   # True on start(), False on stop() —
                                                # e.g. ControlPanel's safety-interlock
                                                # controls listen for this

    def __init__(self, poll_interval_ms: int = 100):
        super().__init__()
        self._interval = poll_interval_ms / 1000.0
        self._buffer_size: int = MAX_BUFFER   # see set_buffer_size()
        self._devices: List[BaseDevice]                    = []
        self._buffers: Dict[str, Dict[str, SignalBuffer]] = {}
        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.signal_id: SignalBuffer(maxlen=self._buffer_size)
            for ch in dev.info.signals
        }
        for ch in dev.info.signals:
            self._last_units[(dev.info.device_id, ch.signal_id)] = ch.unit

    def set_buffer_size(self, n: int):
        """Change how much history each Signal's buffer holds — settings_
        window.py's "Buffer size (samples)" field, applied live rather than
        only on the next device add/reconfigure. Rebuilds every existing
        buffer at the new size, keeping its most recent samples (a maxlen
        deque naturally drops the oldest first as it's extended past
        capacity, so growing loses nothing and shrinking keeps the tail)."""
        self._buffer_size = n
        for dev_bufs in self._buffers.values():
            for ch_id, buf in list(dev_bufs.items()):
                new_buf = SignalBuffer(maxlen=n)
                new_buf.times.extend(buf.times)
                new_buf.values.extend(buf.values)
                dev_bufs[ch_id] = new_buf

    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.signal_id for ch in dev.info.signals}
        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.signals:
            if ch.signal_id not in bufs:
                bufs[ch.signal_id] = SignalBuffer(maxlen=self._buffer_size)
            self._last_units[(device_id, ch.signal_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_signal_buffer(self, device_id: str, signal_id: str) -> Optional[SignalBuffer]:
        return self._buffers.get(device_id, {}).get(signal_id)

    def rename_channel(self, device_id: str, old_id: str, new_id: str) -> None:
        """Move a channel's buffer/unit-cache entry to a new id after a rename
        — mirrors on_device_reconfigured's bookkeeping but for a single
        in-place rename rather than a full channel-list diff, so history is
        carried over instead of dropped."""
        if old_id == new_id:
            return
        bufs = self._buffers.get(device_id)
        if bufs and old_id in bufs:
            bufs[new_id] = bufs.pop(old_id)
        key = (device_id, old_id)
        if key in self._last_units:
            self._last_units[(device_id, new_id)] = self._last_units.pop(key)

    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()
        self.run_state_changed.emit(True)

    def stop(self):
        self._running = False
        if self._thread:
            self._thread.join(timeout=3.0)
        self.stop_logging()
        self.run_state_changed.emit(False)

    # ── 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.signals:
                headers.append(f"{dev.info.device_id}/{ch.signal_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.signals:
                    key = (dev.info.device_id, ch.signal_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.signal_id, ch.unit)

                    val = readings.get(ch.signal_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.signal_id)
                    if buf is not None:
                        buf.append(elapsed, fval)
                    batch.append((dev.info.device_id, ch.signal_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)