156 lines
4.8 KiB
Python
156 lines
4.8 KiB
Python
from __future__ import annotations
|
|
|
|
import base64
|
|
import csv
|
|
import io
|
|
import logging
|
|
import secrets
|
|
from contextlib import asynccontextmanager
|
|
from datetime import UTC, datetime, time, timedelta
|
|
|
|
from fastapi import FastAPI, Query, Request
|
|
from fastapi.responses import HTMLResponse, PlainTextResponse, Response
|
|
|
|
from .config import Settings
|
|
from .monitor import TapoMonitor
|
|
from .storage import Storage, from_db_time, to_db_time
|
|
from .web import DASHBOARD_HTML
|
|
|
|
logging.basicConfig(level=logging.INFO, format="%(asctime)s | %(levelname)s | %(message)s")
|
|
settings = Settings.from_env()
|
|
storage = Storage(settings.database_path)
|
|
monitor = TapoMonitor(settings, storage)
|
|
|
|
|
|
@asynccontextmanager
|
|
async def lifespan(_: FastAPI):
|
|
storage.open()
|
|
await monitor.start()
|
|
try:
|
|
yield
|
|
finally:
|
|
await monitor.stop()
|
|
storage.close()
|
|
|
|
|
|
app = FastAPI(title="Tapo Pump Monitor", docs_url=None, redoc_url=None, lifespan=lifespan)
|
|
|
|
|
|
@app.middleware("http")
|
|
async def protect_dashboard(request: Request, call_next):
|
|
if request.url.path == "/health" or not settings.dashboard_password:
|
|
return await call_next(request)
|
|
|
|
header = request.headers.get("Authorization", "")
|
|
valid = False
|
|
if header.startswith("Basic "):
|
|
try:
|
|
decoded = base64.b64decode(header[6:], validate=True).decode("utf-8")
|
|
username, password = decoded.split(":", 1)
|
|
valid = secrets.compare_digest(username, settings.dashboard_username) and secrets.compare_digest(
|
|
password, settings.dashboard_password
|
|
)
|
|
except (ValueError, UnicodeDecodeError):
|
|
valid = False
|
|
if not valid:
|
|
return Response(status_code=401, headers={"WWW-Authenticate": 'Basic realm="Pump monitor"'})
|
|
return await call_next(request)
|
|
|
|
|
|
def local_day_bounds() -> tuple[datetime, datetime]:
|
|
now_local = datetime.now(settings.timezone)
|
|
start_local = datetime.combine(now_local.date(), time.min, tzinfo=settings.timezone)
|
|
return start_local.astimezone(UTC), (start_local + timedelta(days=1)).astimezone(UTC)
|
|
|
|
|
|
@app.get("/", response_class=HTMLResponse)
|
|
async def dashboard() -> str:
|
|
return DASHBOARD_HTML
|
|
|
|
|
|
@app.get("/health")
|
|
async def health() -> dict:
|
|
return {"service": "ok", "plug_state": monitor.state}
|
|
|
|
|
|
@app.get("/api/status")
|
|
async def api_status() -> dict:
|
|
day_start, day_end = local_day_bounds()
|
|
return {
|
|
"monitor": monitor.snapshot(),
|
|
"today": storage.summary_between(day_start, day_end),
|
|
"settings": {
|
|
"start_watts": settings.start_watts,
|
|
"stop_watts": settings.stop_watts,
|
|
"poll_interval_seconds": settings.poll_interval_seconds,
|
|
"timezone": settings.timezone_name,
|
|
},
|
|
"security": {"dashboard_auth_enabled": bool(settings.dashboard_password)},
|
|
}
|
|
|
|
|
|
@app.get("/api/cycles")
|
|
async def api_cycles(limit: int = Query(default=100, ge=1, le=1000)) -> dict:
|
|
return {"cycles": storage.recent_cycles(limit)}
|
|
|
|
|
|
@app.get("/api/history")
|
|
async def api_history(hours: int = Query(default=24, ge=1, le=168)) -> dict:
|
|
window_end = datetime.now(UTC)
|
|
window_start = window_end - timedelta(hours=hours)
|
|
cycles = storage.cycles_between(window_start, window_end)
|
|
preserve_ranges = [
|
|
(
|
|
from_db_time(cycle["started_at"]),
|
|
from_db_time(cycle["ended_at"]) if cycle["ended_at"] else window_end,
|
|
)
|
|
for cycle in cycles
|
|
]
|
|
return {
|
|
"window_start": to_db_time(window_start),
|
|
"window_end": to_db_time(window_end),
|
|
"samples": storage.sample_history(
|
|
window_start,
|
|
window_end,
|
|
max_points=800,
|
|
preserve_ranges=preserve_ranges,
|
|
),
|
|
"cycles": cycles,
|
|
}
|
|
|
|
|
|
@app.get("/api/cycles.csv")
|
|
async def export_cycles(days: int = Query(default=30, ge=1, le=3650)) -> Response:
|
|
end = datetime.now(UTC)
|
|
rows = storage.cycles_between(end - timedelta(days=days), end)
|
|
output = io.StringIO()
|
|
writer = csv.writer(output)
|
|
writer.writerow(
|
|
[
|
|
"started_at_utc",
|
|
"ended_at_utc",
|
|
"duration_seconds",
|
|
"average_watts",
|
|
"maximum_watts",
|
|
"energy_wh",
|
|
"partial_cycle",
|
|
]
|
|
)
|
|
for row in reversed(rows):
|
|
writer.writerow(
|
|
[
|
|
row["started_at"],
|
|
row["ended_at"],
|
|
round(float(row["duration_seconds"] or 0), 3),
|
|
round(float(row["average_watts"] or 0), 3),
|
|
round(float(row["maximum_watts"] or 0), 3),
|
|
round(float(row["energy_wh"] or 0), 5),
|
|
bool(row["incomplete_start"] or row["incomplete_end"]),
|
|
]
|
|
)
|
|
return PlainTextResponse(
|
|
output.getvalue(),
|
|
media_type="text/csv",
|
|
headers={"Content-Disposition": 'attachment; filename="pump-cycles.csv"'},
|
|
)
|