458 lines
15 KiB
Python
458 lines
15 KiB
Python
"""
|
|
FR24 API Service
|
|
Minimal Flask API reading from PostgreSQL fr24 schema.
|
|
"""
|
|
import os
|
|
import time
|
|
import logging
|
|
from datetime import datetime, timezone
|
|
from functools import wraps
|
|
|
|
import psycopg2
|
|
import psycopg2.extras
|
|
from flask import Flask, jsonify, request, send_from_directory
|
|
|
|
logging.basicConfig(
|
|
level=logging.INFO,
|
|
format="%(asctime)s [api] %(levelname)s %(message)s",
|
|
datefmt="%Y-%m-%dT%H:%M:%S",
|
|
)
|
|
log = logging.getLogger("api")
|
|
|
|
app = Flask(__name__, static_folder="/app/static", static_url_path="/static")
|
|
|
|
DB_DSN = (
|
|
f"host={os.environ['POSTGRES_HOST']} "
|
|
f"port={os.environ.get('POSTGRES_PORT', 5432)} "
|
|
f"dbname={os.environ['POSTGRES_DB']} "
|
|
f"user={os.environ['POSTGRES_USER']} "
|
|
f"password={os.environ['POSTGRES_PASSWORD']}"
|
|
)
|
|
API_PORT = int(os.environ.get("API_PORT", 8080))
|
|
HEALTHCHECK_FILE = "/tmp/api-ready"
|
|
START_TIME = datetime.now(timezone.utc)
|
|
|
|
# ── db connection (simple persistent conn with reconnect) ─────────────────────
|
|
|
|
_conn = None
|
|
|
|
def get_conn():
|
|
global _conn
|
|
if _conn is None or _conn.closed:
|
|
_conn = psycopg2.connect(DB_DSN)
|
|
psycopg2.extras.register_uuid(_conn)
|
|
log.info("DB connection established")
|
|
return _conn
|
|
|
|
|
|
def query(sql: str, params=None) -> list:
|
|
for attempt in range(2):
|
|
try:
|
|
conn = get_conn()
|
|
with conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur:
|
|
cur.execute(sql, params)
|
|
return [dict(r) for r in cur.fetchall()]
|
|
except psycopg2.OperationalError:
|
|
global _conn
|
|
_conn = None
|
|
if attempt == 1:
|
|
raise
|
|
|
|
|
|
def query_one(sql: str, params=None) -> dict | None:
|
|
rows = query(sql, params)
|
|
return rows[0] if rows else None
|
|
|
|
|
|
# ── serialisation helper ──────────────────────────────────────────────────────
|
|
|
|
def serial(obj):
|
|
"""Make psycopg2 types JSON-serialisable."""
|
|
import decimal, uuid
|
|
if isinstance(obj, (datetime,)):
|
|
return obj.isoformat()
|
|
if isinstance(obj, decimal.Decimal):
|
|
return float(obj)
|
|
if isinstance(obj, uuid.UUID):
|
|
return str(obj)
|
|
raise TypeError(f"Not serialisable: {type(obj)}")
|
|
|
|
|
|
def ok(data, **kwargs):
|
|
return app.response_class(
|
|
__import__("json").dumps(data, default=serial),
|
|
mimetype="application/json",
|
|
**kwargs,
|
|
)
|
|
|
|
|
|
def err(msg: str, status: int = 500):
|
|
return ok({"error": msg}, status=status)
|
|
|
|
|
|
# ── routes ────────────────────────────────────────────────────────────────────
|
|
|
|
@app.get("/")
|
|
def index():
|
|
return send_from_directory("/app/static", "index.html")
|
|
|
|
|
|
@app.get("/health")
|
|
def health():
|
|
try:
|
|
query_one("SELECT 1")
|
|
db_ok = True
|
|
except Exception:
|
|
db_ok = False
|
|
return ok({
|
|
"status": "ok" if db_ok else "degraded",
|
|
"db": "ok" if db_ok else "error",
|
|
"uptime_seconds": int((datetime.now(timezone.utc) - START_TIME).total_seconds()),
|
|
}, status=200 if db_ok else 503)
|
|
|
|
|
|
@app.get("/dashboard/status")
|
|
def dashboard_status():
|
|
try:
|
|
captures = query_one("SELECT COUNT(*) AS total, COUNT(*) FILTER (WHERE status='active') AS active FROM fr24.captures")
|
|
packets = query_one("SELECT COUNT(*) AS total FROM fr24.raw_packets")
|
|
state = query_one("SELECT state_value FROM fr24.processing_state WHERE state_key='preprocess_cursor'")
|
|
aircraft = query_one("SELECT COUNT(*) AS total FROM fr24.aircraft")
|
|
flights = query_one("SELECT COUNT(*) AS total, COUNT(*) FILTER (WHERE status='active') AS active FROM fr24.flights")
|
|
return ok({
|
|
"captures": captures,
|
|
"raw_packets": packets,
|
|
"processing_state": state["state_value"] if state else None,
|
|
"aircraft": aircraft,
|
|
"flights": flights,
|
|
})
|
|
except Exception as e:
|
|
return err(str(e))
|
|
|
|
|
|
@app.get("/viewer/config")
|
|
def viewer_config():
|
|
return ok({
|
|
"system": "fr24-ingest",
|
|
"version": "0.1.0",
|
|
"stage": "step-3-real-tracks",
|
|
"db_schema": "fr24",
|
|
"center": {"lat": 55.75, "lon": 37.62},
|
|
"zoom": 7,
|
|
"features": {
|
|
"adsb_decode": True,
|
|
"real_rtlsdr": True,
|
|
"noise_model": False,
|
|
},
|
|
})
|
|
|
|
|
|
@app.get("/captures")
|
|
def captures():
|
|
try:
|
|
limit = min(int(request.args.get("limit", 50)), 200)
|
|
rows = query(
|
|
"""
|
|
SELECT capture_id, started_at, ended_at, source, device_index,
|
|
center_frequency_hz, sample_rate_hz, gain_db, status, notes, created_at
|
|
FROM fr24.captures
|
|
ORDER BY started_at DESC
|
|
LIMIT %s
|
|
""",
|
|
(limit,),
|
|
)
|
|
return ok({"captures": rows, "count": len(rows)})
|
|
except Exception as e:
|
|
return err(str(e))
|
|
|
|
|
|
@app.get("/aircraft")
|
|
def aircraft_list():
|
|
try:
|
|
limit = min(int(request.args.get("limit", 100)), 500)
|
|
rows = query(
|
|
"""
|
|
SELECT aircraft_id, icao24, callsign, registration, aircraft_type,
|
|
operator_name, first_seen_at, last_seen_at
|
|
FROM fr24.aircraft
|
|
ORDER BY last_seen_at DESC NULLS LAST
|
|
LIMIT %s
|
|
""",
|
|
(limit,),
|
|
)
|
|
return ok({"aircraft": rows, "count": len(rows)})
|
|
except Exception as e:
|
|
return err(str(e))
|
|
|
|
|
|
@app.get("/flights")
|
|
def flights():
|
|
try:
|
|
limit = min(int(request.args.get("limit", 100)), 500)
|
|
status_filter = request.args.get("status")
|
|
if status_filter:
|
|
rows = query(
|
|
"""
|
|
SELECT f.flight_id, f.aircraft_id, a.icao24, f.callsign,
|
|
f.departure_airport, f.arrival_airport,
|
|
f.started_at, f.ended_at, f.status, f.source
|
|
FROM fr24.flights f
|
|
JOIN fr24.aircraft a USING (aircraft_id)
|
|
WHERE f.status = %s
|
|
ORDER BY f.started_at DESC
|
|
LIMIT %s
|
|
""",
|
|
(status_filter, limit),
|
|
)
|
|
else:
|
|
rows = query(
|
|
"""
|
|
SELECT f.flight_id, f.aircraft_id, a.icao24, f.callsign,
|
|
f.departure_airport, f.arrival_airport,
|
|
f.started_at, f.ended_at, f.status, f.source
|
|
FROM fr24.flights f
|
|
JOIN fr24.aircraft a USING (aircraft_id)
|
|
ORDER BY f.started_at DESC
|
|
LIMIT %s
|
|
""",
|
|
(limit,),
|
|
)
|
|
return ok({"flights": rows, "count": len(rows)})
|
|
except Exception as e:
|
|
return err(str(e))
|
|
|
|
|
|
# ── map / live endpoints ──────────────────────────────────────────────────────
|
|
|
|
@app.get("/api/aircraft/live")
|
|
def aircraft_live():
|
|
"""Active aircraft with their latest position as GeoJSON FeatureCollection."""
|
|
try:
|
|
minutes = int(request.args.get("minutes", 60))
|
|
rows = query(
|
|
"""
|
|
SELECT
|
|
a.icao24,
|
|
a.callsign,
|
|
a.registration,
|
|
a.aircraft_type,
|
|
tp.observed_at,
|
|
ST_X(tp.geom) AS lon,
|
|
ST_Y(tp.geom) AS lat,
|
|
tp.altitude_m,
|
|
tp.ground_speed_kt,
|
|
tp.heading_deg,
|
|
tp.vertical_rate_fpm
|
|
FROM fr24.aircraft a
|
|
JOIN fr24.flights f ON f.aircraft_id = a.aircraft_id
|
|
JOIN LATERAL (
|
|
SELECT tp2.geom, tp2.observed_at, tp2.altitude_m,
|
|
tp2.ground_speed_kt, tp2.heading_deg, tp2.vertical_rate_fpm
|
|
FROM fr24.track_points tp2
|
|
WHERE tp2.flight_id = f.flight_id
|
|
ORDER BY tp2.observed_at DESC
|
|
LIMIT 1
|
|
) tp ON true
|
|
WHERE f.status = 'active'
|
|
AND tp.observed_at >= now() - (%s || ' minutes')::interval
|
|
ORDER BY tp.observed_at DESC
|
|
""",
|
|
(minutes,),
|
|
)
|
|
features = []
|
|
for r in rows:
|
|
if r["lon"] is None or r["lat"] is None:
|
|
continue
|
|
features.append({
|
|
"type": "Feature",
|
|
"geometry": {"type": "Point", "coordinates": [r["lon"], r["lat"]]},
|
|
"properties": {
|
|
"icao24": r["icao24"],
|
|
"callsign": r["callsign"] or r["icao24"],
|
|
"registration": r["registration"],
|
|
"aircraft_type": r["aircraft_type"],
|
|
"altitude_m": r["altitude_m"],
|
|
"ground_speed_kt": r["ground_speed_kt"],
|
|
"heading_deg": r["heading_deg"],
|
|
"vertical_rate_fpm": r["vertical_rate_fpm"],
|
|
"observed_at": r["observed_at"].isoformat() if r["observed_at"] else None,
|
|
},
|
|
})
|
|
return ok({"type": "FeatureCollection", "features": features, "count": len(features)})
|
|
except Exception as e:
|
|
return err(str(e))
|
|
|
|
|
|
@app.get("/api/aircraft/<icao24>")
|
|
def aircraft_detail(icao24: str):
|
|
"""Details for a single aircraft including recent track points."""
|
|
try:
|
|
ac = query_one(
|
|
"SELECT * FROM fr24.aircraft WHERE icao24 = %s",
|
|
(icao24.lower(),),
|
|
)
|
|
if not ac:
|
|
return err("not found", 404)
|
|
|
|
flights_rows = query(
|
|
"""
|
|
SELECT flight_id, callsign, departure_airport, arrival_airport,
|
|
started_at, ended_at, status
|
|
FROM fr24.flights
|
|
WHERE aircraft_id = %s
|
|
ORDER BY started_at DESC
|
|
LIMIT 10
|
|
""",
|
|
(ac["aircraft_id"],),
|
|
)
|
|
|
|
# last 100 track points across all recent flights
|
|
points = query(
|
|
"""
|
|
SELECT tp.observed_at,
|
|
ST_X(tp.geom) AS lon, ST_Y(tp.geom) AS lat,
|
|
tp.altitude_m, tp.ground_speed_kt, tp.heading_deg
|
|
FROM fr24.track_points tp
|
|
JOIN fr24.flights f ON f.flight_id = tp.flight_id
|
|
WHERE f.aircraft_id = %s
|
|
ORDER BY tp.observed_at DESC
|
|
LIMIT 100
|
|
""",
|
|
(ac["aircraft_id"],),
|
|
)
|
|
|
|
return ok({
|
|
"aircraft": ac,
|
|
"flights": flights_rows,
|
|
"recent_points": points,
|
|
})
|
|
except Exception as e:
|
|
return err(str(e))
|
|
|
|
|
|
@app.get("/monitoring")
|
|
def monitoring_page():
|
|
return send_from_directory("/app/static", "monitoring.html")
|
|
|
|
|
|
@app.get("/api/monitoring/status")
|
|
def monitoring_status():
|
|
"""Return latest monitoring metrics + last 20 rows history."""
|
|
try:
|
|
latest = query_one(
|
|
"""
|
|
SELECT id, collected_at, disk_pct, db_size_mb, capture_lag_sec, throughput_5min
|
|
FROM fr24.monitoring_metrics
|
|
ORDER BY id DESC
|
|
LIMIT 1
|
|
"""
|
|
)
|
|
history = query(
|
|
"""
|
|
SELECT id, collected_at, disk_pct, db_size_mb, capture_lag_sec, throughput_5min
|
|
FROM fr24.monitoring_metrics
|
|
ORDER BY id DESC
|
|
LIMIT 20
|
|
"""
|
|
)
|
|
return ok({"latest": latest, "history": history})
|
|
except Exception as e:
|
|
return err(str(e))
|
|
|
|
|
|
@app.get("/api/tracks")
|
|
def tracks():
|
|
"""Track points as GeoJSON LineStrings, filtered by time and optional bbox."""
|
|
try:
|
|
minutes = int(request.args.get("minutes", 30))
|
|
limit = min(int(request.args.get("limit", 50)), 200)
|
|
|
|
# optional bbox: ?bbox=minlon,minlat,maxlon,maxlat
|
|
bbox = request.args.get("bbox")
|
|
bbox_clause = ""
|
|
bbox_params: list = []
|
|
if bbox:
|
|
try:
|
|
minlon, minlat, maxlon, maxlat = map(float, bbox.split(","))
|
|
bbox_clause = (
|
|
"AND tp.geom && ST_MakeEnvelope(%s, %s, %s, %s, 4326)"
|
|
)
|
|
bbox_params = [minlon, minlat, maxlon, maxlat]
|
|
except ValueError:
|
|
pass
|
|
|
|
rows = query(
|
|
f"""
|
|
SELECT
|
|
t.track_id,
|
|
f.flight_id,
|
|
a.icao24,
|
|
f.callsign,
|
|
json_agg(
|
|
json_build_object(
|
|
'lon', ST_X(tp.geom),
|
|
'lat', ST_Y(tp.geom),
|
|
'alt', tp.altitude_m,
|
|
'spd', tp.ground_speed_kt,
|
|
'hdg', tp.heading_deg,
|
|
'ts', tp.observed_at
|
|
)
|
|
ORDER BY tp.point_order
|
|
) AS points
|
|
FROM fr24.tracks t
|
|
JOIN fr24.flights f ON f.flight_id = t.flight_id
|
|
JOIN fr24.aircraft a ON a.aircraft_id = f.aircraft_id
|
|
JOIN fr24.track_points tp ON tp.track_id = t.track_id
|
|
WHERE tp.observed_at >= now() - (%s || ' minutes')::interval
|
|
{bbox_clause}
|
|
GROUP BY t.track_id, f.flight_id, a.icao24, f.callsign
|
|
ORDER BY t.last_point_at DESC NULLS LAST
|
|
LIMIT %s
|
|
""",
|
|
(minutes, *bbox_params, limit),
|
|
)
|
|
|
|
features = []
|
|
for r in rows:
|
|
pts = r["points"] if isinstance(r["points"], list) else []
|
|
coords = [[p["lon"], p["lat"]] for p in pts if p.get("lon") is not None]
|
|
if len(coords) < 2:
|
|
continue
|
|
features.append({
|
|
"type": "Feature",
|
|
"geometry": {"type": "LineString", "coordinates": coords},
|
|
"properties": {
|
|
"track_id": r["track_id"],
|
|
"flight_id": r["flight_id"],
|
|
"icao24": r["icao24"],
|
|
"callsign": r["callsign"] or r["icao24"],
|
|
"point_count": len(coords),
|
|
},
|
|
})
|
|
|
|
return ok({"type": "FeatureCollection", "features": features, "count": len(features)})
|
|
except Exception as e:
|
|
return err(str(e))
|
|
|
|
|
|
# ── startup ───────────────────────────────────────────────────────────────────
|
|
|
|
def wait_for_db(max_attempts: int = 30):
|
|
for attempt in range(1, max_attempts + 1):
|
|
try:
|
|
get_conn()
|
|
return
|
|
except psycopg2.OperationalError as e:
|
|
log.warning("DB not ready (%d/%d): %s", attempt, max_attempts, e)
|
|
time.sleep(2)
|
|
log.error("Could not connect to DB")
|
|
raise SystemExit(1)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
wait_for_db()
|
|
open(HEALTHCHECK_FILE, "w").close()
|
|
log.info("Healthcheck file written: %s", HEALTHCHECK_FILE)
|
|
log.info("Starting API on port %d", API_PORT)
|
|
app.run(host="0.0.0.0", port=API_PORT, debug=False)
|