yaris-obd/modules/app_upload.py

"""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 os
import time
from threading import Thread

import requests
from dotenv import load_dotenv
from requests_toolbelt import MultipartEncoder

from modules import app_sql, app_dash, app_cellular

_BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
load_dotenv(os.path.join(_BASE_DIR, "db", "upload_creds.env"))

SERVER_URL = os.getenv("YARIS_SERVER_URL", "").rstrip("/")
DEVICE_NAME = os.getenv("YARIS_DEVICE_NAME", "")
API_KEY = os.getenv("YARIS_API_KEY", "")

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"

# How long a phone-relay "lease" stays valid without a fresh heartbeat before
# the Pi's own background uploader assumes the phone gave up/closed its tab
# and resumes uploading dashcam files itself. The phone page is expected to
# heartbeat well under this (every ~15-20s) - see /api/dash/relay/heartbeat.
RELAY_LEASE_SECONDS = 45

_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
_relay_lease_until = None  # monotonic time.time() the phone-relay lease expires, or None


def _headers():
    return {"X-Device-Name": DEVICE_NAME, "X-Api-Key": API_KEY}


def is_configured():
    return bool(SERVER_URL and DEVICE_NAME and 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 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(),
        "relay_active": is_relay_active(),
    }


# --- Phone-relay lease (pauses the Pi's own background dashcam uploader
# while a phone is manually relaying files, so they don't fight over the
# Pi's single, often WiFi-congested, radio) -------------------------------

def relay_heartbeat():
    """Called by the phone-relay page every ~15-20s while active."""
    global _relay_lease_until
    _relay_lease_until = time.time() + RELAY_LEASE_SECONDS


def relay_stop():
    """Called when the phone-relay page is done/closed cleanly."""
    global _relay_lease_until
    _relay_lease_until = None


def is_relay_active():
    return _relay_lease_until is not None and time.time() < _relay_lease_until


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 both the background loop and the phone-relay pending
    list 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 list_pending_dashcam_files():
    """Return [{relative_path, sha256, size}, ...] for dashcam files not yet
    confirmed uploaded - used by the phone-relay page to pick what to fetch."""
    pending = []
    for abs_path, rel_path in app_dash.find_video_files(app_dash.DEFAULT_DESTINATION):
        sha256 = _hash_dashcam_file(abs_path, rel_path)
        if app_sql.dashcam_file_marked_uploaded(rel_path, sha256):
            continue
        pending.append({
            "relative_path": rel_path,
            "sha256": sha256,
            "size": os.path.getsize(abs_path),
        })
    return pending


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

    rows = app_sql.get_unsent_tracking_rows(limit=TRACKING_BATCH_SIZE)
    if not rows:
        return 0

    points = [
        {
            "recorded_at": 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"{SERVER_URL}/api/v1/tracking/batch",
            json={"points": points},
            headers=_headers(),
            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...")
        import json as _json

        body_bytes = _json.dumps({"points": points}).encode()
        try:
            status_code, _body = app_cellular.http_post_json(
                f"{SERVER_URL}/api/v1/tracking/batch", body_bytes, _headers()
            )
        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.
    """
    status_resp = requests.get(
        f"{SERVER_URL}/api/v1/dashcam/upload/status",
        params={"relative_path": rel_path, "sha256": sha256},
        headers=_headers(),
        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"{SERVER_URL}/api/v1/dashcam/upload/chunk",
                data=encoder,
                headers={**_headers(), "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, or
    while a phone-relay session is actively leased - tracking-log uploads are
    unaffected by either of these.
    """
    if not is_dashcam_upload_enabled() or is_relay_active():
        return 0

    uploaded_count = 0
    for abs_path, rel_path in app_dash.find_video_files(app_dash.DEFAULT_DESTINATION):
        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

        # Re-check between files so flipping the toggle off, or a phone
        # relay starting, mid-run stops further uploads promptly without
        # waiting for the next cycle.
        if not is_dashcam_upload_enabled() or is_relay_active():
            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()
            if sent or uploaded:
                print(f"Upload: sent {sent} tracking rows, {uploaded} dashcam files")
            _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.")