"""Postgres connection pool + small query helpers for the yaris-obd remote server.""" import os from contextlib import contextmanager from psycopg_pool import ConnectionPool _POOL = None def _dsn(): return ( f"host={os.environ.get('POSTGRES_HOST', 'db')} " f"port={os.environ.get('POSTGRES_PORT', '5432')} " f"dbname={os.environ.get('POSTGRES_DB', 'yarisobd')} " f"user={os.environ.get('POSTGRES_USER', 'yarisobd')} " f"password={os.environ.get('POSTGRES_PASSWORD', '')}" ) def get_pool(): global _POOL if _POOL is None: _POOL = ConnectionPool(conninfo=_dsn(), min_size=1, max_size=5, open=True) return _POOL @contextmanager def get_cursor(commit=False): """Yield a cursor from the pool, optionally committing on clean exit.""" pool = get_pool() with pool.connection() as conn: with conn.cursor() as cur: yield cur if commit: conn.commit() def get_device_by_name(name): with get_cursor() as cur: cur.execute( "SELECT id, name, api_key_hash FROM devices WHERE name = %s", (name,), ) row = cur.fetchone() if row is None: return None return {"id": row[0], "name": row[1], "api_key_hash": row[2]} def touch_device_last_seen(device_id): with get_cursor(commit=True) as cur: cur.execute( "UPDATE devices SET last_seen_at = now() WHERE id = %s", (device_id,), ) def insert_tracking_points(device_id, points): """Bulk-insert tracking points, skipping any that already exist (device_id, recorded_at) so a client can safely re-send a batch after a partial-failure retry without creating duplicates.""" if not points: return 0 with get_cursor(commit=True) as cur: cur.executemany( """ INSERT INTO tracking_points (device_id, vehicle_name, driver_name, recorded_at, gps_lat, gps_lon, speed, odometer) VALUES (%(device_id)s, %(vehicle_name)s, %(driver_name)s, %(recorded_at)s, %(gps_lat)s, %(gps_lon)s, %(speed)s, %(odometer)s) ON CONFLICT (device_id, recorded_at) DO NOTHING """, [ { "device_id": device_id, "vehicle_name": p.get("vehicle_name"), "driver_name": p.get("driver_name"), "recorded_at": p["recorded_at"], "gps_lat": p.get("gps_lat"), "gps_lon": p.get("gps_lon"), "speed": p.get("speed"), "odometer": p.get("odometer"), } for p in points ], ) return len(points) def dashcam_file_exists(device_id, relative_path, sha256): with get_cursor() as cur: cur.execute( "SELECT 1 FROM dashcam_files WHERE device_id = %s AND relative_path = %s AND sha256 = %s", (device_id, relative_path, sha256), ) return cur.fetchone() is not None def insert_dashcam_file(device_id, relative_path, size_bytes, sha256, storage_path): with get_cursor(commit=True) as cur: cur.execute( """ INSERT INTO dashcam_files (device_id, relative_path, size_bytes, sha256, storage_path) VALUES (%s, %s, %s, %s, %s) ON CONFLICT (device_id, relative_path, sha256) DO NOTHING """, (device_id, relative_path, size_bytes, sha256, storage_path), ) # ── User auth ────────────────────────────────────────────────────────────── def get_user_by_id(user_id): with get_cursor() as cur: cur.execute("SELECT id, username, password_hash FROM users WHERE id = %s", (user_id,)) row = cur.fetchone() return {"id": row[0], "username": row[1], "password_hash": row[2]} if row else None def get_user_by_username(username): with get_cursor() as cur: cur.execute("SELECT id, username, password_hash FROM users WHERE username = %s", (username,)) row = cur.fetchone() return {"id": row[0], "username": row[1], "password_hash": row[2]} if row else None def create_user(username, password_hash): with get_cursor(commit=True) as cur: cur.execute( "INSERT INTO users (username, password_hash) VALUES (%s, %s) ON CONFLICT (username) DO NOTHING RETURNING id", (username, password_hash), ) row = cur.fetchone() return row[0] if row else None # ── Server-side trips ────────────────────────────────────────────────────── # Gap threshold: if two consecutive points are more than this many seconds # apart, they belong to different trips. _TRIP_GAP_S = 300 # 5 minutes def compute_trips_for_device(device_id): """Recompute trip boundaries from tracking_points and upsert into trips table.""" with get_cursor() as cur: cur.execute( """ SELECT recorded_at, gps_lat, gps_lon, speed FROM tracking_points WHERE device_id = %s AND gps_lat IS NOT NULL AND gps_lon IS NOT NULL ORDER BY recorded_at ASC """, (device_id,), ) rows = cur.fetchall() if not rows: return 0 import math def haversine(lat1, lon1, lat2, lon2): R = 6371.0 dlat = math.radians(lat2 - lat1) dlon = math.radians(lon2 - lon1) a = math.sin(dlat/2)**2 + math.cos(math.radians(lat1)) * math.cos(math.radians(lat2)) * math.sin(dlon/2)**2 return R * 2 * math.atan2(math.sqrt(a), math.sqrt(1 - a)) # Split into segments by gap segments = [] current = [rows[0]] for prev, curr in zip(rows, rows[1:]): gap = (curr[0] - prev[0]).total_seconds() if gap > _TRIP_GAP_S: segments.append(current) current = [curr] else: current.append(curr) segments.append(current) # Build trip records trips = [] for seg in segments: if len(seg) < 2: continue start_time = seg[0][0] end_time = seg[-1][0] duration_s = int((end_time - start_time).total_seconds()) if duration_s < 30: continue dist = sum( haversine(seg[i][1], seg[i][2], seg[i+1][1], seg[i+1][2]) for i in range(len(seg) - 1) ) if dist < 0.05: # skip <50m blips continue speeds = [r[3] for r in seg if r[3] is not None] trips.append({ "start_time": start_time, "end_time": end_time, "distance_km": round(dist, 3), "duration_s": duration_s, "avg_speed_kmh": round(sum(speeds) / len(speeds), 1) if speeds else None, "max_speed_kmh": round(max(speeds), 1) if speeds else None, "start_lat": seg[0][1], "start_lon": seg[0][2], "end_lat": seg[-1][1], "end_lon": seg[-1][2], "point_count": len(seg), }) with get_cursor(commit=True) as cur: for t in trips: cur.execute( """ INSERT INTO trips (device_id, start_time, end_time, distance_km, duration_s, avg_speed_kmh, max_speed_kmh, start_lat, start_lon, end_lat, end_lon, point_count, computed_at) VALUES (%(device_id)s, %(start_time)s, %(end_time)s, %(distance_km)s, %(duration_s)s, %(avg_speed_kmh)s, %(max_speed_kmh)s, %(start_lat)s, %(start_lon)s, %(end_lat)s, %(end_lon)s, %(point_count)s, now()) ON CONFLICT (device_id, start_time) DO UPDATE SET end_time = EXCLUDED.end_time, distance_km = EXCLUDED.distance_km, duration_s = EXCLUDED.duration_s, avg_speed_kmh = EXCLUDED.avg_speed_kmh, max_speed_kmh = EXCLUDED.max_speed_kmh, end_lat = EXCLUDED.end_lat, end_lon = EXCLUDED.end_lon, point_count = EXCLUDED.point_count, computed_at = now() """, {"device_id": device_id, **t}, ) return len(trips) def list_trips(device_id, limit=200): with get_cursor() as cur: cur.execute( """ SELECT id, start_time, end_time, distance_km, duration_s, avg_speed_kmh, max_speed_kmh, start_lat, start_lon, end_lat, end_lon, point_count FROM trips WHERE device_id = %s ORDER BY start_time DESC LIMIT %s """, (device_id, limit), ) rows = cur.fetchall() return [ { "id": r[0], "start_datetime": r[1].isoformat(), "end_datetime": r[2].isoformat(), "distance_km": r[3], "duration_seconds": r[4], "average_speed_kmh": r[5], "max_speed_kmh": r[6], "start_lat": r[7], "start_lon": r[8], "end_lat": r[9], "end_lon": r[10], "point_count": r[11], "odo_start_km": None, "odo_end_km": None, } for r in rows ] def get_trip_route(device_id, trip_id): with get_cursor() as cur: cur.execute( "SELECT start_time, end_time, distance_km, duration_s, avg_speed_kmh, max_speed_kmh FROM trips WHERE id = %s AND device_id = %s", (trip_id, device_id), ) trip_row = cur.fetchone() if trip_row is None: return None cur.execute( """ SELECT recorded_at, gps_lat, gps_lon, speed FROM tracking_points WHERE device_id = %s AND recorded_at BETWEEN %s AND %s AND gps_lat IS NOT NULL AND gps_lon IS NOT NULL ORDER BY recorded_at ASC """, (device_id, trip_row[0], trip_row[1]), ) points = cur.fetchall() return { "trip": { "distance_km": trip_row[2], "duration_seconds": trip_row[3], "average_speed_kmh": trip_row[4], "max_speed_kmh": trip_row[5], }, "points": [ { "datetime": p[0].isoformat(), "latitude": p[1], "longitude": p[2], "speed": p[3] or 0, "speed_source": "gps", } for p in points ], } def list_dashcam_files(device_id, limit=500): with get_cursor() as cur: cur.execute( """ SELECT id, relative_path, size_bytes, uploaded_at, storage_path FROM dashcam_files WHERE device_id = %s ORDER BY uploaded_at DESC LIMIT %s """, (device_id, limit), ) rows = cur.fetchall() return [ { "id": r[0], "relative_path": r[1], "size_bytes": r[2], "uploaded_at": r[3].isoformat(), "storage_path": r[4], } for r in rows ] def get_dashcam_file(device_id, file_id): with get_cursor() as cur: cur.execute( "SELECT id, relative_path, size_bytes, uploaded_at, storage_path FROM dashcam_files WHERE id = %s AND device_id = %s", (file_id, device_id), ) r = cur.fetchone() if r is None: return None return {"id": r[0], "relative_path": r[1], "size_bytes": r[2], "uploaded_at": r[3].isoformat(), "storage_path": r[4]} def delete_trip(device_id, trip_id): """Delete a computed trip record. Raw tracking_points are NOT removed.""" with get_cursor(commit=True) as cur: cur.execute( "DELETE FROM trips WHERE id = %s AND device_id = %s RETURNING id", (trip_id, device_id), ) return cur.fetchone() is not None def delete_dashcam_file(device_id, file_id): """Delete dashcam file from DB and disk. Returns storage_path if deleted, None if not found.""" with get_cursor(commit=True) as cur: cur.execute( "DELETE FROM dashcam_files WHERE id = %s AND device_id = %s RETURNING storage_path", (file_id, device_id), ) row = cur.fetchone() return row[0] if row else None def get_overview_stats(device_id): with get_cursor() as cur: cur.execute( """ SELECT (SELECT COUNT(*) FROM trips WHERE device_id = %s) AS trip_count, (SELECT COALESCE(SUM(distance_km), 0) FROM trips WHERE device_id = %s) AS total_km, (SELECT COALESCE(SUM(duration_s), 0) FROM trips WHERE device_id = %s) AS total_s, (SELECT COALESCE(MAX(max_speed_kmh), 0) FROM trips WHERE device_id = %s) AS top_speed, (SELECT COUNT(*) FROM dashcam_files WHERE device_id = %s) AS video_count, COALESCE( (SELECT MAX(odometer) FROM tracking_points WHERE device_id = %s AND odometer IS NOT NULL), (SELECT SUM(distance_km) FROM trips WHERE device_id = %s) ) AS odometer_km """, (device_id,) * 7, ) r = cur.fetchone() return { "trip_count": r[0], "total_distance_km": float(r[1]), "total_drive_time_s": int(r[2]), "top_speed_kmh": float(r[3]), "video_count": r[4], "odometer_km": float(r[5]) if r[5] is not None else None, }