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", "")
# What the Supervisor's proxy answers while Home Assistant restarts.
UNAVAILABLE = (502, 503, 504)


class Outage:
    """UX-10: Home Assistant not answering, said in one plain line per outage.

    A restart takes the REST API and the meter websocket down together, and
    each used to log the library's own text for every entity and service
    ("502, message='Bad Gateway', url=URL(...)", "Received message 8:1000 is
    not str"). Now the first channel to fail says it once at WARNING, the raw
    text goes to DEBUG, and the last channel to recover says so at INFO.
    """

    def __init__(self):
        self.down_on = set()

    def down(self, channel: str, what: str, detail) -> None:
        if not self.down_on:
            _LOG.warning("Home Assistant is not answering (%s), perhaps it is "
                         "restarting. The add-on keeps trying.", what)
        self.down_on.add(channel)
        _LOG.debug("Home Assistant not answering on %s: %s", channel, detail)

    def up(self, channel: str) -> None:
        if channel in self.down_on:
            self.down_on.discard(channel)
            if not self.down_on:
                _LOG.info("Home Assistant answers again.")


OUTAGE = Outage()


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 in UNAVAILABLE:
                    OUTAGE.down("api", f"error {resp.status}",
                                f"read {entity_id}: {resp.status}")
                    return None
                OUTAGE.up("api")
                if resp.status == 404:
                    return None
                resp.raise_for_status()
                obj = await resp.json()
        except aiohttp.ClientConnectionError as err:
            OUTAGE.down("api", "no connection", f"read {entity_id}: {err}")
            return None
        # SAFETY-61: an undecodable body is an absence like any other failed
        # read, never an exception into the control loop.
        except (aiohttp.ClientError, TimeoutError, ValueError) as err:
            self._moan(f"read:{entity_id}", "read %s failed: %s", entity_id, err)
            return None
        if not isinstance(obj, dict):
            self._moan(f"shape:{entity_id}", "read %s: not a state object: %.80r",
                       entity_id, obj)
            return None
        return obj

    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 in UNAVAILABLE:
                    OUTAGE.down("api", f"error {resp.status}",
                                f"service {domain}.{service}: {resp.status}")
                    return False
                OUTAGE.up("api")
                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.ClientConnectionError as err:
            OUTAGE.down("api", "no connection",
                        f"service {domain}.{service}: {err}")
            return False
        # ValueError: an undecodable error body (SAFETY-61) is a failed write.
        except (aiohttp.ClientError, TimeoutError, ValueError) 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