summaryrefslogtreecommitdiff
path: root/devices/device_registry.py
diff options
context:
space:
mode:
authorChristian Kolset <christian.kolset@gmail.com>2026-04-13 15:22:56 -0600
committerChristian Kolset <christian.kolset@gmail.com>2026-04-13 15:22:56 -0600
commit2ed9d37da7b27d25173535550fb92702225ac14e (patch)
treed6ff360a656e577629d3cefdaa1ae02af7255864 /devices/device_registry.py
parent8d6acf3a8ea4b37f86b321dbf430be5be01b1267 (diff)
Removed QtPy5 snippet and fixed directories
Diffstat (limited to 'devices/device_registry.py')
-rw-r--r--devices/device_registry.py73
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)