"""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 get_device_by_id(device_id): """Get device by ID, including config + sync metadata.""" with get_cursor() as cur: cur.execute( """ SELECT id, name, api_key_hash, owner_name, car_model, device_name, avatar_path, odometer_baseline, config_version, config_updated_at, created_at, last_seen_at, last_config_sync_at, last_config_sync_version, force_sync_requested_at, api_key_rotated_at FROM devices WHERE id = %s """, (device_id,), ) row = cur.fetchone() if row is None: return None return { "id": row[0], "name": row[1], "api_key_hash": row[2], "owner_name": row[3], "car_model": row[4], "device_name": row[5], "avatar_path": row[6], "odometer_baseline": float(row[7]) if row[7] is not None else None, "config_version": row[8], "config_updated_at": row[9], "created_at": row[10], "last_seen_at": row[11], "last_config_sync_at": row[12], "last_config_sync_version": row[13], "force_sync_requested_at": row[14], "api_key_rotated_at": row[15], } 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 list_devices(): with get_cursor() as cur: cur.execute( """ SELECT id, name, owner_name, car_model, device_name, avatar_path, config_version, config_updated_at, created_at, last_seen_at, last_config_sync_at, last_config_sync_version, force_sync_requested_at, api_key_rotated_at FROM devices ORDER BY created_at ASC, id ASC """ ) rows = cur.fetchall() return [ { "id": row[0], "name": row[1], "owner_name": row[2], "car_model": row[3], "device_name": row[4], "avatar_path": row[5], "config_version": row[6], "config_updated_at": row[7], "created_at": row[8], "last_seen_at": row[9], "last_config_sync_at": row[10], "last_config_sync_version": row[11], "force_sync_requested_at": row[12], "api_key_rotated_at": row[13], } for row in rows ] def create_device(name, api_key_hash): with get_cursor(commit=True) as cur: cur.execute( """ INSERT INTO devices (name, api_key_hash, api_key_rotated_at) VALUES (%s, %s, now()) RETURNING id """, (name, api_key_hash), ) row = cur.fetchone() return row[0] def rotate_device_api_key(device_id, api_key_hash): with get_cursor(commit=True) as cur: cur.execute( """ UPDATE devices SET api_key_hash = %s, api_key_rotated_at = now() WHERE id = %s RETURNING id """, (api_key_hash, device_id), ) row = cur.fetchone() return row[0] if row else None def request_device_sync(device_id): with get_cursor(commit=True) as cur: cur.execute( """ UPDATE devices SET config_version = config_version + 1, config_updated_at = now(), force_sync_requested_at = now() WHERE id = %s RETURNING config_version """, (device_id,), ) row = cur.fetchone() return row[0] if row else None def acknowledge_device_sync(device_id, config_version): with get_cursor(commit=True) as cur: cur.execute( """ UPDATE devices SET last_config_sync_at = now(), last_config_sync_version = %s, force_sync_requested_at = CASE WHEN config_version <= %s THEN NULL ELSE force_sync_requested_at END WHERE id = %s """, (config_version, config_version, device_id), ) def delete_device(device_id): with get_cursor(commit=True) as cur: cur.execute("DELETE FROM devices WHERE id = %s RETURNING id", (device_id,)) row = cur.fetchone() return row[0] if row else None 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), ) # ── DTC events (FEAT-05) ──────────────────────────────────────────────────── def upsert_dtc_snapshot(device_id, mil_on, codes): """Receive a DTC snapshot from the Pi. Inserts new codes, updates existing, marks codes absent from this snapshot as cleared. Returns stats dict.""" with get_cursor(commit=True) as cur: cur.execute( "SELECT code FROM dtc_events WHERE device_id = %s AND is_active = TRUE", (device_id,), ) previously_active = {row[0] for row in cur.fetchall()} current_codes = {c["code"] for c in codes} new_codes = current_codes - previously_active cleared_codes = previously_active - current_codes unchanged = previously_active & current_codes with get_cursor(commit=True) as cur: for c in codes: if c["code"] in new_codes: cur.execute( """INSERT INTO dtc_events (device_id, code, description, mil_on, is_active) VALUES (%s, %s, %s, %s, TRUE) ON CONFLICT (device_id, code) DO UPDATE SET description = EXCLUDED.description, mil_on = EXCLUDED.mil_on, is_active = TRUE, last_seen_at = now()""", (device_id, c["code"], c.get("description", ""), mil_on), ) elif c["code"] in unchanged: cur.execute( """UPDATE dtc_events SET mil_on = %s, last_seen_at = now(), is_active = TRUE WHERE device_id = %s AND code = %s""", (mil_on, device_id, c["code"]), ) for code in cleared_codes: cur.execute( """UPDATE dtc_events SET is_active = FALSE, cleared_at = now() WHERE device_id = %s AND code = %s""", (device_id, code), ) return {"received": len(codes), "cleared": len(cleared_codes), "unchanged": len(unchanged)} def list_dtc_events(device_id): with get_cursor() as cur: cur.execute( """SELECT id, code, description, mil_on, is_active, first_seen_at, last_seen_at, cleared_at FROM dtc_events WHERE device_id = %s ORDER BY last_seen_at DESC""", (device_id,), ) rows = cur.fetchall() return [ { "id": r[0], "code": r[1], "description": r[2], "mil_on": r[3], "is_active": r[4], "first_seen_at": r[5].isoformat() if r[5] else None, "last_seen_at": r[6].isoformat() if r[6] else None, "cleared_at": r[7].isoformat() if r[7] else None, } for r in rows ] def delete_dtc_event(device_id, dtc_id): with get_cursor(commit=True) as cur: cur.execute( "DELETE FROM dtc_events WHERE id = %s AND device_id = %s RETURNING id", (dtc_id, device_id), ) return cur.fetchone() is not None # ── 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. Mirrors the Pi's motion-aware trip-detection logic (modules/app_trip.py): OBD wheel speed > 1 km/h is the primary motion signal. GPS displacement >= 8 m per step (3 consecutive) is the fallback. A 5-minute gap between motion events starts a new trip. This prevents parked idle rows from merging two separate drives into one. """ INACTIVITY_SECONDS = 5 * 60 GPS_MOTION_METERS = 8 MIN_GPS_MOTION_SAMPLES = 3 MAX_GPS_SAMPLE_GAP_SECONDS = 30 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 len(rows) < 2: return 0 import math def haversine_km(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)) # Detect motion events per step obd_motion = [] # indices where OBD speed > 1 km/h gps_motion = [] # indices where GPS displacement >= 8 m for i in range(1, len(rows)): prev = rows[i - 1] curr = rows[i] gap = (curr[0] - prev[0]).total_seconds() speed = curr[3] or 0 has_obd = speed > 1 has_gps = False if 0 < gap <= MAX_GPS_SAMPLE_GAP_SECONDS: d = haversine_km(prev[1], prev[2], curr[1], curr[2]) if d * 1000 >= GPS_MOTION_METERS: has_gps = True if has_obd: obd_motion.append(i) if has_gps: gps_motion.append(i) # Qualify GPS motion: need 3+ consecutive within 30s gap qualified_gps = set() seq = [] for idx in gps_motion: if not seq or (rows[idx][0] - rows[seq[-1]][0]).total_seconds() > MAX_GPS_SAMPLE_GAP_SECONDS: seq = [idx] else: seq.append(idx) if len(seq) >= MIN_GPS_MOTION_SAMPLES: qualified_gps.update(seq) # All motion indices, sorted motion_indices = sorted(set(obd_motion) | qualified_gps) if not motion_indices: return 0 # Group motion indices into trip boundaries by inactivity gap groups = [] for idx in motion_indices: if not groups or (rows[idx][0] - rows[groups[-1][-1]][0]).total_seconds() > INACTIVITY_SECONDS: groups.append([idx]) else: groups[-1].append(idx) # Build trip records from each group trips = [] for group in groups: start_idx = max(0, group[0] - 1) end_idx = group[-1] seg_rows = rows[start_idx:end_idx + 1] distance = 0 speeds = [] # OBD speeds — used for avg max_speeds = [] # OBD + GPS-derived — used for max for i in range(1, len(seg_rows)): d = haversine_km(seg_rows[i - 1][1], seg_rows[i - 1][2], seg_rows[i][1], seg_rows[i][2]) distance += d s = seg_rows[i][3] if s is not None and s > 0: speeds.append(s) max_speeds.append(s) else: # GPS-derived fallback for max (mirror the Pi, but smoothed over # a trailing ~30s window so GPS position jitter can't spike the max). j = max(0, i - 6) w_dt = (seg_rows[i][0] - seg_rows[j][0]).total_seconds() w_d = haversine_km(seg_rows[j][1], seg_rows[j][2], seg_rows[i][1], seg_rows[i][2]) if w_dt > 0 and w_d * 1000 >= 8: w_speed = w_d / w_dt * 3600 if w_speed <= 220: max_speeds.append(w_speed) duration_s = int((seg_rows[-1][0] - seg_rows[0][0]).total_seconds()) if duration_s < 30 or distance < 0.05: continue trips.append({ "start_time": seg_rows[0][0], "end_time": seg_rows[-1][0], "distance_km": round(distance, 3), "duration_s": duration_s, "avg_speed_kmh": round(sum(speeds) / len(speeds), 1) if speeds else None, "max_speed_kmh": round(max(max_speeds), 1) if max_speeds else None, "start_lat": seg_rows[0][1], "start_lon": seg_rows[0][2], "end_lat": seg_rows[-1][1], "end_lon": seg_rows[-1][2], "point_count": len(seg_rows), }) # Skip trips that were manually deleted (recompute won't re-insert them). with get_cursor() as cur: cur.execute( "SELECT start_time FROM deleted_trip_markers WHERE device_id = %s", (device_id,), ) deleted_starts = {row[0] for row in cur.fetchall()} trips = [t for t in trips if t["start_time"] not in deleted_starts] 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}, ) # BIGSERIAL consumes a sequence value on every INSERT attempt, even when # the ON CONFLICT (DO UPDATE) path is taken. Clamp the sequence back to the # real max id so trip IDs don't inflate by trip_count on every recompute. cur.execute( "SELECT setval(pg_get_serial_sequence('trips','id'), " "(SELECT COALESCE(MAX(id), 0) FROM trips))" ) return len(trips) def list_trips(device_id, limit=200): with get_cursor() as cur: cur.execute( "SELECT odometer_baseline FROM devices WHERE id = %s", (device_id,), ) base_row = cur.fetchone() 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 ASC """, (device_id,), ) rows = cur.fetchall() baseline = float(base_row[0]) if base_row and base_row[0] is not None else 0.0 # Simulate a continuous odometer: baseline + cumulative trip distance. cumulative = baseline trips = [] for r in rows: dist = float(r[3]) if r[3] is not None else 0.0 trips.append({ "id": r[0], "start_datetime": r[1].isoformat(), "end_datetime": r[2].isoformat(), "distance_km": dist, "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": round(cumulative, 1), "odo_end_km": round(cumulative + dist, 1), }) cumulative += dist trips.reverse() # newest first for display return trips[:limit] 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, odometer 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() cur.execute( "SELECT COALESCE(SUM(distance_km), 0) FROM trips WHERE device_id = %s AND start_time < %s", (device_id, trip_row[0]), ) prior = cur.fetchone() cur.execute( "SELECT odometer_baseline FROM devices WHERE id = %s", (device_id,), ) base_row = cur.fetchone() baseline = float(base_row[0]) if base_row and base_row[0] is not None else 0.0 prior_km = float(prior[0]) if prior and prior[0] is not None else 0.0 import math def hav(lat1, lon1, lat2, lon2): r = 6371.0088 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)) result_points = [] cumulative = 0.0 prev = None gps_window = [] # trailing (lat, lon, datetime) points for GPS speed smoothing for p in points: lat = float(p[1]) lon = float(p[2]) speed = p[3] or 0 raw_odo = p[4] speed_source = "obd" if speed > 0 else "gps" # Mirror the Pi's route_points GPS-speed fallback: when OBD speed is # missing (adapter dropout), derive speed from a trailing ~30s window # of movement (not the previous point) so the replay doesn't flatline # at 0 km/h while the car moves — and a single jittery 5s segment can't # spike the readout. if speed <= 0 and gps_window: start = gps_window[0] seconds = (p[0] - start[2]).total_seconds() if seconds > 0: dist_km = hav(start[0], start[1], lat, lon) w_speed = dist_km / seconds * 3600 if w_speed <= 220 and dist_km * 1000 >= 8: speed = w_speed if prev is not None: cumulative += hav(prev[0], prev[1], lat, lon) prev = (lat, lon, p[0]) gps_window.append((lat, lon, p[0])) if len(gps_window) > 6: gps_window.pop(0) if raw_odo is not None and float(raw_odo) > 0: odometer = float(raw_odo) else: odometer = baseline + prior_km + cumulative result_points.append({ "datetime": p[0].isoformat(), "latitude": lat, "longitude": lon, "speed": round(speed, 1), # Mirror the Pi's route_points: OBD speed when we have it, # otherwise the point is sourced from GPS. "speed_source": speed_source, "odometer": round(odometer, 1), }) 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": result_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 and mark its time range as deleted so auto-recompute doesn't re-insert it. Raw tracking_points are kept.""" with get_cursor(commit=True) as cur: # Record the start_time before deleting so recompute skips this trip. cur.execute( "SELECT start_time FROM trips WHERE id = %s AND device_id = %s", (trip_id, device_id), ) row = cur.fetchone() if row is None: return False cur.execute( """INSERT INTO deleted_trip_markers (device_id, start_time) VALUES (%s, %s) ON CONFLICT DO NOTHING""", (device_id, row[0]), ) cur.execute( "DELETE FROM trips WHERE id = %s AND device_id = %s", (trip_id, device_id), ) return True def delete_trips_batch(device_id, trip_ids): """Delete multiple trips in one transaction.""" deleted = 0 for tid in trip_ids: if delete_trip(device_id, tid): deleted += 1 return deleted def clear_deleted_trip_markers(device_id): """Remove all deleted trip markers so recompute rebuilds everything.""" with get_cursor(commit=True) as cur: cur.execute( "DELETE FROM deleted_trip_markers WHERE device_id = %s", (device_id,), ) 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, }