summaryrefslogtreecommitdiff
path: root/daq_system/devices/base_device.py
diff options
context:
space:
mode:
Diffstat (limited to 'daq_system/devices/base_device.py')
-rw-r--r--daq_system/devices/base_device.py127
1 files changed, 0 insertions, 127 deletions
diff --git a/daq_system/devices/base_device.py b/daq_system/devices/base_device.py
deleted file mode 100644
index 4898a7a..0000000
--- a/daq_system/devices/base_device.py
+++ /dev/null
@@ -1,127 +0,0 @@
-"""
-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}>"