54 lines
1.8 KiB
Python
54 lines
1.8 KiB
Python
from __future__ import annotations
|
|
|
|
from datetime import datetime, timezone
|
|
import logging
|
|
from time import perf_counter
|
|
|
|
from app.clients.base import BaseHTTPClient
|
|
from app.config import Settings
|
|
from app.models.sources import HomeAssistantSnapshot
|
|
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class HomeAssistantClient(BaseHTTPClient):
|
|
def __init__(self, settings: Settings) -> None:
|
|
super().__init__(settings, "home-assistant", settings.home_assistant_base_url)
|
|
|
|
async def fetch_status(self) -> HomeAssistantSnapshot:
|
|
snapshot = HomeAssistantSnapshot()
|
|
if not self.base_url or not self.settings.home_assistant_token:
|
|
logger.info("home assistant skipped: base URL or token missing")
|
|
return snapshot
|
|
|
|
headers = {
|
|
"Authorization": f"Bearer {self.settings.home_assistant_token}",
|
|
"Content-Type": "application/json",
|
|
}
|
|
|
|
started_at = perf_counter()
|
|
api_info = await self._request_json("GET", "/api/", headers=headers)
|
|
if api_info is None:
|
|
logger.warning("home assistant base API check failed")
|
|
return snapshot
|
|
|
|
logger.info("home assistant raw /api response: %s", api_info)
|
|
|
|
elapsed_ms = int((perf_counter() - started_at) * 1000)
|
|
config = await self._request_json("GET", "/api/config", headers=headers)
|
|
logger.info("home assistant raw /api/config response: %s", config)
|
|
|
|
version = None
|
|
if isinstance(config, dict):
|
|
version = config.get("version")
|
|
|
|
normalized = HomeAssistantSnapshot(
|
|
status="online",
|
|
version=str(version) if version else None,
|
|
response_time_ms=elapsed_ms,
|
|
last_checked=datetime.now(timezone.utc),
|
|
)
|
|
logger.info("home assistant normalized snapshot: %s", normalized.model_dump())
|
|
return normalized
|