65 lines
1.9 KiB
Python
65 lines
1.9 KiB
Python
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
from aiohttp import ClientError, ClientSession, ClientTimeout
|
|
|
|
from .const import DEFAULT_TIMEOUT
|
|
|
|
|
|
class DualPidApiError(Exception):
|
|
"""Base API error."""
|
|
|
|
|
|
class DualPidApiConnectionError(DualPidApiError):
|
|
"""Raised when the API cannot be reached."""
|
|
|
|
|
|
class DualPidApiClient:
|
|
"""Simple client for the Dual PID HTTP API."""
|
|
|
|
def __init__(
|
|
self,
|
|
session: ClientSession,
|
|
host: str,
|
|
port: int,
|
|
timeout: int = DEFAULT_TIMEOUT,
|
|
) -> None:
|
|
self._session = session
|
|
self._host = host.strip().rstrip("/")
|
|
self._port = port
|
|
self._timeout = timeout
|
|
|
|
@property
|
|
def base_url(self) -> str:
|
|
return f"http://{self._host}:{self._port}"
|
|
|
|
async def _async_request(self, method: str, path: str) -> dict[str, Any]:
|
|
url = f"{self.base_url}{path}"
|
|
try:
|
|
response = await self._session.request(
|
|
method,
|
|
url,
|
|
timeout=ClientTimeout(total=self._timeout),
|
|
)
|
|
response.raise_for_status()
|
|
payload = await response.json()
|
|
except (ClientError, TimeoutError, ValueError) as err:
|
|
raise DualPidApiConnectionError(f"Request to {url} failed: {err}") from err
|
|
|
|
if not isinstance(payload, dict):
|
|
raise DualPidApiError("API response is not a JSON object")
|
|
return payload
|
|
|
|
async def async_get_status(self) -> dict[str, Any]:
|
|
return await self._async_request("GET", "/api/status")
|
|
|
|
async def async_start(self) -> dict[str, Any]:
|
|
return await self._async_request("POST", "/api/start")
|
|
|
|
async def async_standby_on(self) -> dict[str, Any]:
|
|
return await self._async_request("POST", "/api/standby/on")
|
|
|
|
async def async_standby_off(self) -> dict[str, Any]:
|
|
return await self._async_request("POST", "/api/standby/off")
|