"""yaris-obd remote server: receives trip/GPS log batches and dashcam files uploaded by the Pi, storing them in Postgres + on-disk file storage. Also serves the remote web portal for browsing and managing the configured car. """ import functools import hashlib import os import secrets import shutil from flask import ( Flask, request, jsonify, g, render_template, redirect, url_for, flash, session, Response, abort ) from flask_login import login_user, logout_user, current_user from modules import db from modules.auth import require_device_auth from modules.user_auth import login_manager, User, check_password, require_login from modules import device_config, device_registry # ── Geocode cache ────────────────────────────────────────────────────────── try: import reverse_geocoder as _rg _RG_AVAILABLE = True except ImportError: _RG_AVAILABLE = False @functools.lru_cache(maxsize=512) def _geocode_cached(lat4, lon4): lat = lat4 / 10000.0 lon = lon4 / 10000.0 try: import urllib.request, urllib.error, json as _json url = ( f"https://nominatim.openstreetmap.org/reverse" f"?lat={lat}&lon={lon}&format=json&addressdetails=1" ) req = urllib.request.Request(url, headers={"User-Agent": "yaris-obd-server/1.0"}) with urllib.request.urlopen(req, timeout=4) as resp: data = _json.loads(resp.read()) addr = data.get("address", {}) road = addr.get("road") or addr.get("pedestrian") or addr.get("cycleway", "") city = addr.get("city") or addr.get("town") or addr.get("village", "") return f"{road}, {city}".strip(", ") if road or city else None except Exception: pass if _RG_AVAILABLE: results = _rg.search([(lat, lon)], verbose=False) if results: r = results[0] return f"{r.get('name', '')}, {r.get('admin1', '')}".strip(", ") return None app = Flask(__name__) # Secret key for browser sessions — read from env in production. app.secret_key = os.environ.get("SECRET_KEY") or secrets.token_hex(32) # Where uploaded dashcam files get written on disk (mounted to the unraid # array via docker-compose - see DASHCAM_STORAGE_DIR in .env). STORAGE_DIR = os.environ.get("DASHCAM_STORAGE_DIR", "/data/dashcam") MAX_DASHCAM_UPLOAD_MB = int(os.environ.get("MAX_DASHCAM_UPLOAD_MB", "2048")) app.config["MAX_CONTENT_LENGTH"] = MAX_DASHCAM_UPLOAD_MB * 1024 * 1024 # Sanity cap on tracking points per request - the Pi client batches in # groups of 500, this just guards against a runaway/misbehaving client. MAX_TRACKING_BATCH_POINTS = 5000 # Initialise Flask-Login login_manager.init_app(app) def _portal_device_id(): """Return the device id for the configured device name (single Pi setup).""" name = os.environ.get("DEVICE_NAME", "yaris-2013") dev = db.get_device_by_name(name) return dev["id"] if dev else None @app.get("/health") def health(): """Unauthenticated liveness check - used by docker-compose healthcheck and to sanity-test connectivity from the Pi before attempting uploads.""" return jsonify(status="ok"), 200 @app.post("/api/v1/tracking/batch") @require_device_auth def upload_tracking_batch(): """Bulk-insert a batch of GPS/trip tracking points for the authenticated device. Expected JSON body: {"points": [{"recorded_at": "...", "gps_lat": ..., ...}, ...]} recorded_at must be an ISO-8601 timestamp string. Duplicate (device, recorded_at) pairs are silently ignored, so the Pi can safely re-send a batch after a network failure without creating duplicate rows. """ body = request.get_json(silent=True) or {} points = body.get("points") if not isinstance(points, list) or not points: return jsonify(error="expected a non-empty 'points' list"), 400 if len(points) > MAX_TRACKING_BATCH_POINTS: return jsonify( error=f"batch too large - max {MAX_TRACKING_BATCH_POINTS} points per request" ), 400 for p in points: if "recorded_at" not in p: return jsonify(error="every point requires 'recorded_at'"), 400 inserted = db.insert_tracking_points(g.device_id, points) # Recompute trips in background thread so portal reflects new data immediately. if inserted > 0: import threading threading.Thread( target=db.compute_trips_for_device, args=(g.device_id,), daemon=True, ).start() return jsonify(status="ok", received=inserted), 201 @app.post("/api/v1/dashcam/upload") @require_device_auth def upload_dashcam_file(): """Upload a single dashcam video file. Multipart form fields: file: the video file itself relative_path: original sub-folder path on the dashcam SD card (form field) sha256: hex digest of the file, computed by the Pi before sending Skips (returns 200, not re-saved) if a file with the same (device, relative_path, sha256) was already uploaded - safe to retry. """ if "file" not in request.files: return jsonify(error="missing 'file' in multipart form"), 400 relative_path = request.form.get("relative_path", "").strip() expected_sha256 = request.form.get("sha256", "").strip().lower() if not relative_path or not expected_sha256: return jsonify(error="missing 'relative_path' or 'sha256' form field"), 400 device_id = g.device_id if db.dashcam_file_exists(device_id, relative_path, expected_sha256): return jsonify(status="already_uploaded"), 200 uploaded = request.files["file"] device_dir = os.path.join(STORAGE_DIR, g.device_name) dest_path = os.path.normpath(os.path.join(device_dir, relative_path)) # Guard against a malicious/buggy relative_path (e.g. "../../etc") escaping # the per-device storage directory. if os.path.commonpath([device_dir, dest_path]) != os.path.normpath(device_dir): return jsonify(error="invalid relative_path"), 400 os.makedirs(os.path.dirname(dest_path), exist_ok=True) hasher = hashlib.sha256() tmp_path = dest_path + ".part" size_bytes = 0 with open(tmp_path, "wb") as f: while True: chunk = uploaded.stream.read(1024 * 1024) if not chunk: break hasher.update(chunk) size_bytes += len(chunk) f.write(chunk) actual_sha256 = hasher.hexdigest() if actual_sha256 != expected_sha256: os.remove(tmp_path) return jsonify(error="sha256 mismatch, upload corrupted in transit"), 422 os.replace(tmp_path, dest_path) db.insert_dashcam_file(device_id, relative_path, size_bytes, actual_sha256, dest_path) return jsonify(status="ok", size_bytes=size_bytes), 201 # --- Chunked dashcam upload --------------------------------------------- # Real dashcam clips run 300+MB, well past Cloudflare Tunnel's ~100MB max # request body size (confirmed with a real 250MB file: a direct LAN upload # succeeds, the same request through the tunnel comes back 413). The Pi # client splits files into small chunks (see CHUNK_SIZE in # modules/app_upload.py) and PUTs them one at a time through this endpoint # instead, staying comfortably under that limit. Chunks for a given upload # are staged on disk under _CHUNK_TMP_DIR and reassembled once complete. _CHUNK_TMP_DIRNAME = ".upload_tmp" def _chunk_tmp_dir(device_name, sha256): return os.path.join(STORAGE_DIR, _CHUNK_TMP_DIRNAME, device_name, sha256) def _received_chunk_indices(tmp_dir): if not os.path.isdir(tmp_dir): return set() indices = set() for name in os.listdir(tmp_dir): if name.endswith(".chunk"): try: indices.add(int(name[: -len(".chunk")])) except ValueError: continue return indices @app.get("/api/v1/dashcam/upload/status") @require_device_auth def dashcam_upload_status(): """Report upload progress for (relative_path, sha256) so the Pi client can resume a chunked upload without re-sending chunks the server already has (e.g. after a network drop or app restart mid-transfer).""" relative_path = request.args.get("relative_path", "").strip() sha256 = request.args.get("sha256", "").strip().lower() if not relative_path or not sha256: return jsonify(error="missing 'relative_path' or 'sha256' query param"), 400 if db.dashcam_file_exists(g.device_id, relative_path, sha256): return jsonify(status="already_uploaded"), 200 tmp_dir = _chunk_tmp_dir(g.device_name, sha256) received = sorted(_received_chunk_indices(tmp_dir)) return jsonify(status="in_progress" if received else "not_started", received_chunk_indices=received), 200 @app.post("/api/v1/dashcam/upload/chunk") @require_device_auth def dashcam_upload_chunk(): """Receive one chunk of a dashcam video file. Multipart form fields: file: the chunk's bytes relative_path: original sub-folder path on the dashcam SD card sha256: hex digest of the *whole* (reassembled) file chunk_index: 0-based index of this chunk total_chunks: total number of chunks for this file Once every chunk 0..total_chunks-1 has arrived, the chunks are concatenated in order, the combined sha256 is verified, and the result is recorded exactly like the single-shot /upload endpoint. """ if "file" not in request.files: return jsonify(error="missing 'file' in multipart form"), 400 relative_path = request.form.get("relative_path", "").strip() expected_sha256 = request.form.get("sha256", "").strip().lower() try: chunk_index = int(request.form.get("chunk_index", "")) total_chunks = int(request.form.get("total_chunks", "")) except ValueError: return jsonify(error="'chunk_index'/'total_chunks' must be integers"), 400 if not relative_path or not expected_sha256 or total_chunks < 1 or not (0 <= chunk_index < total_chunks): return jsonify(error="missing/invalid form fields"), 400 device_id = g.device_id if db.dashcam_file_exists(device_id, relative_path, expected_sha256): return jsonify(status="already_uploaded"), 200 device_dir = os.path.join(STORAGE_DIR, g.device_name) dest_path = os.path.normpath(os.path.join(device_dir, relative_path)) if os.path.commonpath([device_dir, dest_path]) != os.path.normpath(device_dir): return jsonify(error="invalid relative_path"), 400 tmp_dir = _chunk_tmp_dir(g.device_name, expected_sha256) # A stale tmp_dir left over from a previous attempt with a *different* # total_chunks (e.g. the client's chunk size changed between app # versions) would reassemble incorrectly - start clean if that happens. marker_path = os.path.join(tmp_dir, ".total_chunks") if os.path.isdir(tmp_dir) and os.path.exists(marker_path): with open(marker_path) as marker_file: if marker_file.read().strip() != str(total_chunks): shutil.rmtree(tmp_dir) os.makedirs(tmp_dir, exist_ok=True) with open(marker_path, "w") as marker_file: marker_file.write(str(total_chunks)) chunk_path = os.path.join(tmp_dir, f"{chunk_index:06d}.chunk") uploaded = request.files["file"] with open(chunk_path, "wb") as f: while True: data = uploaded.stream.read(1024 * 1024) if not data: break f.write(data) received = _received_chunk_indices(tmp_dir) if len(received) < total_chunks: return jsonify(status="chunk_received", received_chunks=len(received), total_chunks=total_chunks), 202 # All chunks in - reassemble, verify, and store. hasher = hashlib.sha256() tmp_path = dest_path + ".part" os.makedirs(os.path.dirname(dest_path), exist_ok=True) size_bytes = 0 with open(tmp_path, "wb") as out: for idx in range(total_chunks): with open(os.path.join(tmp_dir, f"{idx:06d}.chunk"), "rb") as chunk_file: while True: data = chunk_file.read(1024 * 1024) if not data: break hasher.update(data) size_bytes += len(data) out.write(data) actual_sha256 = hasher.hexdigest() shutil.rmtree(tmp_dir, ignore_errors=True) if actual_sha256 != expected_sha256: os.remove(tmp_path) return jsonify(error="sha256 mismatch, upload corrupted in transit"), 422 os.replace(tmp_path, dest_path) db.insert_dashcam_file(device_id, relative_path, size_bytes, actual_sha256, dest_path) return jsonify(status="complete", size_bytes=size_bytes), 201 # ── Browser auth ─────────────────────────────────────────────────────────── @app.get("/login") def login_page(): if current_user.is_authenticated: return redirect(url_for("portal_index")) csrf_token = secrets.token_hex(16) session["csrf_token"] = csrf_token return render_template("login.html", csrf_token=csrf_token) @app.post("/login") def login_submit(): # CSRF check token = request.form.get("csrf_token", "") if not token or token != session.pop("csrf_token", None): flash("Invalid request. Please try again.", "error") return redirect(url_for("login_page")) username = request.form.get("username", "").strip() password = request.form.get("password", "") user_row = db.get_user_by_username(username) if user_row and check_password(password, user_row["password_hash"]): login_user(User(user_row["id"], user_row["username"]), remember=False) return redirect(url_for("portal_index")) flash("Invalid username or password.", "error") return redirect(url_for("login_page")) @app.get("/logout") @require_login def logout(): logout_user() return redirect(url_for("login_page")) @app.get("/account/password") @require_login def change_password_page(): csrf_token = secrets.token_hex(16) session["csrf_token"] = csrf_token return render_template("change_password.html", csrf_token=csrf_token) @app.post("/account/password") @require_login def change_password(): token = request.form.get("csrf_token", "") if not token or token != session.pop("csrf_token", None): flash("Invalid request. Please try again.", "error") return redirect(url_for("change_password_page")) current = request.form.get("current_password", "") new_pw = request.form.get("new_password", "") confirm = request.form.get("confirm_password", "") user_row = db.get_user_by_id(current_user.id) if not user_row or not check_password(current, user_row["password_hash"]): flash("Current password is incorrect.", "error") return redirect(url_for("change_password_page")) if len(new_pw) < 8: flash("New password must be at least 8 characters.", "error") return redirect(url_for("change_password_page")) if new_pw != confirm: flash("Passwords do not match.", "error") return redirect(url_for("change_password_page")) from modules.user_auth import hash_password with db.get_cursor(commit=True) as cur: cur.execute("UPDATE users SET password_hash = %s WHERE id = %s", (hash_password(new_pw), current_user.id)) flash("Password updated successfully.", "success") return redirect(url_for("change_password_page")) # ── Portal UI pages ──────────────────────────────────────────────────────── @app.get("/") @require_login def portal_index(): return render_template("index.html") @app.get("/history") @require_login def portal_history(): return render_template("history.html") @app.get("/dashcam") @require_login def portal_dashcam(): return render_template("dashcam.html") @app.get("/settings/device") @require_login def portal_device_settings_page(): return render_template( "device_settings.html", configured_device_name=os.environ.get("DEVICE_NAME", "yaris-2013"), ) # ── Portal JSON API ──────────────────────────────────────────────────────── @app.get("/api/v1/portal/stats") @require_login def portal_stats(): device_id = _portal_device_id() if device_id is None: return jsonify(error="device not configured"), 404 return jsonify(db.get_overview_stats(device_id)) @app.get("/api/v1/portal/trips") @require_login def portal_trips(): device_id = _portal_device_id() if device_id is None: return jsonify(error="device not configured"), 404 try: limit = min(int(request.args.get("limit", 200)), 1000) except (ValueError, TypeError): limit = 200 return jsonify(db.list_trips(device_id, limit=limit)) @app.get("/api/v1/portal/trip/") @require_login def portal_trip_detail(trip_id): device_id = _portal_device_id() if device_id is None: return jsonify(error="device not configured"), 404 data = db.get_trip_route(device_id, trip_id) if data is None: return jsonify(error="trip not found"), 404 return jsonify(data) @app.get("/api/v1/portal/dashcam") @require_login def portal_dashcam_list(): device_id = _portal_device_id() if device_id is None: return jsonify(error="device not configured"), 404 try: limit = min(int(request.args.get("limit", 500)), 2000) except (ValueError, TypeError): limit = 500 return jsonify(db.list_dashcam_files(device_id, limit=limit)) @app.get("/api/v1/portal/dashcam/stream/") @require_login def portal_dashcam_stream(file_id): """Stream a dashcam file with Range request support for video seeking.""" device_id = _portal_device_id() if device_id is None: abort(404) file_row = db.get_dashcam_file(device_id, file_id) if file_row is None: abort(404) storage_path = file_row["storage_path"] # Safety check: must still be under STORAGE_DIR try: safe = os.path.commonpath([STORAGE_DIR, os.path.normpath(storage_path)]) == os.path.normpath(STORAGE_DIR) except ValueError: safe = False if not safe: abort(403) if not os.path.isfile(storage_path): abort(404) file_size = os.path.getsize(storage_path) range_header = request.headers.get("Range") mime = "video/mp4" filename = file_row["relative_path"].split("/")[-1] if not range_header: # Full file download def generate_full(): with open(storage_path, "rb") as f: while True: chunk = f.read(1024 * 1024) if not chunk: break yield chunk headers = { "Content-Disposition": f'inline; filename="{filename}"', "Content-Length": file_size, "Accept-Ranges": "bytes", } return Response(generate_full(), status=200, headers=headers, mimetype=mime) # Parse Range header (bytes=start-end) try: range_val = range_header.replace("bytes=", "") start_s, end_s = range_val.split("-") start = int(start_s) end = int(end_s) if end_s else file_size - 1 except Exception: abort(416) if start >= file_size or end >= file_size or start > end: abort(416) length = end - start + 1 def generate_range(): with open(storage_path, "rb") as f: f.seek(start) remaining = length while remaining > 0: chunk = f.read(min(1024 * 1024, remaining)) if not chunk: break remaining -= len(chunk) yield chunk headers = { "Content-Range": f"bytes {start}-{end}/{file_size}", "Content-Length": length, "Accept-Ranges": "bytes", "Content-Disposition": f'inline; filename="{filename}"', } return Response(generate_range(), status=206, headers=headers, mimetype=mime) # ── Geocode (same offline logic as Pi) ──────────────────────────────────── @app.get("/api/geocode") @require_login def geocode(): try: lat = float(request.args["lat"]) lon = float(request.args["lon"]) except (KeyError, ValueError): return jsonify(error="lat and lon required"), 400 lat4 = int(round(lat, 4) * 10000) lon4 = int(round(lon, 4) * 10000) name = _geocode_cached(lat4, lon4) if name: return jsonify(label=name) return jsonify(name=None), 204 # ── Uplink trigger (recompute trips after new batch) ────────────────────── @app.post("/api/v1/portal/recompute-trips") @require_login def recompute_trips(): """Manually trigger trip recomputation. Called after upload batch for freshness.""" device_id = _portal_device_id() if device_id is None: return jsonify(error="device not configured"), 404 n = db.compute_trips_for_device(device_id) return jsonify(status="ok", trips_computed=n) @app.delete("/api/v1/portal/trip/") @require_login def portal_delete_trip(trip_id): device_id = _portal_device_id() if device_id is None: return jsonify(error="device not configured"), 404 deleted = db.delete_trip(device_id, trip_id) if not deleted: return jsonify(error="trip not found"), 404 return jsonify(status="ok") @app.delete("/api/v1/portal/dashcam/") @require_login def portal_delete_dashcam(file_id): device_id = _portal_device_id() if device_id is None: return jsonify(error="device not configured"), 404 storage_path = db.delete_dashcam_file(device_id, file_id) if storage_path is None: return jsonify(error="file not found"), 404 # Remove from disk (best-effort — don't fail if file is already gone) try: if os.path.isfile(storage_path): os.remove(storage_path) except OSError: pass return jsonify(status="ok") @app.get("/api/trips") @require_login def portal_trips_compat(): """Pi history.js compatibility alias — wraps response in {trips: [...]}.""" device_id = _portal_device_id() if device_id is None: return jsonify(error="device not configured"), 404 limit = min(int(request.args.get("limit", 200)), 1000) trips = db.list_trips(device_id, limit=limit) return jsonify(trips=trips) @app.get("/api/trip/") @require_login def portal_trip_compat(trip_id): """Pi history.js compatibility alias.""" return portal_trip_detail(trip_id) # ── Phase 25.5a: Device configuration sync ────────────────────────────────── @app.get("/api/v1/portal/devices") @require_login def portal_devices(): devices = device_registry.list_devices() configured_name = os.environ.get("DEVICE_NAME", "yaris-2013") for device in devices: device["portal_default"] = device["name"] == configured_name return jsonify( devices=devices, configured_device_name=configured_name, ) @app.post("/api/v1/portal/devices") @require_login def portal_device_create(): body = request.get_json(silent=True) or {} result, errors = device_registry.create_device(body.get("name")) if errors: return jsonify(error="Invalid device.", fields=errors), 400 return jsonify(result), 201 @app.get("/api/v1/portal/devices//config") @require_login def portal_device_config_detail(device_id): config = device_config.get_device_config(device_id) if not config: return jsonify(error="device not found"), 404 device = db.get_device_by_id(device_id) return jsonify( device=device, config=config, ) @app.post("/api/v1/portal/devices//config") @require_login def portal_device_config_update(device_id): success, errors = device_config.update_device_config( device_id, request.get_json(silent=True) or {}, ) if not success: return jsonify(error="Invalid device config.", fields=errors), 400 return jsonify( device=db.get_device_by_id(device_id), config=device_config.get_device_config(device_id), ) @app.post("/api/v1/portal/devices//rotate-key") @require_login def portal_device_rotate_key(device_id): result, errors = device_registry.rotate_device_key(device_id) if errors: return jsonify(error=errors.get("error", "Could not rotate key.")), 404 return jsonify(result) @app.post("/api/v1/portal/devices//request-sync") @require_login def portal_device_request_sync(device_id): result = device_registry.request_sync(device_id) if result is None: return jsonify(error="device not found"), 404 return jsonify(result) @app.get("/api/v1/portal/device-config") @require_login def portal_device_config_get(): """Get the configured device identity for the remote portal.""" device_id = _portal_device_id() if device_id is None: return jsonify(error="device not configured"), 404 config = device_config.get_device_config(device_id) if not config: return jsonify(error="device not found"), 404 config["configured_device_name"] = os.environ.get("DEVICE_NAME", "yaris-2013") return jsonify(config) @app.get("/api/device-config") @require_device_auth def api_device_config_get(): """Get device configuration for the authenticated device. Pi calls this to fetch config from server (connection-driven sync). Returns owner_name, car_model, device_name, avatar_path, config_version. """ config = device_config.get_device_config(g.device_id) if not config: return jsonify(error="device not found"), 404 return jsonify(config) @app.post("/api/v1/device-config/ack") @require_device_auth def api_device_config_ack(): body = request.get_json(silent=True) or {} try: config_version = int(body.get("config_version")) except (TypeError, ValueError): return jsonify(error="config_version is required"), 400 device_registry.acknowledge_sync(g.device_id, config_version) return jsonify(status="ok") @app.post("/api/device-config") @require_login def api_device_config_update(): """Update device configuration (server portal/admin only). Updates owner_name, car_model, device_name, avatar_path. Increments config_version to signal change to Pi. """ device_id = _portal_device_id() if device_id is None: return jsonify(error="device not configured"), 404 body = request.get_json(silent=True) or {} success, errors = device_config.update_device_config(device_id, body) if not success: return jsonify(error="Invalid device config.", fields=errors), 400 config = device_config.get_device_config(device_id) return jsonify(config) def main(): app.run(host="0.0.0.0", port=8098) if __name__ == "__main__": main()