"""Uploads locally-queued trip/GPS tracking rows and dashcam files to the remote server (see server/ in this repo) over HTTP, with retry/backoff. Tracking-log uploads always run when configured. Dashcam file uploads can be paused independently via a persisted toggle (settings table in the same SQLite db used for tracking) so a user can stop shipping large video files over a slow/metered connection while trip logs keep flowing. Network path: normal WiFi/Ethernet (via `requests`) is always tried first for everything. If a tracking-batch upload fails with a connection-level error (no route/DNS/timeout - not an HTTP error status, which means the server WAS reached) and the SIM868 cellular fallback is configured, the same small JSON batch is retried over the modem's AT-command HTTP client (see modules/app_cellular.py). Dashcam uploads never use cellular - only WiFi/Ethernet - since video files are far too large/slow for a GPRS/3G link; they simply wait for connectivity to return, same as before. *** The cellular fallback path is UNTESTED (no SIM card available) - see *** modules/app_cellular.py's docstring. It's opt-in/feature-flagged off by *** default (YARIS_CELLULAR_ENABLED unset) so this has zero effect on the *** already-verified WiFi upload path unless explicitly configured. Dashcam files are sent in DASHCAM_CHUNK_SIZE pieces via /api/v1/dashcam/upload/chunk rather than one request per file - confirmed against the real deployed server that Cloudflare Tunnel's free-tier ~100MB request-body limit otherwise rejects real dashcam clips (300+MB) outright with a 413, and that memory-safe streaming alone (MultipartEncoder) doesn't help there since the *proxy*, not this app, is what enforces the cap. Chunking also makes uploads resumable: a /status check before sending finds out which chunks the server already has, so an interrupted transfer picks back up instead of restarting from byte 0. Credentials/server URL are loaded from db/upload_creds.env (gitignored, never commit real values - mirrors the db/creds.env pattern already used by app_db.py for MongoDB). """ import hashlib import json import os import time from datetime import datetime from threading import Thread import requests from requests_toolbelt import MultipartEncoder from modules import app_sql, app_dash, app_cellular, app_remote_config, app_video UPLOAD_INTERVAL = 30 # seconds between upload attempts once caught up CATCHUP_INTERVAL = 1 # seconds between batches while draining a backlog TRACKING_BATCH_SIZE = 500 BACKOFF_INITIAL = 5 # seconds, doubles on each consecutive failure BACKOFF_MAX = 300 # Real dashcam clips run 300+MB - Cloudflare Tunnel's free-tier ~100MB max # request body size rejects those outright with a 413 (confirmed against # the real tunnel: a 250MB file uploads fine direct-LAN but 413s through # yaris.ceaf.dev). Files are split into chunks this size and sent one at a # time via /api/v1/dashcam/upload/chunk instead, comfortably under that cap. DASHCAM_CHUNK_SIZE = 40 * 1024 * 1024 DASHCAM_UPLOAD_SETTING_KEY = "dashcam_upload_enabled" TRIPS_UPLOAD_SETTING_KEY = "trips_upload_enabled" _upload_thread = None _last_error = None _last_success_at = None _last_network_path = None # "wifi" or "cellular" - last path a batch actually went out on def _credentials(): return app_remote_config.server_credentials() def _headers(credentials): return { "X-Device-Name": credentials["device_name"], "X-Api-Key": credentials["api_key"], } def is_configured(): credentials = _credentials() return bool( credentials["server_url"] and credentials["device_name"] and credentials["api_key"] ) def is_dashcam_upload_enabled(): """Defaults to enabled - the user opts out, not in.""" return app_sql.get_setting(DASHCAM_UPLOAD_SETTING_KEY, "1") == "1" def set_dashcam_upload_enabled(enabled): app_sql.set_setting(DASHCAM_UPLOAD_SETTING_KEY, "1" if enabled else "0") def is_trips_upload_enabled(): """Defaults to enabled.""" return app_sql.get_setting(TRIPS_UPLOAD_SETTING_KEY, "1") == "1" def get_status(): """Snapshot used by the /api/upload/status endpoint for the web UI.""" return { "configured": is_configured(), "dashcam_upload_enabled": is_dashcam_upload_enabled(), "pending_tracking_rows": app_sql.count_unsent_tracking_rows(), "last_error": _last_error, "last_success_at": _last_success_at, "last_network_path": _last_network_path, "cellular_fallback_configured": app_cellular.is_configured(), } def _sha256_file(path): hasher = hashlib.sha256() with open(path, "rb") as f: while True: chunk = f.read(1024 * 1024) if not chunk: break hasher.update(chunk) return hasher.hexdigest() def _hash_dashcam_file(abs_path, rel_path): """sha256 of a dashcam file, cached by (path, size, mtime) so unchanged files already copied from the dashcam never get rehashed - real backlogs here run 20GB+, and the background upload loop would otherwise rehash everything, including already-uploaded files, on every single call.""" stat = os.stat(abs_path) cached = app_sql.get_cached_dashcam_hash(rel_path, stat.st_size, stat.st_mtime) if cached is not None: return cached sha256 = _sha256_file(abs_path) app_sql.set_cached_dashcam_hash(rel_path, stat.st_size, stat.st_mtime, sha256) return sha256 def _to_tz_aware(dt_str): """Convert a naive datetime string (from SQLite) to a timezone-aware ISO string the server's TIMESTAMPTZ column will interpret correctly. SQLite stores naive local-time strings like '2026-08-10 19:27:30.123456'. Without a timezone, Postgres interprets these as UTC, shifting displayed times by the local offset. This helper attaches the Pi's local IANA offset so the server stores and displays the correct local time. """ try: dt = datetime.fromisoformat(str(dt_str)) except (TypeError, ValueError): return dt_str # unparseable — send as-is and let the server reject it if dt.tzinfo is not None: return dt_str # already TZ-aware — nothing to do return dt.astimezone().isoformat() def _upload_tracking_batch(): """Send up to TRACKING_BATCH_SIZE not-yet-uploaded rows. Returns count sent. Tries WiFi/Ethernet (`requests`) first. If that fails at the connection level (no route, DNS failure, timeout - as opposed to the server responding with an HTTP error, which means it WAS reached) and the SIM868 cellular fallback is configured, retries the same batch over the modem's AT-command HTTP client before giving up. """ global _last_network_path if not is_trips_upload_enabled(): return 0 rows = app_sql.get_unsent_tracking_rows(limit=TRACKING_BATCH_SIZE) if not rows: return 0 credentials = _credentials() points = [ { "recorded_at": _to_tz_aware(row["datetime"]), "gps_lat": row["gpsx"], "gps_lon": row["gpsy"], "speed": row["speed"], "odometer": row["odometer"], "vehicle_name": row["vn"], "driver_name": row["name"], } for row in rows ] row_ids = [row["id"] for row in rows] try: resp = requests.post( f"{credentials['server_url']}/api/v1/tracking/batch", json={"points": points}, headers=_headers(credentials), timeout=15, ) resp.raise_for_status() _last_network_path = "wifi" except requests.exceptions.ConnectionError as wifi_err: if not app_cellular.is_configured(): raise print(f"WiFi/Ethernet upload unreachable ({wifi_err}), trying cellular fallback...") body_bytes = json.dumps({"points": points}).encode() try: status_code, _body = app_cellular.http_post_json( f"{credentials['server_url']}/api/v1/tracking/batch", body_bytes, _headers(credentials), ) except app_cellular.CellularError as cell_err: # Neither path worked - raise the original WiFi error so the # normal backoff/retry logic in _upload_loop handles it. print(f"Cellular fallback also failed: {cell_err}") raise wifi_err if status_code >= 400: raise app_cellular.CellularError( f"cellular upload got HTTP {status_code}" ) _last_network_path = "cellular" app_sql.mark_tracking_rows_uploaded(row_ids) return len(rows) def _upload_video_chunked(abs_path, rel_path, sha256): """Upload one dashcam video file in DASHCAM_CHUNK_SIZE pieces, resuming from whatever the server already has for this exact (path, sha256) combo. Raises on any request-level failure - the caller's normal backlog/backoff handling takes it from there; a retry will simply re-query /status and resume from the first missing chunk instead of starting over. """ credentials = _credentials() status_resp = requests.get( f"{credentials['server_url']}/api/v1/dashcam/upload/status", params={"relative_path": rel_path, "sha256": sha256}, headers=_headers(credentials), timeout=15, ) status_resp.raise_for_status() status = status_resp.json() if status.get("status") == "already_uploaded": return already_received = set(status.get("received_chunk_indices", [])) size = os.path.getsize(abs_path) total_chunks = max(1, -(-size // DASHCAM_CHUNK_SIZE)) # ceil division with open(abs_path, "rb") as f: for idx in range(total_chunks): if idx in already_received: continue f.seek(idx * DASHCAM_CHUNK_SIZE) chunk_bytes = f.read(DASHCAM_CHUNK_SIZE) encoder = MultipartEncoder( fields={ "relative_path": rel_path, "sha256": sha256, "chunk_index": str(idx), "total_chunks": str(total_chunks), "file": (f"chunk_{idx:06d}", chunk_bytes, "application/octet-stream"), } ) resp = requests.post( f"{credentials['server_url']}/api/v1/dashcam/upload/chunk", data=encoder, headers={**_headers(credentials), "Content-Type": encoder.content_type}, timeout=120, ) resp.raise_for_status() def _upload_dashcam_files(): """Upload any not-yet-uploaded dashcam files, one at a time. Returns count sent. Skipped entirely (returns 0) while the dashcam-upload toggle is off - tracking-log uploads are unaffected. """ if not is_dashcam_upload_enabled(): return 0 uploaded_count = 0 for abs_path, rel_path in app_dash.find_video_files(app_dash.DEFAULT_DESTINATION): if not app_dash.begin_local_video_use(rel_path): continue try: if not os.path.isfile(abs_path): continue sha256 = _hash_dashcam_file(abs_path, rel_path) if app_sql.dashcam_file_marked_uploaded(rel_path, sha256): continue _upload_video_chunked(abs_path, rel_path, sha256) app_sql.mark_dashcam_file_uploaded(rel_path, sha256) uploaded_count += 1 finally: app_dash.end_local_video_use(rel_path) # Re-check between files so flipping the toggle off mid-run stops # further uploads promptly without waiting for the next cycle. if not is_dashcam_upload_enabled(): break return uploaded_count def _upload_loop(): global _last_error, _last_success_at import datetime as _dt backoff = BACKOFF_INITIAL while True: if not is_configured(): time.sleep(UPLOAD_INTERVAL) continue try: sent = _upload_tracking_batch() uploaded = _upload_dashcam_files() cleanup = app_video.cleanup_uploaded_videos() if sent or uploaded: print(f"Upload: sent {sent} tracking rows, {uploaded} dashcam files") if cleanup["deleted"]: print( f"Retention: deleted {len(cleanup['deleted'])} uploaded local videos" ) if cleanup["errors"]: print("Retention cleanup errors:", "; ".join(cleanup["errors"])) _last_error = None _last_success_at = _dt.datetime.now().isoformat() backoff = BACKOFF_INITIAL # A full batch means there's likely more backlog queued up (the # tracking table logs a new row every 5s, so a stale/offline # device can build up thousands of rows quickly) - keep draining # back-to-back instead of idling for the full interval. Only # settle into the slow/idle cadence once a batch comes back # short, i.e. we're actually caught up. if sent >= TRACKING_BATCH_SIZE: time.sleep(CATCHUP_INTERVAL) else: time.sleep(UPLOAD_INTERVAL) except (requests.RequestException, app_cellular.CellularError) as e: _last_error = str(e) print(f"Upload failed ({e}), retrying in {backoff}s") time.sleep(backoff) backoff = min(backoff * 2, BACKOFF_MAX) def start_uploading(): """Start the background upload thread, if the server is configured.""" global _upload_thread if not is_configured(): print( "Upload client not configured (set YARIS_SERVER_URL/YARIS_DEVICE_NAME/" "YARIS_API_KEY in db/upload_creds.env) - uploads disabled." ) return if _upload_thread is None or not _upload_thread.is_alive(): _upload_thread = Thread(target=_upload_loop, daemon=True) _upload_thread.start() print("Upload-client thread started.") else: print("Upload-client thread already running.")