daily-briefing/server/services/ha_service.py

from __future__ import annotations

import httpx
from typing import Any, Dict, List, Optional


def _friendly_name(entity: Dict[str, Any]) -> str:
    """Extract the friendly name from an entity's attributes, falling back to entity_id."""
    attrs = entity.get("attributes", {})
    return attrs.get("friendly_name", entity.get("entity_id", "unknown"))


def _parse_light(entity: Dict[str, Any]) -> Dict[str, Any]:
    """Parse a light entity into a normalised dictionary."""
    attrs = entity.get("attributes", {})
    state = entity.get("state", "unknown")
    brightness_raw = attrs.get("brightness")
    brightness_pct: Optional[int] = None
    if brightness_raw is not None:
        try:
            brightness_pct = round(int(brightness_raw) / 255 * 100)
        except (ValueError, TypeError):
            brightness_pct = None

    return {
        "entity_id": entity.get("entity_id", ""),
        "name": _friendly_name(entity),
        "state": state,
        "brightness": brightness_pct,
        "color_mode": attrs.get("color_mode"),
    }


def _parse_cover(entity: Dict[str, Any]) -> Dict[str, Any]:
    """Parse a cover entity into a normalised dictionary."""
    attrs = entity.get("attributes", {})
    return {
        "entity_id": entity.get("entity_id", ""),
        "name": _friendly_name(entity),
        "state": entity.get("state", "unknown"),
        "current_position": attrs.get("current_position"),
    }


def _parse_sensor(entity: Dict[str, Any]) -> Dict[str, Any]:
    """Parse a temperature sensor entity into a normalised dictionary."""
    attrs = entity.get("attributes", {})
    state_value = entity.get("state", "unknown")
    try:
        state_value = round(float(state_value), 1)
    except (ValueError, TypeError):
        pass

    return {
        "entity_id": entity.get("entity_id", ""),
        "name": _friendly_name(entity),
        "state": state_value,
        "unit": attrs.get("unit_of_measurement", ""),
        "device_class": attrs.get("device_class", ""),
    }


async def fetch_ha_data(url: str, token: str) -> Dict[str, Any]:
    """Fetch and categorise entity states from a Home Assistant instance.

    Args:
        url: Base URL of the Home Assistant instance (e.g. ``http://192.168.1.100:8123``).
        token: Long-lived access token for authentication.

    Returns:
        Dictionary containing:
            - ``online``: Whether the HA instance is reachable.
            - ``lights``: List of light entities with state and brightness.
            - ``covers``: List of cover entities with state and position.
            - ``sensors``: List of temperature sensor entities.
            - ``lights_on``: Count of lights currently in the ``on`` state.
            - ``lights_total``: Total number of light entities.
            - ``error``: Error message if the request failed, else ``None``.
    """
    result: Dict[str, Any] = {
        "online": False,
        "lights": [],
        "covers": [],
        "sensors": [],
        "lights_on": 0,
        "lights_total": 0,
        "error": None,
    }

    if not url or not token:
        result["error"] = "Missing Home Assistant URL or token"
        return result

    base_url = url.rstrip("/")
    headers = {
        "Authorization": f"Bearer {token}",
        "Content-Type": "application/json",
    }

    try:
        async with httpx.AsyncClient(timeout=15, verify=False) as client:
            resp = await client.get(f"{base_url}/api/states", headers=headers)
            resp.raise_for_status()
            entities: List[Dict[str, Any]] = resp.json()
    except httpx.HTTPStatusError as exc:
        result["error"] = f"HTTP {exc.response.status_code}"
        return result
    except httpx.RequestError as exc:
        result["error"] = f"Connection failed: {exc}"
        return result
    except Exception as exc:
        result["error"] = str(exc)
        return result

    result["online"] = True

    lights: List[Dict[str, Any]] = []
    covers: List[Dict[str, Any]] = []
    sensors: List[Dict[str, Any]] = []

    for entity in entities:
        entity_id: str = entity.get("entity_id", "")
        domain = entity_id.split(".")[0] if "." in entity_id else ""
        attrs = entity.get("attributes", {})
        state = entity.get("state", "")

        if state in ("unavailable", "unknown"):
            continue

        if domain == "light":
            lights.append(_parse_light(entity))

        elif domain == "cover":
            covers.append(_parse_cover(entity))

        elif domain == "sensor":
            device_class = attrs.get("device_class", "")
            if device_class == "temperature":
                sensors.append(_parse_sensor(entity))

    lights_on = sum(1 for light in lights if light["state"] == "on")

    result["lights"] = lights
    result["covers"] = covers
    result["sensors"] = sensors
    result["lights_on"] = lights_on
    result["lights_total"] = len(lights)

    return result