1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
|
"""
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)
|
