diff options
Diffstat (limited to 'daq_system/devices/device_registry.py')
| -rw-r--r-- | daq_system/devices/device_registry.py | 87 |
1 files changed, 0 insertions, 87 deletions
diff --git a/daq_system/devices/device_registry.py b/daq_system/devices/device_registry.py deleted file mode 100644 index e407bc3..0000000 --- a/daq_system/devices/device_registry.py +++ /dev/null @@ -1,87 +0,0 @@ -""" -devices/device_registry.py - -Discovers, instantiates, and manages all device modules. -Add new devices by dropping a .py file into the devices/ directory. -""" - -import importlib -import inspect -import pkgutil -from pathlib import Path -from typing import Dict, List, Optional, Type - -from devices.base_device import BaseDevice, DeviceInfo - - -class DeviceRegistry: - """Central registry for all DAQ device modules.""" - - def __init__(self): - self._device_classes: Dict[str, Type[BaseDevice]] = {} - self._instances: Dict[str, BaseDevice] = {} - self._auto_discover() - - # ------------------------------------------------------------------ # - # Discovery # - # ------------------------------------------------------------------ # - - def _auto_discover(self): - """Scan the devices/ package for BaseDevice subclasses.""" - devices_path = Path(__file__).parent - package = "devices" - - for _, module_name, _ in pkgutil.iter_modules([str(devices_path)]): - if module_name.startswith("_") or module_name in ("base_device", "device_registry"): - continue - try: - module = importlib.import_module(f"{package}.{module_name}") - for name, obj in inspect.getmembers(module, inspect.isclass): - if issubclass(obj, BaseDevice) and obj is not BaseDevice: - self._device_classes[name] = obj - except Exception as e: - print(f"[DeviceRegistry] Failed to load {module_name}: {e}") - - def register_class(self, cls: Type[BaseDevice]) -> None: - """Manually register a device class (for testing / runtime plugins).""" - self._device_classes[cls.__name__] = cls - - # ------------------------------------------------------------------ # - # Instance management # - # ------------------------------------------------------------------ # - - def create_device(self, class_name: str, device_id: str, **kwargs) -> Optional[BaseDevice]: - """Instantiate a device by class name with a unique device_id.""" - cls = self._device_classes.get(class_name) - if cls is None: - raise ValueError(f"Unknown device class: {class_name}") - instance = cls(device_id=device_id, **kwargs) - self._instances[device_id] = instance - return instance - - def add_instance(self, device: BaseDevice) -> None: - """Register a pre-built device instance.""" - self._instances[device.info.device_id] = device - - def remove_instance(self, device_id: str) -> None: - dev = self._instances.pop(device_id, None) - if dev: - try: - dev.disconnect() - except Exception: - pass - - def get_instance(self, device_id: str) -> Optional[BaseDevice]: - return self._instances.get(device_id) - - def all_instances(self) -> List[BaseDevice]: - return list(self._instances.values()) - - def available_classes(self) -> List[str]: - return list(self._device_classes.keys()) - - def get_class(self, class_name: str) -> Optional[Type[BaseDevice]]: - return self._device_classes.get(class_name) - - def __len__(self): - return len(self._instances) |
