generated from Grigo/AndroidTemplate
459 lines
15 KiB
Python
459 lines
15 KiB
Python
"""Primary LoraTester server: REST API + web UI."""
|
|
|
|
import time
|
|
from pathlib import Path
|
|
|
|
import httpx
|
|
from flask import Flask, Response, jsonify, request, send_from_directory
|
|
from flask_cors import CORS
|
|
|
|
from core.auth import ANDROID_CLIENT_HEADER, is_android_client
|
|
from core.config import HOST, PORT, RF_API_TIMEOUT, RF_API_URL
|
|
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 = Flask(__name__, static_folder=str(STATIC_DIR), static_url_path="/static")
|
|
CORS(app)
|
|
|
|
storage.init_db()
|
|
|
|
|
|
@app.route("/")
|
|
def index():
|
|
resp = send_from_directory(STATIC_DIR, "index.html")
|
|
resp.headers["Cache-Control"] = "no-store"
|
|
return resp
|
|
|
|
|
|
@app.route("/rf-api/<path:rf_path>", methods=["GET", "POST"])
|
|
@app.route("/api/rf/<path:rf_path>", methods=["GET", "POST"])
|
|
def rf_api_proxy(rf_path: str):
|
|
target = f"{RF_API_URL}/{rf_path.lstrip('/')}"
|
|
if request.query_string:
|
|
target = f"{target}?{request.query_string.decode('utf-8', errors='ignore')}"
|
|
try:
|
|
with httpx.Client(timeout=RF_API_TIMEOUT) as client:
|
|
upstream = client.request(
|
|
request.method,
|
|
target,
|
|
content=request.get_data() if request.method != "GET" else None,
|
|
headers={
|
|
"Accept": request.headers.get("Accept", "application/json"),
|
|
"Content-Type": request.headers.get(
|
|
"Content-Type", "application/json"
|
|
),
|
|
},
|
|
)
|
|
except httpx.RequestError as e:
|
|
return jsonify({
|
|
"error": "RF API unavailable",
|
|
"detail": str(e),
|
|
"target": target,
|
|
}), 502
|
|
|
|
return Response(
|
|
upstream.content,
|
|
status=upstream.status_code,
|
|
content_type=upstream.headers.get("content-type", "application/json"),
|
|
)
|
|
|
|
|
|
@app.post("/api/telemetry")
|
|
def post_telemetry():
|
|
if not is_android_client(request.headers):
|
|
return jsonify({
|
|
"error": "telemetry only from Android app",
|
|
"hint": f"send header {ANDROID_CLIENT_HEADER}: android",
|
|
}), 403
|
|
body = request.get_json(force=True, silent=True) or {}
|
|
device_id = body.get("device_id")
|
|
if not device_id:
|
|
return jsonify({"error": "device_id required"}), 400
|
|
try:
|
|
data = telemetry_from_body({**body, "device_id": device_id})
|
|
return jsonify(storage.record_telemetry(data))
|
|
except ValueError as e:
|
|
return jsonify({"error": str(e)}), 400
|
|
|
|
|
|
@app.get("/api/devices")
|
|
def get_devices():
|
|
return jsonify(storage.list_devices())
|
|
|
|
|
|
@app.get("/api/telemetry")
|
|
def get_telemetry_history():
|
|
device_id = request.args.get("device_id")
|
|
limit = int(request.args.get("limit", 100))
|
|
since = _float_or_none(request.args.get("since"))
|
|
until = _float_or_none(request.args.get("until"))
|
|
role = request.args.get("role")
|
|
return jsonify(storage.get_telemetry(device_id, limit, since, until, role))
|
|
|
|
|
|
@app.get("/api/stats/history")
|
|
def get_stats_history():
|
|
device_id = request.args.get("device_id")
|
|
if not device_id:
|
|
return jsonify({"error": "device_id required"}), 400
|
|
limit = int(request.args.get("limit", 50))
|
|
since = _float_or_none(request.args.get("since"))
|
|
until = _float_or_none(request.args.get("until"))
|
|
role = request.args.get("role")
|
|
return jsonify(storage.get_telemetry(device_id, limit, since, until, role))
|
|
|
|
|
|
@app.post("/api/tracks/start")
|
|
def tracks_start():
|
|
if not is_android_client(request.headers):
|
|
return jsonify({"error": "Android only"}), 403
|
|
body = request.get_json(force=True, silent=True) or {}
|
|
device_id = body.get("device_id")
|
|
if not device_id:
|
|
return jsonify({"error": "device_id required"}), 400
|
|
try:
|
|
return jsonify(storage.start_track(str(device_id), body.get("label")))
|
|
except ValueError as e:
|
|
return jsonify({"error": str(e)}), 400
|
|
|
|
|
|
@app.post("/api/tracks/<int:track_id>/points")
|
|
def tracks_points(track_id: int):
|
|
if not is_android_client(request.headers):
|
|
return jsonify({"error": "Android only"}), 403
|
|
body = request.get_json(force=True, silent=True) or {}
|
|
points = body.get("points") or []
|
|
try:
|
|
return jsonify(storage.add_track_points(track_id, points))
|
|
except ValueError as e:
|
|
return jsonify({"error": str(e)}), 400
|
|
|
|
|
|
@app.post("/api/tracks/<int:track_id>/finish")
|
|
def tracks_finish(track_id: int):
|
|
if not is_android_client(request.headers):
|
|
return jsonify({"error": "Android only"}), 403
|
|
try:
|
|
return jsonify(storage.finish_track(track_id))
|
|
except ValueError as e:
|
|
return jsonify({"error": str(e)}), 400
|
|
|
|
|
|
@app.get("/api/tracks")
|
|
def tracks_list():
|
|
device_id = request.args.get("device_id")
|
|
limit = int(request.args.get("limit", 50))
|
|
return jsonify(storage.list_tracks(device_id, limit))
|
|
|
|
|
|
@app.get("/api/tracks/<int:track_id>")
|
|
def tracks_get(track_id: int):
|
|
try:
|
|
return jsonify(storage.get_track(track_id))
|
|
except ValueError as e:
|
|
return jsonify({"error": str(e)}), 404
|
|
|
|
|
|
@app.post("/api/reference-points/start")
|
|
def reference_points_start():
|
|
if not is_android_client(request.headers):
|
|
return jsonify({"error": "Android only"}), 403
|
|
body = request.get_json(force=True, silent=True) or {}
|
|
device_id = body.get("device_id")
|
|
if not device_id:
|
|
return jsonify({"error": "device_id required"}), 400
|
|
try:
|
|
return jsonify(
|
|
storage.start_reference_point_session(
|
|
str(device_id),
|
|
body.get("label"),
|
|
_float_or_none(body.get("lat")),
|
|
_float_or_none(body.get("lon")),
|
|
body.get("meta") if isinstance(body.get("meta"), dict) else None,
|
|
)
|
|
)
|
|
except ValueError as e:
|
|
return jsonify({"error": str(e)}), 400
|
|
|
|
|
|
@app.post("/api/reference-points/<int:session_id>/samples")
|
|
def reference_points_samples(session_id: int):
|
|
if not is_android_client(request.headers):
|
|
return jsonify({"error": "Android only"}), 403
|
|
body = request.get_json(force=True, silent=True) or {}
|
|
samples = body.get("samples") or []
|
|
try:
|
|
return jsonify(storage.add_reference_point_samples(session_id, samples))
|
|
except ValueError as e:
|
|
return jsonify({"error": str(e)}), 400
|
|
|
|
|
|
@app.post("/api/reference-points/<int:session_id>/finish")
|
|
def reference_points_finish(session_id: int):
|
|
if not is_android_client(request.headers):
|
|
return jsonify({"error": "Android only"}), 403
|
|
try:
|
|
return jsonify(storage.finish_reference_point_session(session_id))
|
|
except ValueError as e:
|
|
return jsonify({"error": str(e)}), 400
|
|
|
|
|
|
@app.get("/api/reference-points")
|
|
def reference_points_list():
|
|
device_id = request.args.get("device_id")
|
|
limit = int(request.args.get("limit", 50))
|
|
return jsonify(storage.list_reference_point_sessions(device_id, limit))
|
|
|
|
|
|
@app.get("/api/reference-points/<int:session_id>")
|
|
def reference_points_get(session_id: int):
|
|
try:
|
|
return jsonify(storage.get_reference_point_session(session_id))
|
|
except ValueError as e:
|
|
return jsonify({"error": str(e)}), 404
|
|
|
|
|
|
@app.get("/api/macro-presets")
|
|
def macro_presets_list():
|
|
include_disabled = str(request.args.get("include_disabled", "")).lower()
|
|
return jsonify(storage.list_macro_presets(include_disabled in ("1", "true", "yes")))
|
|
|
|
|
|
@app.post("/api/macro-presets")
|
|
def macro_presets_post():
|
|
body = request.get_json(force=True, silent=True) or {}
|
|
try:
|
|
return jsonify(_upsert_macro_preset(body))
|
|
except ValueError as e:
|
|
return jsonify({"error": str(e)}), 400
|
|
|
|
|
|
@app.patch("/api/macro-presets/<preset_id>")
|
|
def macro_presets_patch(preset_id: str):
|
|
body = request.get_json(force=True, silent=True) or {}
|
|
body["id"] = preset_id
|
|
try:
|
|
return jsonify(_upsert_macro_preset(body))
|
|
except ValueError as e:
|
|
return jsonify({"error": str(e)}), 400
|
|
|
|
|
|
@app.delete("/api/macro-presets/<preset_id>")
|
|
def macro_presets_delete(preset_id: str):
|
|
try:
|
|
return jsonify(storage.delete_macro_preset(preset_id))
|
|
except ValueError as e:
|
|
return jsonify({"error": str(e)}), 404
|
|
|
|
|
|
def _upsert_macro_preset(body):
|
|
return storage.upsert_macro_preset(
|
|
body.get("id"),
|
|
body.get("name"),
|
|
body.get("description"),
|
|
body.get("params") if isinstance(body.get("params"), dict) else None,
|
|
body.get("lines") if isinstance(body.get("lines"), list) else None,
|
|
body.get("scope"),
|
|
body.get("device_role"),
|
|
bool(body.get("enabled", True)),
|
|
)
|
|
|
|
|
|
@app.post("/api/chat")
|
|
def post_chat():
|
|
body = request.get_json(force=True, silent=True) or {}
|
|
device_id = body.get("device_id")
|
|
text = (body.get("text") or "").strip()
|
|
if not device_id or not text:
|
|
return jsonify({"error": "device_id and text required"}), 400
|
|
data = ChatIn(
|
|
device_id=str(device_id),
|
|
text=text,
|
|
ts=_float_or_none(body.get("ts")),
|
|
)
|
|
return jsonify(storage.add_chat(data))
|
|
|
|
|
|
@app.get("/api/chat")
|
|
def get_chat():
|
|
since = float(request.args.get("since", 0))
|
|
limit = int(request.args.get("limit", 200))
|
|
return jsonify(storage.get_chat(since, limit))
|
|
|
|
|
|
@app.post("/api/commands")
|
|
def post_command():
|
|
body = request.get_json(force=True, silent=True) or {}
|
|
from_id = body.get("from_device_id")
|
|
to_id = body.get("to_device_id")
|
|
kind = body.get("kind")
|
|
if not from_id or not to_id or not kind:
|
|
return jsonify({"error": "from_device_id, to_device_id, kind required"}), 400
|
|
try:
|
|
return jsonify(
|
|
storage.enqueue_command(
|
|
str(from_id), str(to_id), str(kind), body.get("payload")
|
|
)
|
|
)
|
|
except ValueError as e:
|
|
return jsonify({"error": str(e)}), 400
|
|
|
|
|
|
@app.get("/api/commands/pending")
|
|
def commands_pending():
|
|
if not is_android_client(request.headers):
|
|
return jsonify({"error": "Android only"}), 403
|
|
device_id = request.args.get("device_id")
|
|
if not device_id:
|
|
return jsonify({"error": "device_id required"}), 400
|
|
limit = int(request.args.get("limit", 20))
|
|
try:
|
|
return jsonify(storage.poll_pending_commands(str(device_id), limit))
|
|
except ValueError as e:
|
|
return jsonify({"error": str(e)}), 400
|
|
|
|
|
|
@app.get("/api/commands")
|
|
def commands_list():
|
|
to_device_id = request.args.get("to_device_id")
|
|
limit = int(request.args.get("limit", 50))
|
|
return jsonify(storage.list_commands(to_device_id, limit))
|
|
|
|
|
|
@app.post("/api/paired-tracks/start")
|
|
def paired_tracks_start():
|
|
body = request.get_json(force=True, silent=True) or {}
|
|
initiator = body.get("initiator") or (
|
|
body.get("device_id") if is_android_client(request.headers) else storage.WEB_SENDER_ID
|
|
)
|
|
device_ids = body.get("device_ids")
|
|
try:
|
|
return jsonify(
|
|
storage.start_paired_track(
|
|
device_ids if isinstance(device_ids, list) else None,
|
|
str(initiator),
|
|
)
|
|
)
|
|
except ValueError as e:
|
|
return jsonify({"error": str(e)}), 400
|
|
|
|
|
|
@app.get("/api/paired-tracks/active")
|
|
def paired_tracks_active():
|
|
session = storage.get_active_paired_track()
|
|
return jsonify({"active": session is not None, "session": session})
|
|
|
|
|
|
@app.post("/api/paired-tracks/ack")
|
|
def paired_tracks_ack():
|
|
if not is_android_client(request.headers):
|
|
return jsonify({"error": "Android only"}), 403
|
|
body = request.get_json(force=True, silent=True) or {}
|
|
session_id = body.get("session_id")
|
|
device_id = body.get("device_id")
|
|
track_id = body.get("track_id")
|
|
if session_id is None or not device_id or track_id is None:
|
|
return jsonify({"error": "session_id, device_id, track_id required"}), 400
|
|
try:
|
|
return jsonify(
|
|
storage.ack_paired_track(int(session_id), str(device_id), int(track_id))
|
|
)
|
|
except ValueError as e:
|
|
return jsonify({"error": str(e)}), 400
|
|
|
|
|
|
@app.post("/api/paired-tracks/cancel")
|
|
def paired_tracks_cancel():
|
|
body = request.get_json(force=True, silent=True) or {}
|
|
session_id = body.get("session_id")
|
|
try:
|
|
sid = int(session_id) if session_id is not None else None
|
|
return jsonify(storage.cancel_paired_track(sid))
|
|
except ValueError as e:
|
|
return jsonify({"error": str(e)}), 400
|
|
|
|
|
|
@app.post("/api/elevation/profile")
|
|
def elevation_profile():
|
|
from core.elevation import build_elevation_profile
|
|
|
|
body = request.get_json(force=True, silent=True) or {}
|
|
points = body.get("points") or []
|
|
step_m = body.get("step_m", 10)
|
|
try:
|
|
step = float(step_m)
|
|
except (TypeError, ValueError):
|
|
step = 10.0
|
|
target_points = body.get("target_points")
|
|
try:
|
|
tp = int(target_points) if target_points is not None else None
|
|
except (TypeError, ValueError):
|
|
tp = None
|
|
return jsonify(build_elevation_profile(points, step, tp))
|
|
|
|
|
|
@app.get("/api/tracks/<int:track_id>/elevation-profile")
|
|
def track_elevation_profile(track_id: int):
|
|
from core.elevation import build_elevation_profile
|
|
|
|
step_m = request.args.get("step_m", 10, type=float)
|
|
try:
|
|
track = storage.get_track(track_id)
|
|
except ValueError as e:
|
|
return jsonify({"error": str(e)}), 404
|
|
points = track.get("points") or []
|
|
return jsonify(build_elevation_profile(points, step_m or 10.0))
|
|
|
|
|
|
@app.get("/api/elevation/nearest-hill")
|
|
def elevation_nearest_hill():
|
|
from core.elevation import find_nearest_hill
|
|
|
|
lat = request.args.get("lat", type=float)
|
|
lon = request.args.get("lon", type=float)
|
|
if lat is None or lon is None:
|
|
return jsonify({"ok": False, "error": "lat and lon required"}), 400
|
|
radius_m = request.args.get("radius_m", 5000, type=float)
|
|
step_m = request.args.get("step_m", 300, type=float)
|
|
min_prominence_m = request.args.get("min_prominence_m", 8, type=float)
|
|
return jsonify(find_nearest_hill(lat, lon, radius_m, step_m, min_prominence_m))
|
|
|
|
|
|
@app.get("/api/elevation/grid")
|
|
def elevation_grid():
|
|
from core.elevation import build_elevation_grid
|
|
|
|
lat = request.args.get("lat", type=float)
|
|
lon = request.args.get("lon", type=float)
|
|
if lat is None or lon is None:
|
|
return jsonify({"ok": False, "error": "lat and lon required"}), 400
|
|
radius_m = request.args.get("radius_m", 200, type=float)
|
|
step_m = request.args.get("step_m", 0, type=float)
|
|
return jsonify(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 jsonify(
|
|
{"ok": status["db_ok"], "ts": time.time(), **status, **elevation_status()}
|
|
)
|
|
|
|
|
|
def _float_or_none(value):
|
|
if value is None or value == "":
|
|
return None
|
|
try:
|
|
return float(value)
|
|
except (TypeError, ValueError):
|
|
return None
|
|
|
|
|
|
if __name__ == "__main__":
|
|
app.run(host=HOST, port=PORT, debug=True)
|