generated from Grigo/AndroidTemplate
377 lines
10 KiB
Python
377 lines
10 KiB
Python
"""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)
|
|
|
|
|
|
class CommandBody(BaseModel):
|
|
from_device_id: str
|
|
to_device_id: str
|
|
kind: str
|
|
payload: Optional[dict[str, Any]] = None
|
|
|
|
|
|
class PairedTrackStartBody(BaseModel):
|
|
device_ids: Optional[list[str]] = None
|
|
initiator: Optional[str] = None
|
|
device_id: Optional[str] = None
|
|
|
|
|
|
class PairedTrackAckBody(BaseModel):
|
|
session_id: int
|
|
device_id: str
|
|
track_id: int
|
|
|
|
|
|
class PairedTrackCancelBody(BaseModel):
|
|
session_id: Optional[int] = None
|
|
|
|
|
|
@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.post("/api/commands")
|
|
def post_command(body: CommandBody):
|
|
try:
|
|
return storage.enqueue_command(
|
|
body.from_device_id,
|
|
body.to_device_id,
|
|
body.kind,
|
|
body.payload,
|
|
)
|
|
except ValueError as e:
|
|
raise HTTPException(400, detail=str(e)) from e
|
|
|
|
|
|
@app.get("/api/commands/pending")
|
|
def commands_pending(
|
|
device_id: str = Query(...),
|
|
limit: int = Query(20, ge=1, le=50),
|
|
x_lora_client: Optional[str] = Header(None, alias=ANDROID_CLIENT_HEADER),
|
|
):
|
|
_require_android(x_lora_client)
|
|
try:
|
|
return storage.poll_pending_commands(device_id, limit)
|
|
except ValueError as e:
|
|
raise HTTPException(400, detail=str(e)) from e
|
|
|
|
|
|
@app.get("/api/commands")
|
|
def commands_list(
|
|
to_device_id: Optional[str] = None,
|
|
limit: int = Query(50, ge=1, le=200),
|
|
):
|
|
return storage.list_commands(to_device_id, limit)
|
|
|
|
|
|
@app.post("/api/paired-tracks/start")
|
|
def paired_tracks_start(
|
|
body: PairedTrackStartBody,
|
|
x_lora_client: Optional[str] = Header(None, alias=ANDROID_CLIENT_HEADER),
|
|
):
|
|
if body.initiator:
|
|
initiator = body.initiator
|
|
elif body.device_id:
|
|
initiator = body.device_id
|
|
elif (x_lora_client or "").strip().lower() == ANDROID_CLIENT_VALUE:
|
|
raise HTTPException(400, detail="initiator or device_id required")
|
|
else:
|
|
initiator = storage.WEB_SENDER_ID
|
|
try:
|
|
return storage.start_paired_track(body.device_ids, str(initiator))
|
|
except ValueError as e:
|
|
raise HTTPException(400, detail=str(e)) from e
|
|
|
|
|
|
@app.get("/api/paired-tracks/active")
|
|
def paired_tracks_active():
|
|
session = storage.get_active_paired_track()
|
|
return {"active": session is not None, "session": session}
|
|
|
|
|
|
@app.post("/api/paired-tracks/ack")
|
|
def paired_tracks_ack(
|
|
body: PairedTrackAckBody,
|
|
x_lora_client: Optional[str] = Header(None, alias=ANDROID_CLIENT_HEADER),
|
|
):
|
|
_require_android(x_lora_client)
|
|
try:
|
|
return storage.ack_paired_track(
|
|
body.session_id, body.device_id, body.track_id
|
|
)
|
|
except ValueError as e:
|
|
raise HTTPException(400, detail=str(e)) from e
|
|
|
|
|
|
@app.post("/api/paired-tracks/cancel")
|
|
def paired_tracks_cancel(body: PairedTrackCancelBody):
|
|
try:
|
|
return storage.cancel_paired_track(body.session_id)
|
|
except ValueError as e:
|
|
raise HTTPException(400, detail=str(e)) from e
|
|
|
|
|
|
class ElevationPoint(BaseModel):
|
|
lat: float
|
|
lon: float
|
|
ts: Optional[float] = None
|
|
|
|
|
|
class ElevationProfileBody(BaseModel):
|
|
points: list[ElevationPoint] = Field(default_factory=list)
|
|
step_m: float = 10.0
|
|
target_points: Optional[int] = Field(None, ge=2, le=500)
|
|
|
|
|
|
@app.post("/api/elevation/profile")
|
|
def elevation_profile(body: ElevationProfileBody):
|
|
from core.elevation import build_elevation_profile
|
|
|
|
pts = [p.model_dump(exclude_none=True) for p in body.points]
|
|
return build_elevation_profile(pts, body.step_m, body.target_points)
|
|
|
|
|
|
@app.get("/api/tracks/{track_id}/elevation-profile")
|
|
def track_elevation_profile(
|
|
track_id: int,
|
|
step_m: float = Query(10.0, ge=5.0, le=10.0),
|
|
):
|
|
from core.elevation import build_elevation_profile
|
|
|
|
try:
|
|
track = storage.get_track(track_id)
|
|
except ValueError as e:
|
|
raise HTTPException(404, detail=str(e)) from e
|
|
return build_elevation_profile(track.get("points") or [], step_m)
|
|
|
|
|
|
@app.get("/api/elevation/nearest-hill")
|
|
def elevation_nearest_hill(
|
|
lat: float = Query(..., ge=-90.0, le=90.0),
|
|
lon: float = Query(..., ge=-180.0, le=180.0),
|
|
radius_m: float = Query(5000.0, ge=500.0, le=15000.0),
|
|
step_m: float = Query(300.0, ge=100.0, le=500.0),
|
|
min_prominence_m: float = Query(8.0, ge=3.0, le=100.0),
|
|
):
|
|
from core.elevation import find_nearest_hill
|
|
|
|
return find_nearest_hill(lat, lon, radius_m, step_m, min_prominence_m)
|
|
|
|
|
|
@app.get("/api/elevation/grid")
|
|
def elevation_grid(
|
|
lat: float = Query(..., ge=-90.0, le=90.0),
|
|
lon: float = Query(..., ge=-180.0, le=180.0),
|
|
radius_m: float = Query(200.0, ge=50.0, le=500.0),
|
|
step_m: float = Query(0.0, ge=1.0, le=100.0),
|
|
):
|
|
from core.elevation import build_elevation_grid
|
|
|
|
return build_elevation_grid(lat, lon, radius_m, step_m)
|
|
|
|
|
|
@app.get("/api/health")
|
|
def health():
|
|
from core.elevation import elevation_status
|
|
|
|
status = storage.db_status()
|
|
return {
|
|
"ok": status["db_ok"],
|
|
"ts": time.time(),
|
|
**status,
|
|
**elevation_status(),
|
|
}
|
|
|
|
|
|
if __name__ == "__main__":
|
|
import uvicorn
|
|
|
|
uvicorn.run("fastapi_app:app", host=HOST, port=PORT, reload=True)
|