summaryrefslogtreecommitdiff
path: root/daq_system/devices/device_registry.py
diff options
context:
space:
mode:
Diffstat (limited to 'daq_system/devices/device_registry.py')
-rw-r--r--daq_system/devices/device_registry.py87
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)