api.py

from __future__ import annotations

import asyncio
import json
import random
import re
import time
from dataclasses import dataclass
from typing import Any, Callable, Iterable
from urllib.parse import quote

import aiohttp

from .const import MAX_TEMP_C, MIN_TEMP_C, NS, STEP_TEMP_C

ID_SETPOINT = 0
ID_SET_REQ = 66


def _round_to_step(value: float, step: float) -> float:
    return round(value / step) * step


def _clamp(value: float, lo: float, hi: float) -> float:
    return max(lo, min(hi, value))


def _c_to_raw_tenths(c: float) -> int:
    return int(round(c * 10.0))


def _raw_tenths_to_c(v: int) -> float:
    return v / 10.0


def _build_req66(temp_c: float, addr: int) -> int:
    raw = _c_to_raw_tenths(temp_c) & 1023
    return raw | ((addr & 0xFF) << 10)


_RE_SID = re.compile(r'"sid"\s*:\s*"([^"]+)"')


def _extract_sid(payload: str) -> str | None:
    m = _RE_SID.search(payload)
    return m.group(1) if m else None


def _decode_engineio_payload(text: str) -> list[str]:
    """Decode Engine.IO 'len:packet' framing; fall back to a single packet."""
    pkts: list[str] = []
    i = 0
    n = len(text)
    while i < n:
        if not text[i].isdigit():
            if pkts:
                i += 1
                continue
            return [text]
        j = i
        while j < n and text[j].isdigit():
            j += 1
        if j >= n or text[j] != ":":
            return [text]
        try:
            ln = int(text[i:j])
        except ValueError:
            return [text]
        start = j + 1
        end = start + ln
        if end > n:
            return [text]
        pkts.append(text[start:end])
        i = end
    return pkts or [text]


_RE_SIO_EVENT = re.compile(r"42(?:/1\.0\.0,)?(\[[\s\S]*?\])")


def _parse_socketio_events(packets: Iterable[str]) -> list[tuple[str, Any]]:
    out: list[tuple[str, Any]] = []
    for pkt in packets:
        for m in _RE_SIO_EVENT.finditer(pkt):
            try:
                arr = json.loads(m.group(1))
                if isinstance(arr, list) and len(arr) >= 1 and isinstance(arr[0], str):
                    event_name = arr[0]
                    data = arr[1] if len(arr) > 1 else None
                    out.append((event_name, data))
            except json.JSONDecodeError:
                continue
    return out


def _root_connect_packet() -> str:
    return "40"


def _ns_connect_packet() -> str:
    return f"40/{NS},"


def _ns_event_packet(event: str, data: Any) -> str:
    return f"42/{NS},{json.dumps([event, data], separators=(',', ':'))}"


def _ns_message_packet(command_obj: dict[str, Any]) -> str:
    return _ns_event_packet("message", command_obj)


def _disconnect_root_packet() -> str:
    return "41"


def _disconnect_ns_packet() -> str:
    return f"41/{NS},"


@dataclass(frozen=True)
class StiebelReadback:
    setpoint_c: float
    is_valid: bool | None


class StiebelDheApi:
    def __init__(
        self,
        session: aiohttp.ClientSession,
        host: str,
        token: str,
        name: str,
        token_update_cb: Callable[[str], None] | None = None,
    ) -> None:
        self._session = session
        self._host = host.rstrip("/")
        self._token = token
        self._name = name
        self._token_update_cb = token_update_cb

    @property
    def token(self) -> str:
        return self._token

    def update_token(self, token: str) -> None:
        self._token = token
        if self._token_update_cb:
            self._token_update_cb(token)

    async def _get_text(self, url: str, *, timeout_s: float = 30.0) -> str:
        timeout = aiohttp.ClientTimeout(total=timeout_s)
        async with self._session.get(url, timeout=timeout) as resp:
            txt = await resp.text()
            resp.raise_for_status()
            return txt

    async def _post_packet(self, url: str, packet: str, *, timeout_s: float = 30.0) -> str:
        body = f"{len(packet)}:{packet}"
        timeout = aiohttp.ClientTimeout(total=timeout_s)
        async with self._session.post(
            url,
            data=body,
            headers={"Content-Type": "text/plain;charset=UTF-8"},
            timeout=timeout,
        ) as resp:
            txt = await resp.text()
            resp.raise_for_status()
            return txt

    async def _connect(self) -> tuple[str, Callable[[], str]]:
        # Engine.IO open (polling)
        t = format(int(time.time() * 1000), "x")
        open_url = (
            f"{self._host}/socket.io/?EIO=3&transport=polling&token={quote(self._token)}&t={t}"
        )
        open_payload = await self._get_text(open_url, timeout_s=15.0)
        sid = _extract_sid(open_payload)
        if not sid:
            raise RuntimeError(f"Could not extract sid from open payload: {open_payload[:200]}")

        base = (
            f"{self._host}/socket.io/?EIO=3&transport=polling&sid={quote(sid)}"
            f"&token={quote(self._token)}"
        )

        def poll_url() -> str:
            return f"{base}&t={format(int(time.time() * 1000), 'x')}"

        # Correct connect sequence: root, then namespace
        await self._post_packet(poll_url(), _root_connect_packet(), timeout_s=10.0)
        await self._post_packet(poll_url(), _ns_connect_packet(), timeout_s=10.0)

        # Auth + pairing
        await self._post_packet(
            poll_url(),
            _ns_event_packet("token_request", {"token": self._token, "name": self._name}),
            timeout_s=10.0,
        )

        authenticated = False
        paired = False
        deadline = time.time() + 30.0

        while time.time() < deadline and not (authenticated and paired):
            raw = await self._get_text(poll_url(), timeout_s=20.0)
            packets = _decode_engineio_payload(raw)
            events = _parse_socketio_events(packets)

            for event_name, data in events:
                if event_name == "token_response" and isinstance(data, str) and len(data) > 20:
                    # Persist refreshed token and authenticate with it.
                    self.update_token(data)
                    await self._post_packet(
                        poll_url(), _ns_event_packet("authenticate", {"token": self._token}), timeout_s=10.0
                    )
                elif event_name == "authenticated":
                    authenticated = True
                elif event_name == "pairing_result" and isinstance(data, dict) and data.get("result") is True:
                    paired = True

            if authenticated and not paired:
                await self._post_packet(
                    poll_url(),
                    _ns_event_packet("token_request", {"token": self._token, "name": self._name}),
                    timeout_s=10.0,
                )
                await asyncio.sleep(0.3)

        if not authenticated:
            raise RuntimeError("Auth timeout: did not receive authenticated=true.")
        if not paired:
            raise RuntimeError("Pairing timeout: did not receive pairing_result=true.")

        return sid, poll_url

    async def _disconnect(self, poll_url: Callable[[], str]) -> None:
        # Best-effort; device works fine without explicit disconnect.
        try:
            await self._post_packet(poll_url(), _disconnect_ns_packet(), timeout_s=5.0)
        except Exception:
            pass
        try:
            await self._post_packet(poll_url(), _disconnect_root_packet(), timeout_s=5.0)
        except Exception:
            pass

    async def get_setpoint(self) -> StiebelReadback:
        _, poll_url = await self._connect()
        try:
            await self._post_packet(
                poll_url(),
                _ns_message_packet(
                    {"command": "get:ste.common.odb:value", "value": {"id": ID_SETPOINT, "value": ""}}
                ),
                timeout_s=10.0,
            )

            deadline = time.time() + 8.0
            while time.time() < deadline:
                raw = await self._get_text(poll_url(), timeout_s=20.0)
                packets = _decode_engineio_payload(raw)
                events = _parse_socketio_events(packets)

                for event_name, data in events:
                    if (
                        event_name == "message"
                        and isinstance(data, dict)
                        and data.get("command") == "set:ste.common.odb:value"
                        and isinstance(data.get("value"), dict)
                        and data["value"].get("id") == ID_SETPOINT
                    ):
                        val = data["value"].get("value")
                        if isinstance(val, (int, float)):
                            setpoint_c = _raw_tenths_to_c(int(val))
                            is_valid = data["value"].get("isValid")
                            return StiebelReadback(setpoint_c=setpoint_c, is_valid=is_valid)
                await asyncio.sleep(0.05)
            raise RuntimeError("Timeout waiting for setpoint readback (ODB id 0).")
        finally:
            await self._disconnect(poll_url)

    async def set_setpoint(self, requested_c: float, *, max_attempts: int = 1) -> StiebelReadback:
        requested_c = _round_to_step(_clamp(float(requested_c), MIN_TEMP_C, MAX_TEMP_C), STEP_TEMP_C)
        _, poll_url = await self._connect()
        last: StiebelReadback | None = None
        try:
            for _attempt in range(max_attempts):
                addr = random.randint(1, 63)
                req_val = _build_req66(requested_c, addr)

                await self._post_packet(
                    poll_url(),
                    _ns_message_packet(
                        {
                            "command": "assign:ste.common.odb:value",
                            "value": {"id": ID_SET_REQ, "value": req_val},
                        }
                    ),
                    timeout_s=10.0,
                )
                await self._post_packet(
                    poll_url(),
                    _ns_message_packet(
                        {"command": "get:ste.common.odb:value", "value": {"id": ID_SETPOINT, "value": ""}}
                    ),
                    timeout_s=10.0,
                )

                deadline = time.time() + 8.0
                while time.time() < deadline:
                    raw = await self._get_text(poll_url(), timeout_s=20.0)
                    packets = _decode_engineio_payload(raw)
                    events = _parse_socketio_events(packets)

                    for event_name, data in events:
                        if (
                            event_name == "message"
                            and isinstance(data, dict)
                            and data.get("command") == "set:ste.common.odb:value"
                            and isinstance(data.get("value"), dict)
                            and data["value"].get("id") == ID_SETPOINT
                        ):
                            val = data["value"].get("value")
                            if isinstance(val, (int, float)):
                                setpoint_c = _raw_tenths_to_c(int(val))
                                is_valid = data["value"].get("isValid")
                                last = StiebelReadback(setpoint_c=setpoint_c, is_valid=is_valid)
                                if abs(setpoint_c - requested_c) < 0.01:
                                    return last
                    await asyncio.sleep(0.05)

                await asyncio.sleep(0.9)
            if last:
                raise RuntimeError(f"Not set. Readback: {last.setpoint_c:.1f}C (isValid={last.is_valid}).")
            raise RuntimeError("No readback received.")
        finally:
            await self._disconnect(poll_url)