LoraMapTester/server/fastapi_app.py

"""Alternate LoraTester entry point — same API as Flask."""

import time
from pathlib import Path
from typing import Any, Optional

from fastapi import FastAPI, Header, HTTPException, Query
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import FileResponse
from pydantic import BaseModel, Field

from core.auth import ANDROID_CLIENT_HEADER, ANDROID_CLIENT_VALUE
from core.config import HOST, PORT
from core.models import ChatIn, TelemetryIn
from core import storage
from core.telemetry_body import telemetry_from_body

STATIC_DIR = Path(__file__).resolve().parent / "static"

app = FastAPI(title="LoraTester")
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_methods=["*"],
    allow_headers=["*"],
)

storage.init_db()


class TelemetryBody(BaseModel):
    device_id: str
    lat: Optional[float] = None
    lon: Optional[float] = None
    rssi: Optional[float] = None
    range_m: Optional[float] = None
    raw_frame: Optional[str] = None
    meta: Optional[Any] = None
    fields: Optional[dict[str, Any]] = None
    role: Optional[str] = None
    ts: Optional[float] = None


class ChatBody(BaseModel):
    device_id: str
    text: str
    ts: Optional[float] = None


class TrackStartBody(BaseModel):
    device_id: str
    label: Optional[str] = None


class TrackPoint(BaseModel):
    ts: Optional[float] = None
    lat: float
    lon: float
    altitude_gps: Optional[float] = None
    rssi: Optional[float] = None
    role: Optional[str] = None
    meta: Optional[Any] = None


class TrackPointsBody(BaseModel):
    points: list[TrackPoint] = Field(default_factory=list)


@app.get("/")
def index():
    return FileResponse(
        STATIC_DIR / "index.html",
        headers={"Cache-Control": "no-store"},
    )


@app.post("/api/telemetry")
def post_telemetry(
    body: TelemetryBody,
    x_lora_client: Optional[str] = Header(None, alias=ANDROID_CLIENT_HEADER),
):
    if (x_lora_client or "").strip().lower() != ANDROID_CLIENT_VALUE:
        raise HTTPException(
            403,
            detail=f"telemetry only from Android app (header {ANDROID_CLIENT_HEADER})",
        )
    try:
        data = telemetry_from_body(body.model_dump(exclude_none=True))
        return storage.record_telemetry(data)
    except ValueError as e:
        raise HTTPException(400, detail=str(e)) from e


@app.get("/api/devices")
def get_devices():
    return storage.list_devices()


@app.get("/api/telemetry")
def get_telemetry_history(
    device_id: Optional[str] = None,
    limit: int = Query(100, ge=1, le=500),
    since: Optional[float] = None,
    until: Optional[float] = None,
    role: Optional[str] = None,
):
    return storage.get_telemetry(device_id, limit, since, until, role)


@app.get("/api/stats/history")
def get_stats_history(
    device_id: str = Query(...),
    limit: int = Query(50, ge=1, le=500),
    since: Optional[float] = None,
    until: Optional[float] = None,
    role: Optional[str] = None,
):
    return storage.get_telemetry(device_id, limit, since, until, role)


@app.post("/api/tracks/start")
def tracks_start(
    body: TrackStartBody,
    x_lora_client: Optional[str] = Header(None, alias=ANDROID_CLIENT_HEADER),
):
    _require_android(x_lora_client)
    try:
        return storage.start_track(body.device_id, body.label)
    except ValueError as e:
        raise HTTPException(400, detail=str(e)) from e


@app.post("/api/tracks/{track_id}/points")
def tracks_points(
    track_id: int,
    body: TrackPointsBody,
    x_lora_client: Optional[str] = Header(None, alias=ANDROID_CLIENT_HEADER),
):
    _require_android(x_lora_client)
    try:
        points = [p.model_dump(exclude_none=True) for p in body.points]
        return storage.add_track_points(track_id, points)
    except ValueError as e:
        raise HTTPException(400, detail=str(e)) from e


@app.post("/api/tracks/{track_id}/finish")
def tracks_finish(
    track_id: int,
    x_lora_client: Optional[str] = Header(None, alias=ANDROID_CLIENT_HEADER),
):
    _require_android(x_lora_client)
    try:
        return storage.finish_track(track_id)
    except ValueError as e:
        raise HTTPException(400, detail=str(e)) from e


@app.get("/api/tracks")
def tracks_list(
    device_id: Optional[str] = None,
    limit: int = Query(50, ge=1, le=200),
):
    return storage.list_tracks(device_id, limit)


@app.get("/api/tracks/{track_id}")
def tracks_get(track_id: int):
    try:
        return storage.get_track(track_id)
    except ValueError as e:
        raise HTTPException(404, detail=str(e)) from e


def _require_android(x_lora_client: Optional[str]) -> None:
    if (x_lora_client or "").strip().lower() != ANDROID_CLIENT_VALUE:
        raise HTTPException(
            403,
            detail=f"Android header {ANDROID_CLIENT_HEADER} required",
        )


@app.post("/api/chat")
def post_chat(body: ChatBody):
    text = body.text.strip()
    if not text:
        raise HTTPException(400, "text required")
    data = ChatIn(device_id=body.device_id, text=text, ts=body.ts)
    return storage.add_chat(data)


@app.get("/api/chat")
def get_chat(since: float = 0, limit: int = Query(200, ge=1, le=500)):
    return storage.get_chat(since, limit)


@app.get("/api/health")
def health():
    status = storage.db_status()
    return {"ok": status["db_ok"], "ts": time.time(), **status}


if __name__ == "__main__":
    import uvicorn

    uvicorn.run("fastapi_app:app", host=HOST, port=PORT, reload=True)