From 2ed9d37da7b27d25173535550fb92702225ac14e Mon Sep 17 00:00:00 2001 From: Christian Kolset Date: Mon, 13 Apr 2026 15:22:56 -0600 Subject: Removed QtPy5 snippet and fixed directories --- daq_system/devices/base_device.py | 127 -------------------------------------- 1 file changed, 127 deletions(-) delete mode 100644 daq_system/devices/base_device.py (limited to 'daq_system/devices/base_device.py') 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}>" -- cgit v1.2.3