kayos
37aefb84c8
Clean Python Flask replacement for odc-api (434k lines Node.js → ~350 lines Python)
- GET /api/1/landmarks/last/{N} - last N detections from SQLite
- POST /api/1/landmarks - ingest detections + forward to AdaMaps
- GET /api/1/gnssConcise/latestValid - GPS fix from Redis
- GET /api/1/status - device status
- GET /api/1/deviceinfo - device identity
- GET /api/1/recording/frames/latest - latest frame path
No /api/1/cmd - that's the CVE, it's gone.
Includes:
- SQLite for local storage + offline queue
- Background thread for AdaMaps retry
- systemd service unit
- install.sh for device deployment
40 lines
1.2 KiB
Python
40 lines
1.2 KiB
Python
"""Redis client for GPS and IMU data."""
|
|
import json
|
|
import redis
|
|
|
|
_client = None
|
|
|
|
|
|
def get_client():
|
|
"""Get Redis connection."""
|
|
global _client
|
|
if _client is None:
|
|
_client = redis.Redis(host="localhost", port=6379, decode_responses=True)
|
|
return _client
|
|
|
|
|
|
def get_latest_gnss():
|
|
"""Get latest GPS fix from GNSSFusion30Hz ZSET."""
|
|
client = get_client()
|
|
try:
|
|
# Get the most recent entry (highest score = most recent timestamp)
|
|
results = client.zrevrange("GNSSFusion30Hz", 0, 0, withscores=True)
|
|
if results:
|
|
data = json.loads(results[0][0])
|
|
return {
|
|
"lat_deg": data.get("lat_deg"),
|
|
"lon_deg": data.get("lon_deg"),
|
|
"alt_m": data.get("alt_m"),
|
|
"unix_milliseconds": int(data.get("unix_milliseconds", 0)),
|
|
"hdop": data.get("hdop"),
|
|
"num_satellites": data.get("num_satellites", 0),
|
|
}
|
|
except (redis.RedisError, json.JSONDecodeError):
|
|
pass
|
|
return None
|
|
|
|
|
|
def has_gps_lock():
|
|
"""Check if we have a valid GPS lock."""
|
|
gnss = get_latest_gnss()
|
|
return gnss is not None and gnss.get("num_satellites", 0) >= 4
|