1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
|
"""
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]]:
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)
device_status_changed(device_id, status_str)
log_started(filepath)
log_stopped(filepath)
"""
new_data = pyqtSignal(str, str, float, float)
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
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
}
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)
def clear_history(self):
for ch_map in self._buffers.values():
for buf in ch_map.values():
buf.clear()
# ── 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 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}"]
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 or not ch.enabled:
log_row.append("")
continue
buf = self._buffers.get(dev.info.device_id, {}).get(ch.channel_id)
if buf is not None:
buf.append(elapsed, float(val))
self.new_data.emit(dev.info.device_id, ch.channel_id, elapsed, float(val))
log_row.append(f"{val:.5f}")
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)
|
