goodwe-addon/goodwe_controller/app/hass.py

"""Talking to Home Assistant through the Supervisor proxy.

Add-ons authenticate with SUPERVISOR_TOKEN against http://supervisor/core/api,
so there is no long-lived token to create, store, paste into a config file, or
leak at a client site. That is one of the main reasons this is an add-on.
"""

import logging
import os
import time

import aiohttp

_LOG = logging.getLogger("goodwe.hass")

CORE_API = "http://supervisor/core/api"
SUPERVISOR_API = "http://supervisor"
BAD_STATES = ("unknown", "unavailable", "none", "")


class HomeAssistant:
    def __init__(self, session: aiohttp.ClientSession, token: str | None = None):
        self.session = session
        self.token = token or os.environ.get("SUPERVISOR_TOKEN", "")
        self._last_moan: dict[str, float] = {}
        if not self.token:
            _LOG.error("SUPERVISOR_TOKEN missing - is this running as an add-on?")

    def _moan(self, key: str, msg: str, *args) -> None:
        """Log a recurring failure at most once a minute.

        ⚠️ The control loop retries every second, so an unthrottled warning here
        writes six lines a second forever - which rolls the add-on's log buffer
        and destroys exactly the startup diagnostics an installer needs. A fault
        that repeats is not more informative for being repeated.
        """
        now = time.monotonic()
        if now - self._last_moan.get(key, -999) >= 60:
            self._last_moan[key] = now
            _LOG.warning(msg, *args)

    @property
    def _headers(self) -> dict:
        return {"Authorization": f"Bearer {self.token}", "Content-Type": "application/json"}

    async def state(self, entity_id: str) -> dict | None:
        """Full state object, or None if it does not exist.

        ⚠️ A non-existent entity is not an error anywhere in Home Assistant - it
        simply never produces a value. On the reference install two safety
        alarms pointed at entity ids that did not exist and were therefore dead
        for a day while looking perfectly healthy. So: None here is always
        surfaced to the operator, never treated as zero.
        """
        if not entity_id:
            return None
        try:
            async with self.session.get(
                f"{CORE_API}/states/{entity_id}", headers=self._headers, timeout=10
            ) as resp:
                if resp.status == 404:
                    return None
                resp.raise_for_status()
                return await resp.json()
        except (aiohttp.ClientError, TimeoutError) as err:
            self._moan(f"read:{entity_id}", "read %s failed: %s", entity_id, err)
            return None

    async def number(self, entity_id: str, invert: bool = False) -> float | None:
        """Numeric state, or None. Never substitutes a default."""
        obj = await self.state(entity_id)
        if obj is None:
            return None
        raw = str(obj.get("state", "")).strip().lower()
        if raw in BAD_STATES:
            return None
        try:
            value = float(raw)
        except ValueError:
            self._moan(f"nan:{entity_id}", "%s is not numeric: %r", entity_id, raw)
            return None
        return -value if invert else value

    async def limits(self, entity_id: str) -> tuple[float | None, float | None, float | None]:
        """(min, max, step) of a number entity, so we never send out of range.

        ⚠️ Worth the extra call. On the reference install the control range and
        the firmware clamp disagreed (±1500 asked, ±500 enforced) and the result
        was silent: Home Assistant reported one value while the wire carried
        another, with no error anywhere.
        """
        obj = await self.state(entity_id)
        if obj is None:
            return (None, None, None)
        attrs = obj.get("attributes", {})
        def _f(key):
            try:
                return float(attrs[key])
            except (KeyError, TypeError, ValueError):
                return None
        return (_f("min"), _f("max"), _f("step"))

    async def call(self, domain: str, service: str, data: dict) -> bool:
        try:
            async with self.session.post(
                f"{CORE_API}/services/{domain}/{service}",
                headers=self._headers, json=data, timeout=10,
            ) as resp:
                if resp.status >= 400:
                    body = await resp.text()
                    # 400 here usually means out-of-range for the entity - the
                    # value is rejected outright, not clamped. Always log it:
                    # silently dropped commands are how a controller ends up
                    # believing something the hardware never did.
                    self._moan(f"svc:{domain}.{service}",
                               "service %s.%s rejected (%s): %s",
                               domain, service, resp.status, body[:200])
                    return False
                return True
        except (aiohttp.ClientError, TimeoutError) as err:
            self._moan(f"svcerr:{domain}.{service}", "service %s.%s failed: %s",
                       domain, service, err)
            return False

    async def set_number(self, entity_id: str, value: float) -> bool:
        return await self.call("number", "set_value",
                               {"entity_id": entity_id, "value": value})

    async def mqtt_service(self) -> dict | None:
        """Broker details from the Supervisor, if an MQTT service is available."""
        try:
            async with self.session.get(
                f"{SUPERVISOR_API}/services/mqtt", headers=self._headers, timeout=10
            ) as resp:
                if resp.status != 200:
                    return None
                payload = await resp.json()
                return payload.get("data")
        except (aiohttp.ClientError, TimeoutError):
            return None