diff options
| author | Christian Kolset <christian.kolset@gmail.com> | 2026-04-13 15:22:56 -0600 |
|---|---|---|
| committer | Christian Kolset <christian.kolset@gmail.com> | 2026-04-13 15:22:56 -0600 |
| commit | 2ed9d37da7b27d25173535550fb92702225ac14e (patch) | |
| tree | d6ff360a656e577629d3cefdaa1ae02af7255864 /devices/device_registry.py | |
| parent | 8d6acf3a8ea4b37f86b321dbf430be5be01b1267 (diff) | |
Removed QtPy5 snippet and fixed directories
Diffstat (limited to 'devices/device_registry.py')
| -rw-r--r-- | devices/device_registry.py | 73 |
1 files changed, 73 insertions, 0 deletions
diff --git a/devices/device_registry.py b/devices/device_registry.py new file mode 100644 index 0000000..8ed6886 --- /dev/null +++ b/devices/device_registry.py @@ -0,0 +1,73 @@ +""" +devices/device_registry.py + +Auto-discovers and manages all BaseDevice subclasses. +Drop a new .py file in devices/ and it appears automatically. +""" + +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: + def __init__(self): + self._classes: Dict[str, Type[BaseDevice]] = {} + self._instances: Dict[str, BaseDevice] = {} + self._discover() + + # ── Discovery ──────────────────────────────────────────────────────── + + def _discover(self): + path = Path(__file__).parent + package = "devices" + skip = {"base_device", "device_registry"} + + for _, mod_name, _ in pkgutil.iter_modules([str(path)]): + if mod_name.startswith("_") or mod_name in skip: + continue + try: + mod = importlib.import_module(f"{package}.{mod_name}") + for name, obj in inspect.getmembers(mod, inspect.isclass): + if issubclass(obj, BaseDevice) and obj is not BaseDevice: + self._classes[name] = obj + except Exception as e: + print(f"[Registry] Could not load {mod_name}: {e}") + + # ── Instance management ────────────────────────────────────────────── + + def add_instance(self, device: BaseDevice) -> None: + 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._classes.keys()) + + def get_class(self, name: str) -> Optional[Type[BaseDevice]]: + return self._classes.get(name) + + def create(self, class_name: str, device_id: str, **kw) -> BaseDevice: + cls = self._classes.get(class_name) + if not cls: + raise ValueError(f"Unknown device class: {class_name}") + dev = cls(device_id=device_id, **kw) + self._instances[device_id] = dev + return dev + + def __len__(self): + return len(self._instances) |
