sunnylink: enable uploader option for admins (#1152)

* feat: add sunnylink uploader option for admins in sunnylink panel

* feat: enhance uploader to support zstd compression and improve route handling

* feat: update sunnylink uploader description and enablement criteria for admin tiers

* ui cleanup

---------

Co-authored-by: Jason Wen <haibin.wen3@gmail.com>

selfdrive/ui/sunnypilot/qt/offroad/settings/sunnylink_panel.cc

@@ -66,6 +66,14 @@ SunnylinkPanel::SunnylinkPanel(QWidget *parent) : QFrame(parent) {
   });
   list->addItem(horizontal_line());
 
+  QString sunnylinkUploaderDesc = tr("Enable sunnylink uploader to allow sunnypilot to upload your driving data to sunnypilot servers. (only for highest tiers, and does NOT bring ANY benefit to you. We are just testing data volume.)");
+  sunnylinkUploaderEnabledBtn = new ParamControlSP(
+    "EnableSunnylinkUploader",
+    tr("[Don't use] Enable sunnylink uploader"),
+    sunnylinkUploaderDesc,
+    "", nullptr, true);
+  list->addItem(sunnylinkUploaderEnabledBtn);
+
   connect(sunnylinkEnabledBtn, &ParamControl::showDescriptionEvent, [=]() {
     // resets the description to the default one for the Easter egg
     sunnylinkEnabledBtn->setDescription(sunnylinkEnabledBtnDesc);
@@ -261,6 +269,8 @@ void SunnylinkPanel::updatePanel() {
   pairSponsorBtn->setEnabled(!is_onroad && is_sunnylink_enabled);
   pairSponsorBtn->setValue(is_paired ? tr("Paired") : tr("Not Paired"));
 
+  sunnylinkUploaderEnabledBtn->setEnabled(max_current_sponsor_rule.roleTier == SponsorTier::Guardian && is_sunnylink_enabled);
+
   if (!is_sunnylink_enabled) {
     sunnylinkEnabledBtn->setValue("");
     sponsorBtn->setValue("");

selfdrive/ui/sunnypilot/qt/offroad/settings/sunnylink_panel.h

@@ -49,6 +49,7 @@ private:
   QString sunnylinkBtnDescription;
   PushButtonSP *restoreSettings;
   PushButtonSP *backupSettings;
+  ParamControlSP * sunnylinkUploaderEnabledBtn;
 
   void stopSunnylink() const;
   void startSunnylink() const;

sunnypilot/sunnylink/athena/sunnylinkd.py

@@ -15,7 +15,7 @@ from openpilot.common.params import Params
 from openpilot.common.realtime import set_core_affinity
 from openpilot.common.swaglog import cloudlog
 from openpilot.system.athena.athenad import ws_send, jsonrpc_handler, \
-  recv_queue, UploadQueueCache, upload_queue, cur_upload_items, backoff, ws_manage, log_handler, start_local_proxy_shim
+  recv_queue, UploadQueueCache, upload_queue, cur_upload_items, backoff, ws_manage, log_handler, start_local_proxy_shim, upload_handler
 from websocket import (ABNF, WebSocket, WebSocketException, WebSocketTimeoutException,
                        create_connection)
 
@@ -47,7 +47,7 @@ def handle_long_poll(ws: WebSocket, exit_event: threading.Event | None) -> None:
               threading.Thread(target=ws_send, args=(ws, end_event), name='ws_send'),
               threading.Thread(target=ws_ping, args=(ws, end_event), name='ws_ping'),
               threading.Thread(target=ws_queue, args=(end_event,), name='ws_queue'),
-              # threading.Thread(target=upload_handler, args=(end_event,), name='upload_handler'),
+              threading.Thread(target=upload_handler, args=(end_event,), name='upload_handler'),
               # threading.Thread(target=sunny_log_handler, args=(end_event, comma_prime_cellular_end_event), name='log_handler'),
               # threading.Thread(target=stat_handler, args=(end_event,), name='stat_handler'),
             ] + [

sunnypilot/sunnylink/uploader.py

@@ -1,33 +1,33 @@
 #!/usr/bin/env python3
-import bz2
-import datetime
-import io
 import json
 import os
 import random
+import requests
 import threading
 import time
 import traceback
+import datetime
 from collections.abc import Iterator
-from typing import BinaryIO
 
-import requests
+from cereal import log
+import cereal.messaging as messaging
+from sunnypilot.sunnylink.api import SunnylinkApi
+from openpilot.common.file_helpers import get_upload_stream
 from openpilot.common.params import Params
 from openpilot.common.realtime import set_core_affinity
-from openpilot.common.swaglog import cloudlog
 from openpilot.system.hardware.hw import Paths
 from openpilot.system.loggerd.xattr_cache import getxattr, setxattr
-
-import cereal.messaging as messaging
-from cereal import log
-from sunnypilot.sunnylink.api import SunnylinkApi
+from openpilot.common.swaglog import cloudlog
 
 NetworkType = log.DeviceState.NetworkType
 UPLOAD_ATTR_NAME = 'user.sunny.upload'
-
 UPLOAD_ATTR_VALUE = b'1'
 
-UPLOAD_QLOG_QCAM_MAX_SIZE = 5 * 1e6  # MB
+MAX_UPLOAD_SIZES = {
+  "qlog": 25*1e6,  # can't be too restrictive here since we use qlogs to find
+  # bugs, including ones that can cause massive log sizes
+  "qcam": 5*1e6,
+}
 
 allow_sleep = bool(os.getenv("UPLOADER_SLEEP", "1"))
 force_wifi = os.getenv("FORCEWIFI") is not None
@@ -50,7 +50,6 @@ def get_directory_sort(d: str) -> list[str]:
   o = ["0", ] if d.startswith("2024-") else ["1", ]
   return o + [s.rjust(10, '0') for s in d.rsplit('--', 1)]
 
-
 def listdir_by_creation(d: str) -> list[str]:
   if not os.path.isdir(d):
     return []
@@ -63,7 +62,6 @@ def listdir_by_creation(d: str) -> list[str]:
     cloudlog.exception("listdir_by_creation failed")
     return []
 
-
 def clear_locks(root: str) -> None:
   for logdir in os.listdir(root):
     path = os.path.join(root, logdir)
@@ -87,11 +85,11 @@ class Uploader:
     self.last_filename = ""
 
     self.immediate_folders = ["crash/", "boot/"]
-    self.immediate_priority = {"qlog": 0, "qlog.bz2": 0, "qcamera.ts": 1}
+    self.immediate_priority = {"qlog": 0, "qlog.zst": 0, "qcamera.ts": 1}
 
   def list_upload_files(self, metered: bool) -> Iterator[tuple[str, str, str]]:
     r = self.params.get("AthenadRecentlyViewedRoutes")
-    requested_routes = [] if r is None else r.split(",")
+    requested_routes = [] if r is None else [route for route in r.split(",") if route]
 
     for logdir in listdir_by_creation(self.root):
       path = os.path.join(self.root, logdir)
@@ -135,11 +133,11 @@ class Uploader:
       if any(f in fn for f in self.immediate_folders):
         return name, key, fn
 
-    return next(
-      ((name, key, fn)
-       for name, key, fn in upload_files if name in self.immediate_priority),
-      None,
-    )
+    for name, key, fn in upload_files:
+      if name in self.immediate_priority:
+        return name, key, fn
+
+    return None
 
   def do_upload(self, key: str, fn: str):
     url_resp = self.api.get(
@@ -159,15 +157,15 @@ class Uploader:
     if fake_upload:
       return FakeResponse()
 
-    with open(fn, "rb") as f:
-      data: BinaryIO
-      if key.endswith('.bz2') and not fn.endswith('.bz2'):
-        compressed = bz2.compress(f.read())
-        data = io.BytesIO(compressed)
-      else:
-        data = f
-
-      return requests.put(url, data=data, headers=headers, timeout=10)
+    stream = None
+    try:
+      compress = key.endswith('.zst') and not fn.endswith('.zst')
+      stream, _ = get_upload_stream(fn, compress)
+      response = requests.put(url, data=stream, headers=headers, timeout=10)
+      return response
+    finally:
+      if stream:
+        stream.close()
 
   def upload(self, name: str, key: str, fn: str, network_type: int, metered: bool) -> bool:
     try:
@@ -181,7 +179,7 @@ class Uploader:
     if sz == 0:
       # tag files of 0 size as uploaded
       success = True
-    elif name in self.immediate_priority and sz > UPLOAD_QLOG_QCAM_MAX_SIZE:
+    elif name in MAX_UPLOAD_SIZES and sz > MAX_UPLOAD_SIZES[name]:
       cloudlog.event("uploader_too_large", key=key, fn=fn, sz=sz)
       success = True
     else:
@@ -222,6 +220,7 @@ class Uploader:
 
     return success
 
+
   def step(self, network_type: int, metered: bool) -> bool | None:
     d = self.next_file_to_upload(metered)
     if d is None:
@@ -230,8 +229,8 @@ class Uploader:
     name, key, fn = d
 
     # qlogs and bootlogs need to be compressed before uploading
-    if key.endswith(('qlog', 'rlog')) or (key.startswith('boot/') and not key.endswith('.bz2')):
-      key += ".bz2"
+    if key.endswith(('qlog', 'rlog')) or (key.startswith('boot/') and not key.endswith('.zst')):
+      key += ".zst"
 
     return self.upload(name, key, fn, network_type, metered)
 

system/manager/process_config.py

@@ -174,7 +174,7 @@ procs += [
 if os.path.exists("./github_runner.sh"):
   procs += [NativeProcess("github_runner_start", "system/manager", ["./github_runner.sh", "start"], and_(only_offroad, use_github_runner), sigkill=False)]
 
-if os.path.exists("../sunnypilot/sunnylink/uploader.py"):
+if os.path.exists("../../sunnypilot/sunnylink/uploader.py"):
   procs += [PythonProcess("sunnylink_uploader", "sunnypilot.sunnylink.uploader", use_sunnylink_uploader_shim)]
 
 managed_processes = {p.name: p for p in procs}