370 lines
14 KiB
Python
370 lines
14 KiB
Python
"""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.")
|