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
|
"""
devices/base_device.py
Abstract base class for all DAQ I/O modules.
Every device plugin must subclass BaseDevice and implement the required methods.
"""
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from typing import Any, Dict, List, Optional
from enum import Enum
class DeviceStatus(Enum):
DISCONNECTED = "disconnected"
CONNECTING = "connecting"
CONNECTED = "connected"
ERROR = "error"
SIMULATED = "simulated"
@dataclass
class SignalConfig:
"""Configuration for a single physical Signal — one I/O stream coming
off a Device (e.g. an Arduino analog pin, a digital line). Contrast
with a Channel (core/signal_processor.py), which takes one or more
Signals and applies math/filters to produce a new derived stream."""
signal_id: str
name: str
unit: str = ""
min_value: float = 0.0
max_value: float = 100.0
enabled: bool = True
color: str = "#00d4ff"
writable: bool = False # True if this signal accepts write_channel() calls
extra: Dict[str, Any] = field(default_factory=dict)
@dataclass
class DeviceInfo:
"""Metadata describing a device module."""
device_id: str
name: str
device_type: str # "arduino" | "serial" | "nidaqmx" | ...
description: str = ""
manufacturer: str = ""
model: str = ""
version: str = "1.0.0"
icon: str = "⚙"
signals: List[SignalConfig] = field(default_factory=list)
class BaseDevice(ABC):
"""
Abstract base for all DAQ device plugins.
To create a new module:
1. Subclass BaseDevice
2. Implement all @abstractmethod methods
3. Drop the file in devices/ — DeviceRegistry auto-discovers it
"""
def __init__(self, device_info: DeviceInfo):
self.info = device_info
self.status = DeviceStatus.DISCONNECTED
self._callbacks: List[Any] = []
# ── Required interface ──────────────────────────────────────────────
@abstractmethod
def connect(self) -> bool:
"""Open connection. Returns True on success."""
@abstractmethod
def disconnect(self) -> None:
"""Close connection and release resources."""
@abstractmethod
def read_channels(self) -> Dict[str, float]:
"""Return {signal_id: value} for all enabled signals."""
@abstractmethod
def write_channel(self, channel_id: str, value: Any) -> bool:
"""Write value to an output channel. Returns True on success."""
@abstractmethod
def get_config_widget(self):
"""Return a QWidget with device-specific configuration controls."""
# ── Shared helpers ──────────────────────────────────────────────────
def add_data_callback(self, cb) -> None:
self._callbacks.append(cb)
def _emit(self, channel_id: str, value: float, timestamp: float) -> None:
for cb in self._callbacks:
try:
cb(self.info.device_id, channel_id, value, timestamp)
except Exception:
pass
def get_signal(self, signal_id: str) -> Optional[SignalConfig]:
return next((s for s in self.info.signals if s.signal_id == signal_id), None)
def __repr__(self):
return f"<{self.__class__.__name__} id={self.info.device_id} status={self.status.value}>"
def merge_signal_metadata(old_signals: List[SignalConfig], new_signals: List[SignalConfig]) -> List[SignalConfig]:
"""Reapply user-customized Signal metadata (name/unit/enabled/color) onto a
freshly rebuilt signal list, keyed by signal_id. Mirrors exactly the fields
ProfileManager treats as user overrides (core/profile.py capture/apply) so
identity set at profile-load time survives a later device reconfigure too —
otherwise a device's own reconfigure()/remap_pins() rebuilding info.signals
from scratch silently resets names back to hardware defaults. Signal ids
not present in old_signals (new pins) keep their fresh defaults; ids no
longer in new_signals (removed pins) are simply dropped."""
old_by_id = {s.signal_id: s for s in old_signals}
for sig in new_signals:
old = old_by_id.get(sig.signal_id)
if old is not None:
sig.name, sig.unit, sig.enabled, sig.color = old.name, old.unit, old.enabled, old.color
return new_signals
|
