""" 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)