Adolfo Reyna
128 lines
4.0 KiB
Python
128 lines
4.0 KiB
Python
from __future__ import annotations
|
|
|
|
import os
|
|
from pathlib import Path
|
|
from typing import Any, Dict, List, Optional
|
|
|
|
import yaml
|
|
from pydantic import BaseModel, Field
|
|
|
|
DEFAULT_REGISTRY_PATH = Path.home() / ".hermes" / "local_devices.yaml"
|
|
CACHE_DIR = Path(os.environ.get("REYNA_CLI_CACHE_DIR", Path.home() / ".cache" / "reyna-cli"))
|
|
|
|
|
|
class Device(BaseModel):
|
|
id: str
|
|
display_name: str = ""
|
|
type: str = ""
|
|
status: str = ""
|
|
host: str = ""
|
|
url: str = ""
|
|
fallback_url: str = ""
|
|
reserved_ip: str = ""
|
|
ports: Dict[str, Any] = Field(default_factory=dict)
|
|
capabilities: List[str] = Field(default_factory=list)
|
|
preferred_actions: Dict[str, str] = Field(default_factory=dict)
|
|
notes: List[str] = Field(default_factory=list)
|
|
|
|
@property
|
|
def name(self) -> str:
|
|
return self.id
|
|
|
|
def candidate_urls(self) -> List[str]:
|
|
urls: List[str] = []
|
|
for url in (self.url, self.fallback_url):
|
|
if url and url not in urls:
|
|
urls.append(url)
|
|
return urls
|
|
|
|
def public_dict(self) -> Dict[str, Any]:
|
|
return {
|
|
"id": self.id,
|
|
"display_name": self.display_name,
|
|
"type": self.type,
|
|
"status": self.status,
|
|
"host": self.host,
|
|
"url": self.url,
|
|
"fallback_url": self.fallback_url,
|
|
"reserved_ip": self.reserved_ip,
|
|
"ports": self.ports,
|
|
"capabilities": self.capabilities,
|
|
"preferred_actions": self.preferred_actions,
|
|
"notes": self.notes,
|
|
}
|
|
|
|
|
|
class Registry(BaseModel):
|
|
path: Optional[str] = None
|
|
version: int = 1
|
|
updated: str = ""
|
|
aliases: Dict[str, str] = Field(default_factory=dict)
|
|
devices: List[Device] = Field(default_factory=list)
|
|
|
|
|
|
def _device_from_mapping(device_id: str, raw: Dict[str, Any]) -> Device:
|
|
data = dict(raw or {})
|
|
data["id"] = device_id
|
|
return Device(**data)
|
|
|
|
|
|
def load_registry(path: Optional[Path] = None) -> Registry:
|
|
paths_to_try: List[Path] = []
|
|
if path:
|
|
paths_to_try.append(Path(path).expanduser())
|
|
env_path = os.environ.get("REYNA_DEVICES_REGISTRY")
|
|
if env_path:
|
|
paths_to_try.append(Path(env_path).expanduser())
|
|
paths_to_try.extend([DEFAULT_REGISTRY_PATH, Path.cwd() / "local_devices.yaml"])
|
|
|
|
seen: set[Path] = set()
|
|
for p in paths_to_try:
|
|
p = p.expanduser()
|
|
if p in seen:
|
|
continue
|
|
seen.add(p)
|
|
if not p.exists():
|
|
continue
|
|
with p.open("r", encoding="utf-8") as f:
|
|
data = yaml.safe_load(f) or {}
|
|
devices_raw = data.get("devices", {}) or {}
|
|
devices: List[Device] = []
|
|
if isinstance(devices_raw, dict):
|
|
devices = [_device_from_mapping(k, v) for k, v in devices_raw.items()]
|
|
elif isinstance(devices_raw, list):
|
|
for item in devices_raw:
|
|
if isinstance(item, dict):
|
|
device_id = item.get("id") or item.get("name") or item.get("display_name")
|
|
if device_id:
|
|
devices.append(_device_from_mapping(str(device_id), item))
|
|
return Registry(
|
|
path=str(p),
|
|
version=data.get("version", 1),
|
|
updated=str(data.get("updated", "")),
|
|
aliases=data.get("aliases", {}) or {},
|
|
devices=devices,
|
|
)
|
|
return Registry(path=None)
|
|
|
|
|
|
def get_device(name: str, registry: Optional[Registry] = None) -> Optional[Device]:
|
|
registry = registry or load_registry()
|
|
normalized = name.strip()
|
|
shorthand = {
|
|
"screen": "esp32_screen",
|
|
"esp32": "esp32_screen",
|
|
"iphone": "iphone_mcp",
|
|
"phone": "iphone_mcp",
|
|
"arm": "robot_arm",
|
|
}
|
|
normalized = shorthand.get(normalized, normalized)
|
|
for device in registry.devices:
|
|
if normalized in {device.id, device.host, device.display_name}:
|
|
return device
|
|
return None
|
|
|
|
|
|
def cache_path_for(device_id: str) -> Path:
|
|
return CACHE_DIR / "devices" / f"{device_id}-tools.json"
|