69 lines
2.3 KiB
Python
69 lines
2.3 KiB
Python
|
"""Repo-root discovery and `instances.yaml` access.
|
||
|
|
|
||
|
|
The MCP server is a dev-station tool that reads the repo it lives in: HA
|
||
|
|
instance definitions come from services/home-assistant/instances.yaml (same
|
||
|
|
file scripts/ha/import.sh and deploy.sh read), automations come from
|
||
|
|
services/home-assistant/config/<instance>/automations/.
|
||
|
|
"""
|
||
|
|
import os
|
||
|
|
from pathlib import Path
|
||
|
|
|
||
|
|
import yaml
|
||
|
|
|
||
|
|
DEFAULT_INSTANCE = "ken"
|
||
|
|
|
||
|
|
#: Where scripts/ha/lib lives, so ha_api.py / ha_ws.py can be imported
|
||
|
|
#: instead of reimplemented (DESIGN.md, "Deploy path: adapter per instance").
|
||
|
|
_HA_LIB_RELPATH = "scripts/ha/lib"
|
||
|
|
|
||
|
|
|
||
|
|
class ConfigError(RuntimeError):
|
||
|
|
"""Raised for a missing/broken instances.yaml or an unknown instance."""
|
||
|
|
|
||
|
|
|
||
|
|
def repo_root():
|
||
|
|
"""Absolute path to the repo checkout this server was started from.
|
||
|
|
|
||
|
|
`HA_MCP_REPO` overrides it (useful when the server is launched from a
|
||
|
|
different worktree than the one it should read); otherwise it is derived
|
||
|
|
from this file's location: src/ha_mcp/config.py -> services/ha-mcp -> repo.
|
||
|
|
"""
|
||
|
|
override = os.environ.get("HA_MCP_REPO")
|
||
|
|
if override:
|
||
|
|
return Path(override).expanduser().resolve()
|
||
|
|
return Path(__file__).resolve().parents[4]
|
||
|
|
|
||
|
|
|
||
|
|
def ha_service_dir(root=None):
|
||
|
|
return (root or repo_root()) / "services" / "home-assistant"
|
||
|
|
|
||
|
|
|
||
|
|
def ha_lib_dir(root=None):
|
||
|
|
return (root or repo_root()) / _HA_LIB_RELPATH
|
||
|
|
|
||
|
|
|
||
|
|
def load_instances(root=None):
|
||
|
|
"""Parse instances.yaml -> {name: dict}. Raises ConfigError if unusable."""
|
||
|
|
path = ha_service_dir(root) / "instances.yaml"
|
||
|
|
try:
|
||
|
|
with open(path, "r", encoding="utf-8") as f:
|
||
|
|
data = yaml.safe_load(f)
|
||
|
|
except OSError as exc:
|
||
|
|
raise ConfigError(f"cannot read {path}: {exc}") from exc
|
||
|
|
except yaml.YAMLError as exc:
|
||
|
|
raise ConfigError(f"{path} is not valid YAML: {exc}") from exc
|
||
|
|
instances = (data or {}).get("instances") or {}
|
||
|
|
if not instances:
|
||
|
|
raise ConfigError(f"{path} defines no instances")
|
||
|
|
return instances
|
||
|
|
|
||
|
|
|
||
|
|
def get_instance(name=None, root=None):
|
||
|
|
"""Return (name, config-dict) for `name` (default: `ken`)."""
|
||
|
|
instances = load_instances(root)
|
||
|
|
name = name or DEFAULT_INSTANCE
|
||
|
|
if name not in instances:
|
||
|
|
known = ", ".join(sorted(instances))
|
||
|
|
raise ConfigError(f"unknown instance '{name}' (known: {known})")
|
||
|
|
return name, dict(instances[name] or {})
|