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