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
|
"""
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 ChannelConfig:
"""Configuration for a single I/O channel."""
channel_id: str
name: str
unit: str = ""
min_value: float = 0.0
max_value: float = 100.0
alarm_low: Optional[float] = None
alarm_high: Optional[float] = None
enabled: bool = True
color: str = "#00d4ff" # For plotting
extra: Dict[str, Any] = field(default_factory=dict)
@dataclass
class DeviceInfo:
"""Metadata describing a device module."""
device_id: str
name: str
device_type: str # e.g. "analog_input", "digital_io", "serial", "temperature"
description: str = ""
manufacturer: str = ""
model: str = ""
version: str = "1.0.0"
icon: str = "⚙" # Unicode icon for UI display
channels: List[ChannelConfig] = field(default_factory=list)
class BaseDevice(ABC):
"""
Abstract base class for all DAQ device plugins.
To create a new device module:
1. Subclass BaseDevice
2. Implement all @abstractmethod methods
3. Place the file in the devices/ directory
4. The DeviceRegistry will auto-discover it
"""
def __init__(self, device_info: DeviceInfo):
self.info = device_info
self.status = DeviceStatus.DISCONNECTED
self._callbacks: List[callable] = []
# ------------------------------------------------------------------ #
# Abstract interface — every device must implement these #
# ------------------------------------------------------------------ #
@abstractmethod
def connect(self) -> bool:
"""
Open connection to the physical device.
Returns True on success, False on failure.
Sets self.status appropriately.
"""
@abstractmethod
def disconnect(self) -> None:
"""Close the connection and release resources."""
@abstractmethod
def read_channels(self) -> Dict[str, float]:
"""
Read current values from all enabled channels.
Returns dict mapping channel_id -> float value.
Called repeatedly by the acquisition loop.
"""
@abstractmethod
def write_channel(self, channel_id: str, value: Any) -> bool:
"""
Write a value to an output channel (if supported).
Returns True on success.
"""
@abstractmethod
def get_config_widget(self):
"""
Return a QWidget with device-specific configuration controls.
This widget is embedded in the Device Config panel.
"""
# ------------------------------------------------------------------ #
# Shared helpers #
# ------------------------------------------------------------------ #
def add_data_callback(self, callback: callable) -> None:
"""Register a callback: callback(device_id, channel_id, value, timestamp)"""
self._callbacks.append(callback)
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_channel(self, channel_id: str) -> Optional[ChannelConfig]:
for ch in self.info.channels:
if ch.channel_id == channel_id:
return ch
return None
def __repr__(self):
return f"<{self.__class__.__name__} id={self.info.device_id} status={self.status.value}>"
|
