feat: add DknCloudNaClient stub

This commit is contained in:
Sebastien Lavoie
2026-03-29 08:44:43 -04:00
parent 17e44e4546
commit 0beaa71fdc
+125
View File
@@ -0,0 +1,125 @@
"""DKN Cloud NA API client stub.
Real HTTP calls are implemented in a future phase. This stub defines the
interface and returns hardcoded data so the rest of the integration can be
developed and tested without live credentials.
"""
from __future__ import annotations
from typing import Any
from aiohttp import ClientSession
from .const import LOGGER
class DknAuthError(Exception):
"""Raised when authentication fails (401)."""
class DknConnectionError(Exception):
"""Raised when the API cannot be reached."""
class DknCloudNaClient:
"""Async client for the DKN Cloud NA REST API.
Instantiated once per config entry. Password is cleared after login and
never stored beyond the initial token exchange.
"""
def __init__(
self,
username: str,
session: ClientSession,
*,
password: str | None = None,
token: str | None = None,
refresh_token: str | None = None,
) -> None:
self._username = username
self._session = session
self._password = password
self.token = token
self.refresh_token = refresh_token
def clear_password(self) -> None:
"""Discard password from memory after token exchange."""
self._password = None
async def login(self) -> None:
"""Exchange email+password for access and refresh tokens.
Sets self.token and self.refresh_token on success.
Raises DknAuthError on bad credentials, DknConnectionError on network failure.
"""
# TODO: implement real POST to API_LOGIN
LOGGER.debug("DknCloudNaClient.login() stub called for %s", self._username)
raise NotImplementedError("login() not yet implemented")
async def is_logged_in(self) -> bool:
"""Return True if the current token is still valid."""
# TODO: implement real GET to API_IS_LOGGED_IN
raise NotImplementedError("is_logged_in() not yet implemented")
async def refresh_access_token(self) -> None:
"""Use the refresh token to obtain a new access token.
Updates self.token on success.
Raises DknAuthError if the refresh token is also expired.
"""
# TODO: implement real GET to API_REFRESH_TOKEN
raise NotImplementedError("refresh_access_token() not yet implemented")
async def fetch_installations(self) -> list[dict[str, Any]]:
"""Return all installations and their devices.
Each installation contains a list of DeviceInfo dicts.
Returns stub data for scaffolding.
"""
LOGGER.debug("DknCloudNaClient.fetch_installations() stub — returning fake data")
return [
{
"_id": "stub-installation-1",
"name": "My Home",
"devices": [
{
"mac": "aa:bb:cc:dd:ee:ff",
"name": "Living Room AC",
"power": False,
"mode": 1,
"real_mode": 1,
"work_temp": 22.0,
"ext_temp": 18.0,
"units": 0,
"setpoint_air_auto": 22.0,
"setpoint_air_cool": 24.0,
"setpoint_air_heat": 20.0,
"range_sp_auto_air_min": 16,
"range_sp_auto_air_max": 32,
"range_sp_cool_air_min": 16,
"range_sp_cool_air_max": 32,
"range_sp_hot_air_min": 16,
"range_sp_hot_air_max": 32,
"speed_state": 0,
"speed_available": [0, 2, 3, 4, 5, 6],
"slats_vertical_1": 0,
"machineready": True,
"isConnected": True,
"tsensor_error": False,
"stat_rssi": -65,
"stat_ssid": "MyWiFi",
"version": "1.0.0",
"error_value": 0,
"error_ascii1": "",
"error_ascii2": "",
}
],
}
]
def __repr__(self) -> str:
first = self._username[0] if self._username else "?"
token_state = "set" if self.token else "none"
return f"DknCloudNaClient(u={first}***, token={token_state})"