init

1 files changed, 127 insertions, 0 deletions

diff --git a/daq_system/devices/base_device.py b/daq_system/devices/base_device.py
new file mode 100644
index 0000000..4898a7a
--- /dev/null
+++ b/daq_system/devices/base_device.py
@@ -0,0 +1,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}>"