webcontrol/scripts/update_ha_entities.py

#!/usr/bin/env python3
"""
Update ha_entities.csv with all entities from Home Assistant.

Reads HA URL and token from public/config.json (or HA_URL / HA_TOKEN env vars).
Fetches GET /api/states, then writes entity_id, domain, friendly_name,
device_class, and a suggested description to the CSV.

Usage:
  python scripts/update_ha_entities.py [--config path] [--output path]
"""

import argparse
import csv
import json
import os
import re
import sys
from pathlib import Path
from urllib.request import Request, urlopen
from urllib.error import URLError, HTTPError

# Default paths relative to project root
PROJECT_ROOT = Path(__file__).resolve().parent.parent
DEFAULT_CONFIG = PROJECT_ROOT / "public" / "config.json"
DEFAULT_OUTPUT = PROJECT_ROOT / "ha_entities.csv"

# Domain -> human label for generating descriptions
DOMAIN_LABELS = {
    "light": "Light",
    "switch": "Switch",
    "sensor": "Sensor",
    "binary_sensor": "Binary sensor",
    "climate": "Climate",
    "cover": "Cover",
    "fan": "Fan",
    "lock": "Lock",
    "media_player": "Media player",
    "camera": "Camera",
    "automation": "Automation",
    "script": "Script",
    "scene": "Scene",
    "input_boolean": "Input boolean",
    "input_number": "Input number",
    "input_text": "Input text",
    "input_datetime": "Input datetime",
    "input_select": "Input select",
    "device_tracker": "Device tracker",
    "person": "Person",
    "zone": "Zone",
    "sun": "Sun",
    "weather": "Weather",
    "update": "Update",
    "button": "Button",
    "number": "Number",
    "select": "Select",
    "vacuum": "Vacuum",
    "humidifier": "Humidifier",
    "water_heater": "Water heater",
    "calendar": "Calendar",
    "timer": "Timer",
    "counter": "Counter",
}


def humanize_id(object_id: str) -> str:
    """Turn entity object_id into readable text (e.g. living_room -> Living room)."""
    return re.sub(r"[_.]+", " ", object_id).strip().title()


def suggested_description(entity: dict) -> str:
    """Build a short description for the entity."""
    attrs = entity.get("attributes") or {}
    friendly = attrs.get("friendly_name") or attrs.get("name")
    if friendly:
        return str(friendly).strip()
    domain = (entity.get("entity_id") or "").split(".", 1)[0]
    object_id = (entity.get("entity_id") or "").split(".", 1)[-1]
    label = DOMAIN_LABELS.get(domain, domain.replace("_", " ").title())
    return f"{label}: {humanize_id(object_id)}"


def load_config(config_path: Path) -> tuple[str, str]:
    """Return (ha_url, ha_token). Prefer env HA_URL, HA_TOKEN over config file."""
    url = os.environ.get("HA_URL")
    token = os.environ.get("HA_TOKEN")
    if url and token:
        return url.rstrip("/"), token
    if not config_path.is_file():
        raise SystemExit(
            f"Config not found: {config_path}. Set HA_URL and HA_TOKEN or create config."
        )
    with open(config_path, encoding="utf-8") as f:
        data = json.load(f)
    settings = data.get("settings") or {}
    url = settings.get("haUrl") or os.environ.get("HA_URL")
    token = settings.get("haToken") or os.environ.get("HA_TOKEN")
    if not url or not token:
        raise SystemExit("Missing haUrl or haToken in config (or HA_URL/HA_TOKEN env).")
    return url.rstrip("/"), token


def fetch_states(ha_url: str, ha_token: str) -> list[dict]:
    """GET /api/states and return list of state objects."""
    req = Request(
        f"{ha_url}/api/states",
        headers={"Authorization": f"Bearer {ha_token}"},
        method="GET",
    )
    try:
        with urlopen(req, timeout=30) as resp:
            return json.load(resp)
    except HTTPError as e:
        raise SystemExit(f"Home Assistant API error: {e.code} {e.reason}")
    except URLError as e:
        raise SystemExit(f"Cannot reach Home Assistant: {e.reason}")


def main() -> None:
    parser = argparse.ArgumentParser(description="Update ha_entities.csv from Home Assistant.")
    parser.add_argument(
        "--config",
        type=Path,
        default=DEFAULT_CONFIG,
        help="Path to config.json (default: public/config.json)",
    )
    parser.add_argument(
        "--output",
        "-o",
        type=Path,
        default=DEFAULT_OUTPUT,
        help="Output CSV path (default: ha_entities.csv)",
    )
    args = parser.parse_args()

    ha_url, ha_token = load_config(args.config)
    states = fetch_states(ha_url, ha_token)

    rows = []
    for entity in states:
        eid = entity.get("entity_id") or ""
        if not eid:
            continue
        domain = eid.split(".", 1)[0]
        attrs = entity.get("attributes") or {}
        friendly_name = (attrs.get("friendly_name") or attrs.get("name") or "").strip()
        device_class = (attrs.get("device_class") or "").strip()
        description = suggested_description(entity)
        rows.append(
            {
                "entity_id": eid,
                "domain": domain,
                "friendly_name": friendly_name,
                "device_class": device_class,
                "description": description,
            }
        )

    rows.sort(key=lambda r: (r["domain"], r["entity_id"]))

    args.output.parent.mkdir(parents=True, exist_ok=True)
    with open(args.output, "w", newline="", encoding="utf-8") as f:
        w = csv.DictWriter(
            f,
            fieldnames=["entity_id", "domain", "friendly_name", "device_class", "description"],
            quoting=csv.QUOTE_MINIMAL,
        )
        w.writeheader()
        w.writerows(rows)

    print(f"Wrote {len(rows)} entities to {args.output}", file=sys.stderr)


if __name__ == "__main__":
    main()