refactor: complete rewrite as React+FastAPI dashboard
Replace monolithic Jinja2 template with modern stack: Backend (FastAPI): - Modular router/service architecture - Async PostgreSQL (asyncpg) for news from n8n pipeline - Live Unraid server stats (2 servers via API) - Home Assistant, Vikunja tasks, weather (wttr.in) - WebSocket broadcast for real-time updates (15s) - TTL cache per endpoint, all config via ENV vars Frontend (React + Vite + TypeScript): - Glassmorphism dark theme with Tailwind CSS - Responsive grid: mobile/tablet/desktop/ultrawide - Weather cards, hourly forecast, news with category tabs - Server stats (CPU ring, RAM bar, Docker list) - Home Assistant controls, task management - Live clock, WebSocket connection indicator Infrastructure: - Multi-stage Dockerfile (node:22-alpine + python:3.11-slim) - docker-compose with full ENV configuration - Kaniko CI/CD pipeline for GitLab registry Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
parent
4bbc125a67
commit
9f7330e217
48 changed files with 6390 additions and 1461 deletions
149
server/services/ha_service.py
Normal file
149
server/services/ha_service.py
Normal file
|
|
@ -0,0 +1,149 @@
|
|||
from __future__ import annotations
|
||||
|
||||
import httpx
|
||||
from typing import Any, Dict, List, Optional
|
||||
|
||||
|
||||
def _friendly_name(entity: Dict[str, Any]) -> str:
|
||||
"""Extract the friendly name from an entity's attributes, falling back to entity_id."""
|
||||
attrs = entity.get("attributes", {})
|
||||
return attrs.get("friendly_name", entity.get("entity_id", "unknown"))
|
||||
|
||||
|
||||
def _parse_light(entity: Dict[str, Any]) -> Dict[str, Any]:
|
||||
"""Parse a light entity into a normalised dictionary."""
|
||||
attrs = entity.get("attributes", {})
|
||||
state = entity.get("state", "unknown")
|
||||
brightness_raw = attrs.get("brightness")
|
||||
brightness_pct: Optional[int] = None
|
||||
if brightness_raw is not None:
|
||||
try:
|
||||
brightness_pct = round(int(brightness_raw) / 255 * 100)
|
||||
except (ValueError, TypeError):
|
||||
brightness_pct = None
|
||||
|
||||
return {
|
||||
"entity_id": entity.get("entity_id", ""),
|
||||
"name": _friendly_name(entity),
|
||||
"state": state,
|
||||
"brightness": brightness_pct,
|
||||
"color_mode": attrs.get("color_mode"),
|
||||
}
|
||||
|
||||
|
||||
def _parse_cover(entity: Dict[str, Any]) -> Dict[str, Any]:
|
||||
"""Parse a cover entity into a normalised dictionary."""
|
||||
attrs = entity.get("attributes", {})
|
||||
return {
|
||||
"entity_id": entity.get("entity_id", ""),
|
||||
"name": _friendly_name(entity),
|
||||
"state": entity.get("state", "unknown"),
|
||||
"current_position": attrs.get("current_position"),
|
||||
}
|
||||
|
||||
|
||||
def _parse_sensor(entity: Dict[str, Any]) -> Dict[str, Any]:
|
||||
"""Parse a temperature sensor entity into a normalised dictionary."""
|
||||
attrs = entity.get("attributes", {})
|
||||
state_value = entity.get("state", "unknown")
|
||||
try:
|
||||
state_value = round(float(state_value), 1)
|
||||
except (ValueError, TypeError):
|
||||
pass
|
||||
|
||||
return {
|
||||
"entity_id": entity.get("entity_id", ""),
|
||||
"name": _friendly_name(entity),
|
||||
"state": state_value,
|
||||
"unit": attrs.get("unit_of_measurement", ""),
|
||||
"device_class": attrs.get("device_class", ""),
|
||||
}
|
||||
|
||||
|
||||
async def fetch_ha_data(url: str, token: str) -> Dict[str, Any]:
|
||||
"""Fetch and categorise entity states from a Home Assistant instance.
|
||||
|
||||
Args:
|
||||
url: Base URL of the Home Assistant instance (e.g. ``http://192.168.1.100:8123``).
|
||||
token: Long-lived access token for authentication.
|
||||
|
||||
Returns:
|
||||
Dictionary containing:
|
||||
- ``online``: Whether the HA instance is reachable.
|
||||
- ``lights``: List of light entities with state and brightness.
|
||||
- ``covers``: List of cover entities with state and position.
|
||||
- ``sensors``: List of temperature sensor entities.
|
||||
- ``lights_on``: Count of lights currently in the ``on`` state.
|
||||
- ``lights_total``: Total number of light entities.
|
||||
- ``error``: Error message if the request failed, else ``None``.
|
||||
"""
|
||||
result: Dict[str, Any] = {
|
||||
"online": False,
|
||||
"lights": [],
|
||||
"covers": [],
|
||||
"sensors": [],
|
||||
"lights_on": 0,
|
||||
"lights_total": 0,
|
||||
"error": None,
|
||||
}
|
||||
|
||||
if not url or not token:
|
||||
result["error"] = "Missing Home Assistant URL or token"
|
||||
return result
|
||||
|
||||
base_url = url.rstrip("/")
|
||||
headers = {
|
||||
"Authorization": f"Bearer {token}",
|
||||
"Content-Type": "application/json",
|
||||
}
|
||||
|
||||
try:
|
||||
async with httpx.AsyncClient(timeout=15, verify=False) as client:
|
||||
resp = await client.get(f"{base_url}/api/states", headers=headers)
|
||||
resp.raise_for_status()
|
||||
entities: List[Dict[str, Any]] = resp.json()
|
||||
except httpx.HTTPStatusError as exc:
|
||||
result["error"] = f"HTTP {exc.response.status_code}"
|
||||
return result
|
||||
except httpx.RequestError as exc:
|
||||
result["error"] = f"Connection failed: {exc}"
|
||||
return result
|
||||
except Exception as exc:
|
||||
result["error"] = str(exc)
|
||||
return result
|
||||
|
||||
result["online"] = True
|
||||
|
||||
lights: List[Dict[str, Any]] = []
|
||||
covers: List[Dict[str, Any]] = []
|
||||
sensors: List[Dict[str, Any]] = []
|
||||
|
||||
for entity in entities:
|
||||
entity_id: str = entity.get("entity_id", "")
|
||||
domain = entity_id.split(".")[0] if "." in entity_id else ""
|
||||
attrs = entity.get("attributes", {})
|
||||
state = entity.get("state", "")
|
||||
|
||||
if state in ("unavailable", "unknown"):
|
||||
continue
|
||||
|
||||
if domain == "light":
|
||||
lights.append(_parse_light(entity))
|
||||
|
||||
elif domain == "cover":
|
||||
covers.append(_parse_cover(entity))
|
||||
|
||||
elif domain == "sensor":
|
||||
device_class = attrs.get("device_class", "")
|
||||
if device_class == "temperature":
|
||||
sensors.append(_parse_sensor(entity))
|
||||
|
||||
lights_on = sum(1 for light in lights if light["state"] == "on")
|
||||
|
||||
result["lights"] = lights
|
||||
result["covers"] = covers
|
||||
result["sensors"] = sensors
|
||||
result["lights_on"] = lights_on
|
||||
result["lights_total"] = len(lights)
|
||||
|
||||
return result
|
||||
Loading…
Add table
Add a link
Reference in a new issue