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 --- devices/base_device.py | 104 +++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 104 insertions(+) create mode 100644 devices/base_device.py (limited to 'devices/base_device.py') diff --git a/devices/base_device.py b/devices/base_device.py new file mode 100644 index 0000000..922a743 --- /dev/null +++ b/devices/base_device.py @@ -0,0 +1,104 @@ +""" +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" + extra: Dict[str, Any] = field(default_factory=dict) + + +@dataclass +class DeviceInfo: + """Metadata describing a device module.""" + device_id: str + name: str + device_type: str # "analog_input" | "digital_io" | "serial" | "temperature" + description: str = "" + manufacturer: str = "" + model: str = "" + version: str = "1.0.0" + icon: str = "⚙" + channels: List[ChannelConfig] = 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 {channel_id: value} for all enabled channels.""" + + @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_channel(self, channel_id: str) -> Optional[ChannelConfig]: + return next((c for c in self.info.channels if c.channel_id == channel_id), None) + + def __repr__(self): + return f"<{self.__class__.__name__} id={self.info.device_id} status={self.status.value}>" -- cgit v1.2.3