102 lines
3.0 KiB
Python
102 lines
3.0 KiB
Python
"""
|
|
Demeter Server — Dashboard Routes
|
|
|
|
Serves the web UI using Jinja2 templates with DaisyUI (Tailwind CSS).
|
|
Auto-refresh is handled by periodic fetch() calls in the templates.
|
|
|
|
Routes:
|
|
GET / — redirect to dashboard
|
|
GET /dashboard — main device overview
|
|
GET /dashboard/devices/{device_id} — device detail page
|
|
GET /dashboard/api/readings — JSON readings for auto-refresh
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import time
|
|
from typing import Any
|
|
|
|
from fastapi import APIRouter, Request
|
|
from fastapi.responses import HTMLResponse, RedirectResponse
|
|
from fastapi.templating import Jinja2Templates
|
|
|
|
from .config import TEMPLATES_DIR
|
|
|
|
router = APIRouter(tags=["dashboard"])
|
|
templates = Jinja2Templates(directory=str(TEMPLATES_DIR))
|
|
|
|
|
|
def _get_store(request: Request):
|
|
return request.app.state.store
|
|
|
|
|
|
def _get_observer(request: Request):
|
|
return request.app.state.observer
|
|
|
|
|
|
@router.get("/", include_in_schema=False)
|
|
async def root():
|
|
"""Redirect root to dashboard."""
|
|
return RedirectResponse(url="/dashboard")
|
|
|
|
|
|
@router.get("/dashboard", response_class=HTMLResponse)
|
|
async def dashboard(request: Request):
|
|
"""Main dashboard showing all devices and their latest readings."""
|
|
store = _get_store(request)
|
|
observer = _get_observer(request)
|
|
|
|
devices = await store.get_all_devices()
|
|
settings = request.app.state.settings
|
|
|
|
return templates.TemplateResponse(
|
|
request,
|
|
"dashboard.html",
|
|
{
|
|
"devices": devices,
|
|
"server_version": settings.app_version,
|
|
"active_subscriptions": observer.active_subscriptions,
|
|
"timestamp": time.time(),
|
|
},
|
|
)
|
|
|
|
|
|
@router.get("/dashboard/devices/{device_id}", response_class=HTMLResponse)
|
|
async def device_detail(device_id: str, request: Request):
|
|
"""Detailed view for a single device."""
|
|
store = _get_store(request)
|
|
|
|
device = await store.get_device(device_id)
|
|
if device is None:
|
|
return templates.TemplateResponse(
|
|
request,
|
|
"dashboard.html",
|
|
{
|
|
"devices": await store.get_all_devices(),
|
|
"server_version": request.app.state.settings.app_version,
|
|
"active_subscriptions": _get_observer(request).active_subscriptions,
|
|
"timestamp": time.time(),
|
|
"error": f"Device {device_id!r} not found",
|
|
},
|
|
)
|
|
|
|
return templates.TemplateResponse(
|
|
request,
|
|
"device_detail.html",
|
|
{
|
|
"device": device,
|
|
"server_version": request.app.state.settings.app_version,
|
|
},
|
|
)
|
|
|
|
|
|
@router.get("/dashboard/api/readings")
|
|
async def dashboard_readings(request: Request) -> dict[str, Any]:
|
|
"""JSON endpoint for dashboard auto-refresh polling."""
|
|
store = _get_store(request)
|
|
snapshot = await store.snapshot()
|
|
return {
|
|
"devices": snapshot,
|
|
"timestamp": time.time(),
|
|
}
|