ty

.github/workflows/sunnypilot-build-model.yaml

@@ -103,23 +103,21 @@ jobs:
       - run: |
           cd ${{ github.workspace }}/openpilot/openpilot
           if [ "${{ inputs.target_hardware }}" != "chestnut" ]; then
-            git lfs pull -X "**/selfdrive/modeld/models/big_*.onnx,**/selfdrive/modeld/models/dmonitoring_*.onnx,**/selfdrive/modeld/models/big_*.pkl,**/selfdrive/modeld/models/dmonitoring_*.pkl"
-            rm -f selfdrive/modeld/models/big_*.onnx selfdrive/modeld/models/dmonitoring_*.onnx selfdrive/modeld/models/big_*.pkl selfdrive/modeld/models/dmonitoring_*.pkl
+            git lfs pull -X "**/selfdrive/modeld/models/big_*.onnx,**/selfdrive/modeld/models/dmonitoring_*.onnx"
+            rm -f selfdrive/modeld/models/big_*.onnx selfdrive/modeld/models/dmonitoring_*.onnx
           else
-            git lfs pull -I "**/selfdrive/modeld/models/big_*.onnx,**/selfdrive/modeld/models/big_*.pkl" -X ""
-            find selfdrive/modeld/models -type f \( -name "*.onnx" -o -name "*.pkl" \) ! -name "big_*.onnx" ! -name "big_*.pkl" -delete
+            git lfs pull -I "**/selfdrive/modeld/models/big_*.onnx" -X ""
+            find selfdrive/modeld/models -name "*.onnx" ! -name "big_*.onnx" -delete
           fi
-          if grep -lIF "version https://git-lfs.github.com/spec/v1" selfdrive/modeld/models/*.onnx selfdrive/modeld/models/*.pkl 2>/dev/null; then
-            echo "::error::the ONNX or PKL files above are still LFS pointers, not real models"
+          if grep -lIF "version https://git-lfs.github.com/spec/v1" selfdrive/modeld/models/*.onnx; then
+            echo "::error::the ONNX files above are still LFS pointers, not real models"
             exit 1
           fi
       - name: 'Upload Artifact'
         uses: actions/upload-artifact@v4
         with:
           name: models-${{ env.REF }}${{ inputs.artifact_suffix  }}
-          path: |
-            ${{ github.workspace }}/openpilot/openpilot/selfdrive/modeld/models/*.onnx
-            ${{ github.workspace }}/openpilot/openpilot/selfdrive/modeld/models/*.pkl
+          path: ${{ github.workspace }}/openpilot/openpilot/selfdrive/modeld/models/*.onnx
           if-no-files-found: error
 
   build_model:
@@ -198,75 +196,64 @@ jobs:
             OUTPUT_PKL="${{ env.MODELS_DIR }}/driving_tinygrad.pkl"
           fi
 
-          NATIVE_PKL=$(find "${{ env.MODELS_DIR }}" -maxdepth 1 -name "*.pkl" -print -quit)
+          # Generate metadata for all ONNX files
+          find "${{ env.MODELS_DIR }}" -maxdepth 1 -name '*.onnx' | while IFS= read -r onnx_file; do
+            echo "Generating metadata: $onnx_file"
+            env ${TG_FLAGS_QCOM} python3 "${{ env.MODELS_DIR }}/../get_model_metadata.py" "$onnx_file" || true
+          done
 
-          if [ -n "$NATIVE_PKL" ]; then
-            echo "Found native precompiled pkl: $NATIVE_PKL"
-            if [ "$NATIVE_PKL" != "$OUTPUT_PKL" ]; then
-              mv "$NATIVE_PKL" "$OUTPUT_PKL"
+          # Detect model type and build compile args
+          VISION_ONNX=""
+          for f in "${{ env.MODELS_DIR }}/driving_vision.onnx" "${{ env.MODELS_DIR }}/big_driving_vision.onnx"; do
+            [ -f "$f" ] && VISION_ONNX="$f" && break
+          done
+
+          POLICY_ONNX=""
+          for f in "${{ env.MODELS_DIR }}/driving_policy.onnx" "${{ env.MODELS_DIR }}/big_driving_policy.onnx"; do
+            [ -f "$f" ] && POLICY_ONNX="$f" && break
+          done
+
+          OFF_POLICY_ONNX=""
+          for f in "${{ env.MODELS_DIR }}/driving_off_policy.onnx" "${{ env.MODELS_DIR }}/big_driving_off_policy.onnx"; do
+            [ -f "$f" ] && OFF_POLICY_ONNX="$f" && break
+          done
+
+          ON_POLICY_ONNX=""
+          for f in "${{ env.MODELS_DIR }}/driving_on_policy.onnx" "${{ env.MODELS_DIR }}/big_driving_on_policy.onnx"; do
+            [ -f "$f" ] && ON_POLICY_ONNX="$f" && break
+          done
+
+          SUPERCOMBO_ONNX=""
+          for f in "${{ env.MODELS_DIR }}/supercombo.onnx" "${{ env.MODELS_DIR }}/driving_supercombo.onnx" "${{ env.MODELS_DIR }}/big_supercombo.onnx" "${{ env.MODELS_DIR }}/big_driving_supercombo.onnx"; do
+            [ -f "$f" ] && SUPERCOMBO_ONNX="$f" && break
+          done
+
+          MODEL_TYPE="" ONNX_ARGS="" OUTPUT_NAME=""
+          if [ -f "$VISION_ONNX" ]; then
+            ONNX_ARGS="--vision-onnx $VISION_ONNX"
+            if [ -f "$ON_POLICY_ONNX" ] && [ -f "$OFF_POLICY_ONNX" ]; then
+              MODEL_TYPE=vision_multi_policy
+              ONNX_ARGS="$ONNX_ARGS --off-policy-onnx $OFF_POLICY_ONNX --on-policy-onnx $ON_POLICY_ONNX"
+            elif [ -f "$OFF_POLICY_ONNX" ] && [ -f "$POLICY_ONNX" ]; then
+              MODEL_TYPE=vision_multi_policy
+              ONNX_ARGS="$ONNX_ARGS --policy-onnx $POLICY_ONNX --off-policy-onnx $OFF_POLICY_ONNX"
+            elif [ -f "$POLICY_ONNX" ]; then
+              MODEL_TYPE=vision_policy
+              ONNX_ARGS="$ONNX_ARGS --policy-onnx $POLICY_ONNX"
             fi
-            echo "Chunking pkl"
-            python3 -c "from openpilot.common.file_chunker import chunk_file, get_chunk_targets; import os; p='$OUTPUT_PKL'; chunk_file(p, get_chunk_targets(p, os.path.getsize(p)))"
-          else
-            # Generate metadata for all ONNX files
-            find "${{ env.MODELS_DIR }}" -maxdepth 1 -name '*.onnx' | while IFS= read -r onnx_file; do
-              echo "Generating metadata: $onnx_file"
-              env ${TG_FLAGS_QCOM} python3 "${{ env.MODELS_DIR }}/../get_model_metadata.py" "$onnx_file" || true
-            done
+          elif [ -f "$SUPERCOMBO_ONNX" ]; then
+            MODEL_TYPE=supercombo
+            ONNX_ARGS="--supercombo-onnx $SUPERCOMBO_ONNX"
+          fi
 
-            # Detect model type and build compile args
-            VISION_ONNX=""
-            for f in "${{ env.MODELS_DIR }}/driving_vision.onnx" "${{ env.MODELS_DIR }}/big_driving_vision.onnx"; do
-              [ -f "$f" ] && VISION_ONNX="$f" && break
-            done
-
-            POLICY_ONNX=""
-            for f in "${{ env.MODELS_DIR }}/driving_policy.onnx" "${{ env.MODELS_DIR }}/big_driving_policy.onnx"; do
-              [ -f "$f" ] && POLICY_ONNX="$f" && break
-            done
-
-            OFF_POLICY_ONNX=""
-            for f in "${{ env.MODELS_DIR }}/driving_off_policy.onnx" "${{ env.MODELS_DIR }}/big_driving_off_policy.onnx"; do
-              [ -f "$f" ] && OFF_POLICY_ONNX="$f" && break
-            done
-
-            ON_POLICY_ONNX=""
-            for f in "${{ env.MODELS_DIR }}/driving_on_policy.onnx" "${{ env.MODELS_DIR }}/big_driving_on_policy.onnx"; do
-              [ -f "$f" ] && ON_POLICY_ONNX="$f" && break
-            done
-
-            SUPERCOMBO_ONNX=""
-            for f in "${{ env.MODELS_DIR }}/supercombo.onnx" "${{ env.MODELS_DIR }}/driving_supercombo.onnx" "${{ env.MODELS_DIR }}/big_supercombo.onnx" "${{ env.MODELS_DIR }}/big_driving_supercombo.onnx"; do
-              [ -f "$f" ] && SUPERCOMBO_ONNX="$f" && break
-            done
-
-            MODEL_TYPE="" ONNX_ARGS="" OUTPUT_NAME=""
-            if [ -f "$VISION_ONNX" ]; then
-              ONNX_ARGS="--vision-onnx $VISION_ONNX"
-              if [ -f "$ON_POLICY_ONNX" ] && [ -f "$OFF_POLICY_ONNX" ]; then
-                MODEL_TYPE=vision_multi_policy
-                ONNX_ARGS="$ONNX_ARGS --off-policy-onnx $OFF_POLICY_ONNX --on-policy-onnx $ON_POLICY_ONNX"
-              elif [ -f "$OFF_POLICY_ONNX" ] && [ -f "$POLICY_ONNX" ]; then
-                MODEL_TYPE=vision_multi_policy
-                ONNX_ARGS="$ONNX_ARGS --policy-onnx $POLICY_ONNX --off-policy-onnx $OFF_POLICY_ONNX"
-              elif [ -f "$POLICY_ONNX" ]; then
-                MODEL_TYPE=vision_policy
-                ONNX_ARGS="$ONNX_ARGS --policy-onnx $POLICY_ONNX"
-              fi
-            elif [ -f "$SUPERCOMBO_ONNX" ]; then
-              MODEL_TYPE=supercombo
-              ONNX_ARGS="--supercombo-onnx $SUPERCOMBO_ONNX"
-            fi
-
-            if [ -n "$MODEL_TYPE" ]; then
-              echo "Detected: $MODEL_TYPE -> $OUTPUT_PKL"
-              env ${TG_FLAGS} python3 "$COMPILE_MODELD" \
-                --model-type $MODEL_TYPE \
-                --model-size $MODEL_SIZE \
-                --camera-resolutions $CAMERA_RES \
-                $ONNX_ARGS \
-                --output "$OUTPUT_PKL"
-            fi
+          if [ -n "$MODEL_TYPE" ]; then
+            echo "Detected: $MODEL_TYPE -> $OUTPUT_PKL"
+            env ${TG_FLAGS} python3 "$COMPILE_MODELD" \
+              --model-type $MODEL_TYPE \
+              --model-size $MODEL_SIZE \
+              --camera-resolutions $CAMERA_RES \
+              $ONNX_ARGS \
+              --output "$OUTPUT_PKL"
           fi
 
       - name: Prepare Output

.github/workflows/test_models.yml

@@ -1,79 +0,0 @@
-name: Test Models Compatibility With Tinygrad Changes
-on:
-  pull_request:
-    paths:
-      - 'tinygrad_repo'
-  workflow_dispatch:
-
-jobs:
-  generate-matrix:
-    runs-on: ubuntu-latest
-    outputs:
-      models: ${{ steps.set-matrix.outputs.models }}
-    steps:
-      - uses: actions/checkout@v4
-      - name: Fetch and Parse json
-        id: set-matrix
-        run: |
-          python3 -c '
-          import json, urllib.request, os, re
-
-          with open("openpilot/sunnypilot/models/fetcher.py", "r") as f:
-            urls = re.findall(r"MODEL_URL(?:_CHESTNUT)?\s*=\s*[\"'"'"']([^\"'"'"']+)[\"'"'"']", f.read())
-
-          artifacts = []
-          for url in urls:
-            data = json.loads(urllib.request.urlopen(url).read())
-            for bundle in data.get("bundles", []):
-              for model in bundle.get("models", []):
-                if "artifact" in model:
-                  artifacts.append(model["artifact"])
-
-          with open(os.environ["GITHUB_OUTPUT"], "a") as f:
-            f.write(f"models={json.dumps(artifacts)}\n")
-          '
-
-  test-model:
-    name: Test ${{ matrix.artifact.file_name }}
-    needs: generate-matrix
-    runs-on: ubuntu-latest
-    container: ghcr.io/commaai/openpilot-base:latest
-    strategy:
-      fail-fast: false
-      matrix:
-        artifact: ${{ fromJson(needs.generate-matrix.outputs.models) }}
-    steps:
-      - uses: actions/checkout@v4
-        with:
-          submodules: true
-
-      - name: Download Model Chunks in Parallel
-        run: |
-          mkdir -p /tmp/model_chunks
-          echo '${{ toJson(matrix.artifact.chunks) }}' > chunks.json
-
-          BASE_URL="${{ matrix.artifact.download_uri.url }}"
-          export BASE_DIR=$(dirname "$BASE_URL")
-
-          python3 -c '
-          import json, os
-          with open("chunks.json") as f:
-            chunks = json.load(f)
-          manifest_path = f"/tmp/model_chunks/${{ matrix.artifact.file_name }}.chunkmanifest"
-          with open(manifest_path, "w") as f:
-            f.write(str(len(chunks)))
-          base_dir = os.environ["BASE_DIR"]
-          with open("/tmp/curl_config.txt", "w") as f:
-            for c in chunks:
-              fn = c["file_name"]
-              f.write(f"url = \"{base_dir}/{fn}\"\noutput = \"/tmp/model_chunks/{fn}\"\n")
-          '
-          curl -Z --parallel-immediate --parallel-max 16 -s -S -f -L -K /tmp/curl_config.txt
-
-      - name: Run Model Compatibility Test
-        env:
-          MODEL_BASE_NAME: ${{ matrix.artifact.file_name }}
-          MODEL_CHUNK_DIR: "/tmp/model_chunks"
-          PYTHONPATH: ".:./tinygrad_repo"
-        run: |
-          python3 -m pytest openpilot/sunnypilot/modeld_v2/tests/test_models.py

openpilot/common/params_keys.h

@@ -227,6 +227,11 @@ inline static std::unordered_map<std::string, ParamKeyAttributes> keys = {
     {"SunnylinkEnabled", {PERSISTENT, BOOL, "1"}},
     {"SunnylinkTempFault", {CLEAR_ON_MANAGER_START | CLEAR_ON_OFFROAD_TRANSITION, BOOL, "0"}},
 
+    {"SunnylinkLocalApps", {PERSISTENT, JSON}},
+    {"SunnylinkLocalPairingCode", {CLEAR_ON_MANAGER_START, JSON}},
+    {"SunnylinkLocalDiscoveredApp", {CLEAR_ON_MANAGER_START, JSON}},
+    {"SunnylinkLocalPairingRequest", {CLEAR_ON_MANAGER_START, BOOL}},
+
     // Backup Manager params
     {"BackupManager_CreateBackup", {PERSISTENT, BOOL}},
     {"BackupManager_RestoreVersion", {PERSISTENT, STRING}},

openpilot/selfdrive/modeld/modeld.py

@@ -1,13 +1,10 @@
 #!/usr/bin/env python3
 from collections.abc import Callable
 import ctypes
-import codecs
 from functools import cached_property
 import os
-import pickle
 os.environ['GMMU'] = '0' # for chestnut fast loading, noop for qcom
 from tinygrad.device import Device
-from tinygrad_repo.examples.openpilot.helpers import allocate_inputs, load_pickle
 import usb1
 import struct
 import threading
@@ -26,15 +23,21 @@ from openpilot.common.params import Params
 from openpilot.common.filter_simple import FirstOrderFilter
 from openpilot.common.realtime import config_realtime_process, DT_MDL
 from openpilot.common.transformations.camera import DEVICE_CAMERAS
+from openpilot.system.camerad.cameras.nv12_info import get_nv12_info
 from openpilot.common.transformations.model import get_warp_matrix
 from openpilot.selfdrive.controls.lib.desire_helper import DesireHelper
 from openpilot.selfdrive.controls.lib.drive_helpers import get_accel_from_plan, should_stop, smooth_value, get_curvature_from_plan
 from openpilot.selfdrive.modeld.parse_model_outputs import Parser
+from openpilot.selfdrive.modeld.compile_modeld import make_input_queues, nv12_copy_size, MODELD_INPUTS
 from openpilot.selfdrive.modeld.fill_model_msg import fill_model_msg, fill_driving_model_data, fill_pose_msg, PublishState
 from openpilot.common.file_chunker import open_file_chunked
 from openpilot.common.hardware.usb import CHESTNUT_USB_IDS
 from openpilot.selfdrive.modeld.constants import ModelConstants, Plan
-from openpilot.selfdrive.modeld.helpers import chestnut_present, chestnut_compiled, modeld_pkl_path
+from openpilot.selfdrive.modeld.helpers import chestnut_present, chestnut_compiled, chestnut_ready, modeld_pkl_path, load_oob
+
+from openpilot.sunnypilot.livedelay.helpers import get_lat_delay
+from openpilot.sunnypilot.modeld_v2.modeld_base import ModelStateBase
+from openpilot.sunnypilot.selfdrive.controls.lib.relc import RoadEdgeLaneChangeController
 
 SEND_RAW_PRED = os.getenv('SEND_RAW_PRED')
 
@@ -172,30 +175,28 @@ class FrameMeta:
       self.frame_id, self.timestamp_sof, self.timestamp_eof = vipc.frame_id, vipc.timestamp_sof, vipc.timestamp_eof
 
 
-class ModelState:
+class ModelState(ModelStateBase):
   prev_desire: np.ndarray  # for tracking the rising edge of the pulse
 
   def __init__(self, cam_w: int, cam_h: int, chestnut: bool):
-    jits = load_pickle(open_file_chunked(modeld_pkl_path(chestnut)), out_of_band=True)
-    variant = jits['variants'][f'{cam_w}x{cam_h}']
+    ModelStateBase.__init__(self)
+    jits = load_oob(open_file_chunked(modeld_pkl_path(chestnut)))
+    input_devices = jits['input_devices']
+    self.model_device = input_devices['model']
     metadata = jits['metadata']
     self.input_shapes = metadata['input_shapes']
     self.vision_input_names = [k for k in self.input_shapes if 'img' in k]
-    self.output_slices = pickle.loads(codecs.decode(metadata['metadata']['output_slices'].encode(), 'base64'))
+    self.output_slices = metadata['output_slices']
 
     self.prev_desire = np.zeros(ModelConstants.DESIRE_LEN, dtype=np.float32)
     self.chestnut = chestnut
 
-    self.input_specs = variant['input_specs']
-    self.packed_specs = variant['packed_specs']
-    self.reset_inputs()
+    self.frame_skip = ModelConstants.MODEL_RUN_FREQ // ModelConstants.MODEL_CONTEXT_FREQ
+    self.frame_copy_size = nv12_copy_size(*get_nv12_info(cam_w, cam_h)[:3])
+    self.input_queues, self.npy, self.frame_views = make_input_queues(
+      self.input_shapes, self.frame_skip, device=self.model_device, frame_copy_size=self.frame_copy_size)
     self.parser = Parser()
-    self.run_model = variant['run']
-
-  def reset_inputs(self) -> None:
-    self.input_queues, views = allocate_inputs(self.input_specs, self.packed_specs)
-    self.frame_views = {name: views[name] for name in self.vision_input_names}
-    self.npy = {name: views[name] for name in self.packed_specs if name not in self.frame_views}
+    self.run_model = jits['run_model'][(cam_w,cam_h)]
 
   def slice_outputs(self, model_outputs: np.ndarray, output_slices: dict[str, slice]) -> dict[str, np.ndarray]:
     parsed_model_outputs = {k: model_outputs[np.newaxis, v] for k,v in output_slices.items()}
@@ -204,48 +205,62 @@ class ModelState:
   def run(self, bufs: dict[str, VisionBuf], transforms: dict[str, np.ndarray],
           inputs: dict[str, np.ndarray], after_enqueue: Callable[[], None] | None = None) -> dict[str, np.ndarray]:
     for key, buf in bufs.items():
-      np.copyto(self.frame_views[key], np.frombuffer(buf.data, dtype=np.uint8, count=self.frame_views[key].size))
+      np.copyto(self.frame_views[key], np.frombuffer(buf.data, dtype=np.uint8, count=self.frame_copy_size))
 
     # Model decides when action is completed, so desire input is just a pulse triggered on rising edge
     inputs['desire_pulse'][0] = 0
-    self.npy['desire'].flat[:] = np.where(inputs['desire_pulse'] - self.prev_desire > .99, inputs['desire_pulse'], 0)
+    self.npy['desire'][:] = np.where(inputs['desire_pulse'] - self.prev_desire > .99, inputs['desire_pulse'], 0)
     self.prev_desire[:] = inputs['desire_pulse']
     self.npy['traffic_convention'][:] = inputs['traffic_convention']
     self.npy['action_t'][:] = inputs['action_t']
     self.npy['tfm'][:,:] = transforms['img'][:,:]
     self.npy['big_tfm'][:,:] = transforms['big_img'][:,:]
 
-    outs = self.run_model(**self.input_queues)
+    outs, = self.run_model(**{k: self.input_queues[k] for k in MODELD_INPUTS})
     if after_enqueue is not None:
       after_enqueue()
     model_output = outs.numpy()[0]
     if self.chestnut and not np.all(np.isfinite(model_output)):
       raise RuntimeError("model output not finite")
     outputs_dict = self.parser.parse_outputs(self.slice_outputs(model_output, self.output_slices))
-    self.npy['prev_feat'].flat[:] = model_output[self.output_slices['hidden_state']]
+    self.npy['prev_feat'][:] = model_output[self.output_slices['hidden_state']]
 
     if SEND_RAW_PRED:
       outputs_dict['raw_pred'] = model_output.copy()
     return outputs_dict
 
   def warmup(self) -> None:
-    dummy_frames = {k: np.zeros_like(v) for k, v in self.frame_views.items()}
+    dummy_frames = {k: np.zeros(self.frame_copy_size, dtype=np.uint8) for k in self.vision_input_names}
     eye = np.eye(3, dtype=np.float32)
     dims = {'desire_pulse': ModelConstants.DESIRE_LEN, 'traffic_convention': 2, 'action_t': 2}
     self.run(dummy_frames, dict.fromkeys(self.vision_input_names, eye), {k: np.zeros(v, dtype=np.float32) for k, v in dims.items()})
-    self.reset_inputs()
+    self.input_queues, self.npy, self.frame_views = make_input_queues(
+      self.input_shapes, self.frame_skip, device=self.model_device, frame_copy_size=self.frame_copy_size)
     self.prev_desire[:] = 0
 
 
 def main(demo=False):
   cloudlog.warning("modeld init")
 
-  CHESTNUT = chestnut_present() and chestnut_compiled()
+  chestnut_available = chestnut_present() and chestnut_compiled()
+  CHESTNUT = False
+  if chestnut_available:
+    poller = messaging.Poller()
+    sock = messaging.sub_sock("chestnutState", poller=poller, conflate=True)
+    deadline = time.monotonic() + 4. / SERVICE_LIST['deviceState'].frequency
+    while not CHESTNUT and (remaining := deadline - time.monotonic()) > 0.:
+      if not poller.poll(round(remaining * 1000)):
+        break
+      msg = messaging.recv_one_or_none(sock)
+      CHESTNUT = msg is not None and msg.valid and chestnut_ready(msg.chestnutState)
   if CHESTNUT:
     os.environ['HCQDEV_WAIT_TIMEOUT_MS'] = '3000'
   params = Params()
   params.put_bool("ChestnutLoading", CHESTNUT)
-  params.remove("ChestnutActive")
+  if chestnut_available and not CHESTNUT:
+    params.put_bool("ChestnutActive", False)
+  else:
+    params.remove("ChestnutActive")
 
   config_realtime_process(7, 54)
 
@@ -289,16 +304,21 @@ def main(demo=False):
     loader.start()
     loader.join(BIG_MODEL_TIMEOUT)
     model = big_model
+    if model is None:
+      params.put_bool("ChestnutModelError", True)
     params.put_bool("ChestnutActive", model is not None)
+    if model is not None:
+      params.remove("ChestnutModelError")
 
   small_model = ModelState(vipc_client_main.width, vipc_client_main.height, False) if model is None or CHESTNUT else None
   if model is None:
     model = small_model
   params.put_bool("ChestnutLoading", False)
+  assert model is not None
   cloudlog.warning(f"models loaded in {time.monotonic() - st:.1f}s, modeld starting")
 
   # messaging
-  pub_socks = ["modelV2", "drivingModelData", "cameraOdometry"] + (["chestnutState"] if CHESTNUT else [])
+  pub_socks = ["modelV2", "drivingModelData", "cameraOdometry", "modelDataV2SP"] + (["chestnutState"] if CHESTNUT else [])
   pm = PubMaster(pub_socks)
   sm = SubMaster(["deviceState", "carState", "narrowRoadCameraState", "extrinsicsCalibration", "driverMonitoringState", "carControl", "lateralDelay"])
 
@@ -331,6 +351,7 @@ def main(demo=False):
   prev_action = log.ModelDataV2.Action()
 
   DH = DesireHelper()
+  RELC = RoadEdgeLaneChangeController()
 
   while True:
     # Keep receiving frames until we are at least 1 frame ahead of previous extra frame
@@ -370,6 +391,7 @@ def main(demo=False):
     is_rhd = sm["driverMonitoringState"].isRHD
     frame_id = sm["narrowRoadCameraState"].frameId
     v_ego = max(sm["carState"].vEgo, 0.)
+    model.lat_delay = get_lat_delay(params, sm["lateralDelay"].lateralDelay)
     lat_delay = sm["lateralDelay"].lateralDelay + LAT_SMOOTH_SECONDS
     if sm.updated["extrinsicsCalibration"] and sm.seen['narrowRoadCameraState'] and sm.seen['deviceState']:
       device_from_calib_euler = np.array(sm["extrinsicsCalibration"].rpyCalib, dtype=np.float32)
@@ -420,7 +442,9 @@ def main(demo=False):
         raise
       # fallback to small model
       cloudlog.exception("big model failed, fall back to small")
+      params.put_bool("ChestnutModelError", True)
       params.put_bool("ChestnutActive", False)
+      assert small_model is not None
       model = small_model
       if chestnut_state is not None:
         chestnut_state.big = False
@@ -445,15 +469,19 @@ def main(demo=False):
       l_lane_change_prob = desire_state[log.Desire.laneChangeLeft]
       r_lane_change_prob = desire_state[log.Desire.laneChangeRight]
       lane_change_prob = l_lane_change_prob + r_lane_change_prob
-      DH.update(sm['carState'], sm['carControl'].latActive, lane_change_prob)
+      mdv2sp_send = messaging.new_message('modelDataV2SP')
+      left_edge, right_edge = RELC.update_and_fill(modelv2_send.modelV2, mdv2sp_send.modelDataV2SP, v_ego)
+      DH.update(sm['carState'], sm['carControl'].latActive, lane_change_prob, left_edge, right_edge)
       modelv2_send.modelV2.meta.laneChangeState = DH.lane_change_state
       modelv2_send.modelV2.meta.laneChangeDirection = DH.lane_change_direction
+      mdv2sp_send.modelDataV2SP.laneTurnDirection = DH.lane_turn_direction
 
       fill_driving_model_data(drivingdata_send, modelv2_send)
       fill_pose_msg(posenet_send, model_output, meta_main.frame_id, vipc_dropped_frames, meta_main.timestamp_eof, extrinsics_calibration_seen)
       pm.send('modelV2', modelv2_send)
       pm.send('drivingModelData', drivingdata_send)
       pm.send('cameraOdometry', posenet_send)
+      pm.send('modelDataV2SP', mdv2sp_send)
     last_vipc_frame_id = meta_main.frame_id
 
 if __name__ == "__main__":

openpilot/selfdrive/ui/sunnypilot/layouts/settings/sunnylink.py

@@ -5,22 +5,42 @@ This file is part of sunnypilot and is licensed under the MIT License.
 See the LICENSE.md file in the root directory for more details.
 """
 import pyray as rl
+from functools import partial
 from openpilot.cereal import custom
+from openpilot.common.version import sunnylink_consent_version
 from openpilot.selfdrive.ui.sunnypilot.layouts.onboarding import SunnylinkConsentPage
 from openpilot.selfdrive.ui.ui_state import ui_state
 from openpilot.sunnypilot.sunnylink.api import UNREGISTERED_SUNNYLINK_DONGLE_ID
+from openpilot.sunnypilot.sunnylink.athena.local_discovery import latest_discovered_app
+from openpilot.sunnypilot.sunnylink.athena.local_pairing import (
+  LocalApp,
+  arm_pairing,
+  clear_pairing_request,
+  get_local_apps,
+  local_app_display_name,
+  pairing_requested,
+  read_pairing_code,
+  remove_local_app,
+)
 from openpilot.system.ui.lib.application import gui_app, FontWeight, TextAlignment, TextAlignmentVertical
 from openpilot.system.ui.lib.multilang import tr
-from openpilot.system.ui.sunnypilot.widgets.list_view import button_item_sp
-from openpilot.system.ui.sunnypilot.widgets.list_view import toggle_item_sp
+from openpilot.system.ui.lib.text_measure import measure_text_cached
+from openpilot.system.ui.lib.wrap_text import wrap_text
+from openpilot.system.ui.sunnypilot.widgets.list_view import ListItemSP, button_item_sp, toggle_item_sp
 from openpilot.system.ui.sunnypilot.widgets.sunnylink_pairing_dialog import SunnylinkPairingDialog
 from openpilot.system.ui.widgets import Widget, DialogResult
-from openpilot.system.ui.widgets.button import ButtonStyle, Button
+from openpilot.system.ui.widgets.button import ButtonStyle, Button, IconButton
 from openpilot.system.ui.widgets.confirm_dialog import alert_dialog, ConfirmDialog
 from openpilot.system.ui.widgets.label import UnifiedLabel
 from openpilot.system.ui.widgets.list_view import dual_button_item
+from openpilot.system.ui.widgets.network import NavButton
 from openpilot.system.ui.widgets.scroller_tici import Scroller, LineSeparator
-from openpilot.common.version import sunnylink_consent_version
+
+MAX_LOCAL_APPS = 4
+
+# Read-only value colors used by the local-mode rows.
+_LOCAL_DISCOVERED_COLOR = rl.Color(170, 170, 170, 255)  # grey: no app in sight
+_LOCAL_ACTIVE_COLOR = rl.Color(0, 255, 0, 255)          # green: discovered / pairing code
 
 
 class SunnylinkHeader(Widget):
@@ -192,6 +212,15 @@ class SunnylinkLayout(Widget):
     self._backup_btn.set_button_style(ButtonStyle.NORMAL)
     self._restore_btn.set_button_style(ButtonStyle.PRIMARY)
 
+    self._mobile_app_btn = button_item_sp(
+      title=tr("Sunnylink Local Connections"),
+      button_text=tr("CONFIGURE"),
+      description=tr("Manage the mobile app(s) connected over Wi-Fi: pair a new app ") +
+                  tr("or unpair existing ones."),
+      callback=self._open_local_apps,
+    )
+    self._mobile_app_btn.set_visible(lambda: self._sunnylink_enabled)
+
     items = [
       SunnylinkHeader(),
       LineSeparator(),
@@ -202,9 +231,11 @@ class SunnylinkLayout(Widget):
       LineSeparator(),
       self._pair_btn,
       LineSeparator(),
+      self._mobile_app_btn,
+      LineSeparator(),
       self._sunnylink_uploader_toggle,
       LineSeparator(),
-      self._sunnylink_backup_restore_buttons
+      self._sunnylink_backup_restore_buttons,
     ]
     return items
 
@@ -317,6 +348,8 @@ class SunnylinkLayout(Widget):
       gui_app.push_widget(sl_terms_dlg)
     else:
       ui_state.params.put_bool("SunnylinkEnabled", state)
+      if not state:
+        clear_pairing_request()
       self._update_description(state)
 
   def _update_description(self, state: bool):
@@ -352,6 +385,9 @@ class SunnylinkLayout(Widget):
     self._pair_btn.action_item.set_text(pair_btn_text)
     self._pair_btn.action_item.set_enabled(self._sunnylink_enabled)
 
+  def _open_local_apps(self):
+    gui_app.push_widget(SunnylinkLocalAppLayout())
+
   def _render(self, rect):
     self._scroller.render(rect)
 
@@ -364,3 +400,157 @@ class SunnylinkLayout(Widget):
   def hide_event(self):
     super().hide_event()
     ui_state.sunnylink_state.set_settings_open(False)
+
+
+class SunnylinkLocalAppLayout(Widget):
+
+  def __init__(self):
+    super().__init__()
+    self._local_apps_cache: list[LocalApp] = []
+
+    self._back_button = NavButton(tr("Back"))
+    self._back_button.set_click_callback(gui_app.pop_widget)
+
+    self._pair_app_btn = button_item_sp(
+      title=tr("Pair App"),
+      button_text=tr("PAIR"),
+      description=tr("Open a 5-minute pairing window and show the code to ") +
+                  tr("type into the app. Closing the dialog cancels pairing."),
+      callback=self._show_pairing_code_dialog,
+    )
+
+    self._local_app_rows: list[ListItemSP] = []
+    self._local_app_seps: list[LineSeparator] = []
+    for i in range(MAX_LOCAL_APPS):
+      row = button_item_sp(
+        title=lambda i=i: self._local_app_title(i),
+        button_text=tr("UNPAIR"),
+        description=lambda i=i: self._local_app_endpoint(i),
+        callback=partial(self._unpair_local_app, i),
+      )
+      sep = LineSeparator()
+      row.set_visible(lambda i=i: self._local_row_visible(i))
+      sep.set_visible(lambda i=i: self._local_row_visible(i))
+      self._local_app_rows.append(row)
+      self._local_app_seps.append(sep)
+
+    items = [self._pair_app_btn, LineSeparator()]
+    for row, sep in zip(self._local_app_rows, self._local_app_seps, strict=True):
+      items.extend((row, sep))
+    self._scroller = Scroller(items, line_separator=False, spacing=0)
+
+  def _local_row_visible(self, i: int) -> bool:
+    return i < len(self._local_apps_cache)
+
+  def _local_app_title(self, i: int) -> str:
+    if i >= len(self._local_apps_cache):
+      return ""
+    return local_app_display_name(self._local_apps_cache[i])
+
+  def _local_app_endpoint(self, i: int) -> str:
+    if i >= len(self._local_apps_cache):
+      return ""
+    return self._local_apps_cache[i].endpoint
+
+  def _show_pairing_code_dialog(self):
+    gui_app.push_widget(SunnylinkLocalPairingDialog())
+
+  def _unpair_local_app(self, index: int):
+    apps = self._local_apps_cache
+    if index >= len(apps):
+      return
+    app = apps[index]
+    name = local_app_display_name(app)
+
+    def on_confirm(_dialog_result: int):
+      remove_local_app(app.app_id)
+
+    dialog = ConfirmDialog(
+      text=tr("Unpair") + f" {name}? " + tr("You will need the pairing code again to reconnect it."),
+      confirm_text=tr("Unpair"),
+      callback=on_confirm,
+    )
+    gui_app.push_widget(dialog)
+
+  def _update_state(self):
+    super()._update_state()
+    self._local_apps_cache = get_local_apps()
+
+  def _render(self, rect):
+    self._back_button.set_position(self._rect.x, self._rect.y + 20)
+    self._back_button.render()
+    content_rect = rl.Rectangle(rect.x, rect.y + self._back_button.rect.height + 40,
+                                rect.width, rect.height - self._back_button.rect.height - 40)
+    self._scroller.render(content_rect)
+
+  def show_event(self):
+    super().show_event()
+    self._scroller.show_event()
+
+  def hide_event(self):
+    super().hide_event()
+    self._scroller.hide_event()
+
+
+class SunnylinkLocalPairingDialog(Widget):
+
+  def __init__(self):
+    super().__init__()
+    self._apps_before = len(get_local_apps())
+    arm_pairing()
+    self._close_btn = IconButton(gui_app.texture("icons/close.png", 80, 80))
+    self._close_btn.set_click_callback(self._cancel)
+
+  def _cancel(self):
+    clear_pairing_request()
+    gui_app.pop_widget()
+
+  def _update_state(self):
+    if len(get_local_apps()) > self._apps_before:
+      gui_app.pop_widget()  # paired — window already cleared
+    elif not pairing_requested():
+      gui_app.pop_widget()  # window expired
+
+  def _render(self, rect) -> int:
+    rl.clear_background(rl.Color(224, 224, 224, 255))
+
+    margin = 70
+    content_rect = rl.Rectangle(rect.x + margin, rect.y + margin,
+                                rect.width - 2 * margin, rect.height - 2 * margin)
+    y = content_rect.y
+
+    close_size = 80
+    pad = 20
+    close_rect = rl.Rectangle(content_rect.x - pad, y - pad, close_size + pad * 2, close_size + pad * 2)
+    self._close_btn.render(close_rect)
+    y += close_size + 40
+
+    title_font = gui_app.font(FontWeight.NORMAL)
+    title_wrapped = wrap_text(title_font, tr("Pair with mobile app"), 75, int(content_rect.width))
+    rl.draw_text_ex(title_font, "\n".join(title_wrapped), rl.Vector2(content_rect.x, y), 75, 0.0, rl.BLACK)
+    y += len(title_wrapped) * 75 + 40
+
+    code = read_pairing_code() or "—"
+    code_font = gui_app.font(FontWeight.BOLD)
+    code_size = measure_text_cached(code_font, code, 110)
+    rl.draw_text_ex(code_font, code, rl.Vector2(content_rect.x + (content_rect.width - code_size.x) / 2, y),
+                    110, 0.0, rl.BLACK)
+    y += 170
+
+    hint_font = gui_app.font(FontWeight.NORMAL)
+    hint_wrapped = wrap_text(hint_font, tr("Enter this code in the sunnylink app on your phone."), 45,
+                             int(content_rect.width))
+    rl.draw_text_ex(hint_font, "\n".join(hint_wrapped), rl.Vector2(content_rect.x, y), 45, 0.0, rl.BLACK)
+    y += len(hint_wrapped) * 45 + 30
+
+    discovered = latest_discovered_app()
+    if discovered is not None:
+      endpoint, age = discovered
+      status = endpoint if age < 2 else f"{endpoint} ({age}s)"
+      color = _LOCAL_ACTIVE_COLOR
+    else:
+      status = tr("Waiting for the app…")
+      color = _LOCAL_DISCOVERED_COLOR
+    status_font = gui_app.font(FontWeight.NORMAL)
+    rl.draw_text_ex(status_font, status, rl.Vector2(content_rect.x, y), 40, 0.0, color)
+    return -1

openpilot/selfdrive/ui/sunnypilot/mici/layouts/sunnylink.py

@@ -5,21 +5,34 @@ This file is part of sunnypilot and is licensed under the MIT License.
 See the LICENSE.md file in the root directory for more details.
 """
 import pyray as rl
-
+from functools import partial
 
 from openpilot.cereal import custom
+from openpilot.common.version import sunnylink_consent_version, sunnylink_consent_declined
 from openpilot.selfdrive.ui.mici.widgets.button import BigButton, BigToggle
-from openpilot.selfdrive.ui.mici.widgets.dialog import BigDialog, BigConfirmationDialog
+from openpilot.selfdrive.ui.mici.widgets.dialog import BigDialog, BigConfirmationDialog, BigDialogBase
 from openpilot.selfdrive.ui.sunnypilot.mici.layouts.onboarding import SunnylinkConsentPage
 from openpilot.selfdrive.ui.sunnypilot.mici.widgets.sunnylink_pairing_dialog import SunnylinkPairingDialog
 from openpilot.selfdrive.ui.ui_state import ui_state
 from openpilot.sunnypilot.sunnylink.api import UNREGISTERED_SUNNYLINK_DONGLE_ID
+from openpilot.sunnypilot.sunnylink.athena.local_discovery import latest_discovered_app
+from openpilot.sunnypilot.sunnylink.athena.local_pairing import (
+  LocalApp,
+  arm_pairing,
+  clear_pairing_request,
+  get_local_apps,
+  local_app_display_name,
+  pairing_requested,
+  read_pairing_code,
+  remove_local_app,
+)
 from openpilot.system.ui.lib.application import gui_app, MousePos, FontWeight
 from openpilot.system.ui.lib.multilang import tr
 from openpilot.system.ui.widgets import Widget
 from openpilot.system.ui.widgets.label import UnifiedLabel
 from openpilot.system.ui.widgets.scroller import NavScroller
-from openpilot.common.version import sunnylink_consent_version, sunnylink_consent_declined
+
+MAX_LOCAL_APPS = 4
 
 class SunnylinkInfo(Widget):
   def __init__(self):
@@ -73,11 +86,15 @@ class SunnylinkLayoutMici(NavScroller):
     self._sunnylink_uploader_toggle = BigToggle(text=tr("sunnylink uploader"), initial_state=False,
                                                 toggle_callback=self._sunnylink_uploader_callback)
 
+    self._mobile_app_btn = BigButton(tr("sunnylink local"), "")
+    self._mobile_app_btn.set_click_callback(lambda: gui_app.push_widget(LocalAppsPanelMici()))
+
     self._scroller.add_widgets([
       self._sunnylink_info,
       self._sunnylink_toggle,
       self._sunnylink_sponsor_button,
       self._sunnylink_pair_button,
+      self._mobile_app_btn,
       self._backup_btn,
       self._restore_btn,
       self._sunnylink_uploader_toggle
@@ -110,6 +127,7 @@ class SunnylinkLayoutMici(NavScroller):
       self._sunnylink_pair_button.set_text(tr("paired"))
     else:
       self._sunnylink_pair_button.set_text(tr("pair"))
+    self._mobile_app_btn.set_visible(self._sunnylink_enabled)
 
   def show_event(self):
     super().show_event()
@@ -140,6 +158,8 @@ class SunnylinkLayoutMici(NavScroller):
       gui_app.push_widget(sl_terms_dlg)
     else:
       ui_state.params.put_bool("SunnylinkEnabled", state)
+      if not state:
+        clear_pairing_request()
 
     ui_state.update_params()
 
@@ -252,3 +272,108 @@ class SunnylinkPairBigButton(BigButton):
       dlg = SunnylinkPairingDialog(sponsor_pairing=False)
     if dlg:
       gui_app.push_widget(dlg)
+
+
+class LocalAppsPanelMici(NavScroller):
+
+  def __init__(self):
+    super().__init__()
+    self._local_apps_cache: list[LocalApp] = []
+
+    self._pair_app_btn = BigButton(tr("pair app"), "")
+    self._pair_app_btn.set_click_callback(lambda: gui_app.push_widget(LocalPairingCodeDialogMici()))
+
+    self._local_app_btns: list[BigButton] = []
+    for i in range(MAX_LOCAL_APPS):
+      btn = BigButton("", "")
+      btn.set_click_callback(partial(self._confirm_unpair_local_app, i))
+      self._local_app_btns.append(btn)
+
+    self._scroller.add_widgets([self._pair_app_btn, *self._local_app_btns])
+
+  def _update_state(self):
+    super()._update_state()
+    self._local_apps_cache = get_local_apps()
+    for i, btn in enumerate(self._local_app_btns):
+      btn.set_visible(i < len(self._local_apps_cache))
+      if i < len(self._local_apps_cache):
+        app = self._local_apps_cache[i]
+        btn.set_text(local_app_display_name(app))
+        btn.set_value(app.endpoint)
+
+  def _confirm_unpair_local_app(self, index: int):
+    apps = self._local_apps_cache
+    if index >= len(apps):
+      return
+    app = apps[index]
+
+    def unpair():
+      remove_local_app(app.app_id)
+
+    icon = gui_app.texture("icons_mici/settings/device/update.png", 64, 64)
+    dlg = BigConfirmationDialog(
+      tr("slide to unpair"),
+      icon,
+      confirm_callback=unpair,
+      red=True,
+    )
+    gui_app.push_widget(dlg)
+
+
+class LocalPairingCodeDialogMici(BigDialogBase):
+
+  def __init__(self):
+    super().__init__()
+    self._apps_before = len(get_local_apps())
+    arm_pairing()
+    self.set_back_callback(clear_pairing_request)
+
+    header_color = rl.Color(255, 255, 255, int(255 * 0.9))
+    subheader_color = rl.Color(255, 255, 255, int(255 * 0.9 * 0.65))
+    self._title = UnifiedLabel(tr("pair with mobile app"), font_size=48, font_weight=FontWeight.BOLD,
+                               text_color=header_color, line_height=0.8)
+    self._code_label = UnifiedLabel("", font_size=110, font_weight=FontWeight.DISPLAY,
+                                    text_color=rl.Color(0, 255, 0, 255))
+    self._hint = UnifiedLabel(tr("enter this code in the sunnylink app"), font_size=32,
+                              text_color=subheader_color, line_height=0.9)
+    self._status = UnifiedLabel("", font_size=28,
+                                text_color=rl.Color(255, 255, 255, int(255 * 0.45)), line_height=0.9)
+
+  def _update_state(self):
+    super()._update_state()
+    if self.is_dismissing:
+      return
+    if len(get_local_apps()) > self._apps_before:
+      self.dismiss()  # paired — window already cleared
+    elif not pairing_requested():
+      self.dismiss()  # window expired
+
+  def _render(self, _):
+    self._code_label.set_text(read_pairing_code() or "—")
+
+    discovered = latest_discovered_app()
+    if discovered is not None:
+      endpoint, age = discovered
+      self._status.set_text(endpoint if age < 2 else f"{endpoint} ({age}s)")
+      self._status.set_text_color(rl.Color(0, 255, 0, 255))
+    else:
+      self._status.set_text(tr("waiting for the app…"))
+      self._status.set_text_color(rl.Color(255, 255, 255, int(255 * 0.45)))
+
+    x = self._rect.x + 20
+    width = int(self._rect.width - 40)
+    self._title.set_max_width(width)
+    self._title.set_position(x, self._rect.y + 40)
+    self._title.render()
+
+    self._code_label.set_max_width(width)
+    self._code_label.set_position(x, self._rect.y + 130)
+    self._code_label.render()
+
+    self._hint.set_max_width(width)
+    self._hint.set_position(x, self._rect.y + 290)
+    self._hint.render()
+
+    self._status.set_max_width(width)
+    self._status.set_position(x, self._rect.y + 360)
+    self._status.render()

openpilot/sunnypilot/modeld_v2/compile_modeld.py

@@ -32,7 +32,7 @@ def _patch_tinygrad_fetch_fw():
   helpers.fetch_fw = fetch_fw
 _patch_tinygrad_fetch_fw()
 
-import openpilot.sunnypilot.modeld_v2.stock_dependencies as stock
+import openpilot.selfdrive.modeld.compile_modeld as stock
 from tinygrad import dtypes
 from tinygrad.device import Device
 from tinygrad.engine.jit import TinyJit

openpilot/sunnypilot/modeld_v2/helpers.py

@@ -1,100 +0,0 @@
-"""
-Copyright (c) 2021-, Haibin Wen, sunnypilot, and a number of other contributors.
-
-This file is part of sunnypilot and is licensed under the MIT License.
-See the LICENSE.md file in the root directory for more details.
-"""
-
-import io
-import struct
-import pickle
-import inspect
-import importlib
-import enum
-
-
-def _pad_args(func, args, kwargs):
-  try:
-    sig = inspect.signature(func)
-  except Exception:
-    return args, kwargs
-  params = list(sig.parameters.values())
-  if inspect.isfunction(func) and params and params[0].name in ('cls', 'self'):
-    params = params[1:]
-
-  new_args = list(args)
-  has_varargs = any(p.kind == inspect.Parameter.VAR_POSITIONAL for p in params)
-  if len(new_args) > len(params) and not has_varargs:
-    new_args = new_args[:len(params)]
-
-  for i in range(len(new_args), len(params)):
-    param = params[i]
-    if param.kind in (inspect.Parameter.VAR_POSITIONAL, inspect.Parameter.VAR_KEYWORD):
-      continue
-    val = param.default if param.default is not inspect.Parameter.empty else None
-    new_args.append(val)
-  return new_args, kwargs
-
-
-def _enum_factory(enum_class):
-  def factory(*args, **kwargs):
-    try:
-      return enum_class(*args, **kwargs)
-    # OptOps and UOp objects in the .pkl are left over from the compilation phase,
-    # reassignment does nothing because they aren't tied to the execution graph
-    # It never executes or evaluates the UOp nodes again.
-    except ValueError:
-      return list(enum_class)[0]
-  factory.__name__ = enum_class.__name__
-  factory.__module__ = enum_class.__module__
-  return factory
-
-
-def _dynamic_factory(real_class):
-  if isinstance(real_class, type) and issubclass(real_class, enum.Enum):
-    return _enum_factory(real_class)
-
-  def factory(*args, **kwargs):
-    try:
-      return real_class(*args, **kwargs)
-    except TypeError:
-      new_args, new_kwargs = _pad_args(real_class, args, kwargs)
-      return real_class(*new_args, **new_kwargs)
-
-  class DynamicMeta(type(real_class)):
-    def __call__(cls, *args, **kwargs):
-      return factory(*args, **kwargs)
-
-  class DynamicProxy(real_class, metaclass=DynamicMeta):
-    __slots__ = ()
-
-    def __new__(cls, *args, **kwargs):
-      return factory(*args, **kwargs)
-
-  DynamicProxy.__name__ = real_class.__name__
-  DynamicProxy.__module__ = real_class.__module__
-  return DynamicProxy
-
-
-class DynamicTinygradUnpickler(pickle.Unpickler):
-  def find_class(self, module, name):
-    if module == "tinygrad.ops":
-      try:
-        importlib.import_module("tinygrad.uops")
-        module = "tinygrad.uops"
-      except ImportError:
-        pass
-    real_class = getattr(importlib.import_module(module), name)
-    if module.startswith("tinygrad"):
-      return _dynamic_factory(real_class)
-    return real_class
-
-
-def load_oob(f):
-  opcodes = f.read(struct.unpack('<q', f.read(8))[0])
-  def buffers():
-    while (h := f.read(8)):
-      pb = pickle.PickleBuffer(bytearray(struct.unpack('<q', h)[0]))
-      f.readinto(pb)
-      yield pb
-  return DynamicTinygradUnpickler(io.BytesIO(opcodes), buffers=buffers()).load()

openpilot/sunnypilot/modeld_v2/model_adapters.py

@@ -1,174 +0,0 @@
-"""
-Copyright (c) 2021-, Haibin Wen, sunnypilot, and a number of other contributors.
-
-This file is part of sunnypilot and is licensed under the MIT License.
-See the LICENSE.md file in the root directory for more details.
-"""
-import codecs
-import pickle
-import numpy as np
-from tinygrad.tensor import Tensor
-from tinygrad_repo.examples.openpilot.helpers import allocate_inputs
-
-from openpilot.sunnypilot.modeld_v2.stock_dependencies import MODELD_INPUTS, make_input_queues as make_stock_input_queues
-from openpilot.system.camerad.cameras.nv12_info import get_nv12_info
-from openpilot.sunnypilot.modeld_v2.compile_modeld import (derive_frame_skip, make_split_input_queues, make_supercombo_input_queues,
-                                                           WARP_INPUTS, POLICY_INPUTS, nv12_copy_size)
-
-
-class BaseModelAdapter:
-  def __init__(self, jits, cam_w, cam_h, model_device, queue_device, warp_device, chestnut=False):
-    self.jits = jits
-    self.DEV = model_device
-    self.QUEUE_DEV = queue_device
-    self.WARP_DEV = warp_device
-    self.chestnut = chestnut
-    self.cam_w = cam_w
-    self.cam_h = cam_h
-    self._combined_model_type = 'supercombo'
-    self._policy_slices_list = []
-    self._has_on_policy = False
-    self.policy_output_slices = {}
-    self._policy_keys = []
-    self.full_frames = {}
-    self._blob_cache = {}
-    self.frame_buffers = {}
-    self.frame_views = {}
-    self.nv12_info = get_nv12_info(cam_w, cam_h)
-    self.is_native = False
-
-  def _init_common(self):
-    self._desire_key = next((key for key in getattr(self, 'numpy_inputs', {}) if key.startswith('desire')), 'desire')
-    self._road_key = next((key for key in getattr(self, '_vision_input_names', []) if 'big' not in key), 'img')
-    self._wide_key = next((key for key in getattr(self, '_vision_input_names', []) if 'big' in key), 'big_img')
-    self.frame_buf_params = dict.fromkeys(getattr(self, '_vision_input_names', ['img', 'big_img']), self.nv12_info)
-
-  def get_dummy_inputs(self):
-    dummy_size = getattr(self, 'frame_copy_size', self.frame_buf_params[self._road_key][3])
-    if getattr(self, 'is_run_model', True) is False:
-      dummy_size = self.frame_buf_params[self._road_key][3]
-
-    dummy_frames = {
-      k: np.zeros(self.frame_views[k].size, dtype=np.uint8) if self.is_native else np.zeros(dummy_size, dtype=np.uint8)
-      for k in self._vision_input_names
-    }
-    transforms = {k: np.eye(3, dtype=np.float32) for k in [self._road_key, self._wide_key] if k}
-    dummy_inputs = {k: np.zeros(v.shape, dtype=v.dtype) for k, v in self.numpy_inputs.items() if k not in ['tfm', 'big_tfm', 'prev_feat']}
-    return dummy_frames, transforms, dummy_inputs
-
-
-class LegacyModelAdapter(BaseModelAdapter):
-  def __init__(self, *args, **kwargs):
-    super().__init__(*args, **kwargs)
-
-    metadata = self.jits['metadata']
-    self.is_run_model = 'run_model' in self.jits
-    self.frame_copy_size = nv12_copy_size(*self.nv12_info[:3])
-
-    if self.is_run_model or 'model' in metadata:
-      model_metadata = metadata.get('model', metadata)
-      self.input_shapes = model_metadata['input_shapes']
-      self.vision_output_slices = model_metadata['output_slices']
-      self._vision_input_names = [key for key in self.input_shapes if 'img' in key]
-      self.frame_skip = derive_frame_skip({}, self.input_shapes)
-
-      if self.is_run_model:
-        self.input_queues, self.numpy_inputs, self.frame_buffers = make_stock_input_queues(
-          self.input_shapes, self.frame_skip, device=self.DEV, frame_copy_size=self.frame_copy_size)
-        self.frame_views, self.npy = self.frame_buffers, self.numpy_inputs
-        self.run_model, self.run_policy, self.warp = self.jits['run_model'][(self.cam_w, self.cam_h)], None, None
-      else:
-        self.input_queues, self.numpy_inputs = make_supercombo_input_queues(self.input_shapes, self.frame_skip, device=self.QUEUE_DEV)
-        self.run_model, self.run_policy, self.warp = None, self.jits['run_policy'], self.jits[(self.cam_w, self.cam_h)]
-    else:
-      self.run_model, self.run_policy, self.warp = None, self.jits['run_policy'], self.jits[(self.cam_w, self.cam_h)]
-      vision_metadata = metadata['vision']
-      policy_keys = [k for k in metadata if k not in ('vision', 'warp_dev')]
-      self._combined_model_type = 'split' if policy_keys == ['policy'] else 'multi_policy'
-      self.vision_output_slices = vision_metadata['output_slices']
-      self._policy_keys = policy_keys
-      self._policy_slices_list = [metadata[k]['output_slices'] for k in policy_keys]
-      self.policy_output_slices = self._policy_slices_list[0]
-      self._has_on_policy = any('on' in k.lower() for k in policy_keys)
-      self._vision_input_names = [key for key in vision_metadata['input_shapes'] if 'img' in key]
-      first_policy_meta = metadata[policy_keys[0]]
-      self.frame_skip = derive_frame_skip(vision_metadata['input_shapes'], first_policy_meta['input_shapes'])
-      self.input_queues, self.numpy_inputs = make_split_input_queues(vision_metadata['input_shapes'],
-                                                                     first_policy_meta['input_shapes'],
-                                                                     self.frame_skip, device=self.QUEUE_DEV)
-
-    self._init_common()
-    if self.warp is not None:
-      self.full_frames = {k: Tensor(np.zeros(self.nv12_info[3], dtype=np.uint8), device=self.WARP_DEV).contiguous().realize() for k in self._vision_input_names}
-      self.warp(**{k: self.input_queues[k] for k in WARP_INPUTS}, frame=self.full_frames[self._road_key], big_frame=self.full_frames[self._wide_key])
-
-  def copy_frames(self, bufs):
-    if getattr(self, 'is_run_model', True):
-      for key, buf in bufs.items():
-        data = buf.data if hasattr(buf, 'data') else buf
-        np.copyto(self.frame_buffers[key], np.frombuffer(data, dtype=np.uint8, count=self.frame_copy_size))
-    else:
-      for key, buf in bufs.items():
-        ptr = np.frombuffer(buf.data, dtype=np.uint8).ctypes.data
-        cache_key = (key, ptr)
-        if cache_key not in self._blob_cache:
-          self._blob_cache[cache_key] = Tensor.from_blob(ptr, (self.frame_buf_params[key][3],), dtype='uint8', device=self.WARP_DEV)
-        self.full_frames[key] = self._blob_cache[cache_key]
-
-  def reset_warmup_buffers(self):
-    if getattr(self, 'is_run_model', True):
-      self.input_queues, self.numpy_inputs, self.frame_buffers = make_stock_input_queues(
-        self.input_shapes, self.frame_skip, device=self.DEV, frame_copy_size=self.frame_copy_size)
-      self.frame_views = self.frame_buffers
-      self.npy = self.numpy_inputs
-    else:
-      for v in self.numpy_inputs.values():
-        v[:] = 0
-      self.full_frames.clear()
-      self._blob_cache.clear()
-
-  def run(self):
-    if self.run_model is not None:
-      outs, = self.run_model(**{k: self.input_queues[k] for k in MODELD_INPUTS})
-      return outs
-    else:
-      assert self.warp is not None and self.run_policy is not None
-      warped = self.warp(**{k: self.input_queues[k] for k in WARP_INPUTS}, frame=self.full_frames[self._road_key], big_frame=self.full_frames[self._wide_key])
-      raw_outputs = self.run_policy(**{k: self.input_queues[k] for k in POLICY_INPUTS if k in self.input_queues}, warped=warped)
-      return raw_outputs
-
-
-class NativeTinygradAdapter(BaseModelAdapter):
-  def __init__(self, *args, **kwargs):
-    super().__init__(*args, **kwargs)
-    self.is_native = True
-    variant = self.jits['variants'][f'{self.cam_w}x{self.cam_h}']
-    self.input_specs = variant['input_specs']
-    self.packed_specs = variant['packed_specs']
-    self.run_model = variant['run']
-    self.vision_output_slices = pickle.loads(codecs.decode(self.jits['metadata']['metadata']['output_slices'].encode(), 'base64'))
-    self._vision_input_names = ['img', 'big_img']
-    self.reset_warmup_buffers()
-    self._init_common()
-
-  def copy_frames(self, bufs):
-    for key, buf in bufs.items():
-      data = buf.data if hasattr(buf, 'data') else buf
-      if key in self.frame_views:
-        np.copyto(self.frame_views[key], np.frombuffer(data, dtype=np.uint8, count=self.frame_views[key].size))
-
-  def reset_warmup_buffers(self) -> None:
-    self.input_queues, views = allocate_inputs(self.input_specs, self.packed_specs)
-    self.frame_views = {name: views[name] for name in self._vision_input_names if name in views}
-    self.numpy_inputs = {name: views[name] for name in views if name not in self.frame_views}
-
-  def run(self):
-    outs = self.run_model(**self.input_queues)
-    return outs
-
-
-def get_model_adapter(jits, cam_w, cam_h, model_device, queue_device, warp_device, chestnut=False):
-  if 'variants' in jits:
-    return NativeTinygradAdapter(jits, cam_w, cam_h, model_device, queue_device, warp_device, chestnut=chestnut)
-  else:
-    return LegacyModelAdapter(jits, cam_w, cam_h, model_device, queue_device, warp_device, chestnut=chestnut)

openpilot/sunnypilot/modeld_v2/modeld.py

@@ -13,10 +13,11 @@ import numpy as np
 import threading
 import time
 from setproctitle import setproctitle
+from tinygrad.tensor import Tensor
 
 import openpilot.cereal.messaging as messaging
 from openpilot.common.hardware import COMMA_HARDWARE
-from openpilot.selfdrive.modeld.helpers import chestnut_present
+from openpilot.selfdrive.modeld.helpers import chestnut_present, load_oob
 from openpilot.cereal import log
 from opendbc.car.structs import car
 from openpilot.cereal.services import SERVICE_LIST
@@ -32,19 +33,25 @@ from openpilot.common.realtime import config_realtime_process, DT_MDL
 from openpilot.common.transformations.camera import DEVICE_CAMERAS
 from openpilot.common.transformations.model import get_warp_matrix
 from openpilot.system import sentry
+from openpilot.system.camerad.cameras.nv12_info import get_nv12_info
 from openpilot.selfdrive.controls.lib.desire_helper import DesireHelper
 from openpilot.selfdrive.controls.lib.drive_helpers import get_accel_from_plan, smooth_value
 from openpilot.selfdrive.modeld.modeld import ChestnutState
 
+from openpilot.selfdrive.modeld.compile_modeld import (
+  MODELD_INPUTS,
+  make_input_queues as make_stock_input_queues,
+)
 from openpilot.sunnypilot.modeld_v2.fill_model_msg import fill_model_msg, fill_pose_msg, PublishState, get_curvature_from_output
 from openpilot.sunnypilot.modeld_v2.parse_model_outputs import Parser
 from openpilot.sunnypilot.modeld_v2.constants import ModelConstants, Plan
 from openpilot.sunnypilot.modeld_v2.meta_helper import load_meta_constants
 from openpilot.sunnypilot.modeld_v2.camera_offset_helper import CameraOffsetHelper
+from openpilot.sunnypilot.modeld_v2.compile_modeld import (derive_frame_skip, make_split_input_queues,
+                                                           make_supercombo_input_queues, nv12_copy_size,
+                                                           WARP_INPUTS, POLICY_INPUTS)
 from openpilot.sunnypilot.livedelay.helpers import get_lat_delay
 from openpilot.sunnypilot.modeld_v2.modeld_base import ModelStateBase
-from openpilot.sunnypilot.modeld_v2.helpers import load_oob
-from openpilot.sunnypilot.modeld_v2.model_adapters import get_model_adapter
 from openpilot.sunnypilot.models.helpers import get_active_bundle
 from openpilot.sunnypilot.selfdrive.controls.lib.relc import RoadEdgeLaneChangeController
 
@@ -115,19 +122,52 @@ class ModelState(ModelStateBase):
     self.WARP_DEV = metadata.get('warp_dev', 'QCOM') if COMMA_HARDWARE else 'CPU'
     self.DEV = ('AMD' if self.chestnut else 'QCOM') if COMMA_HARDWARE else 'CPU'
     self.QUEUE_DEV = self.DEV
-    self.adapter = get_model_adapter(jits, cam_w, cam_h, self.DEV, self.QUEUE_DEV, self.WARP_DEV, self.chestnut)
-    self.vision_output_slices = self.adapter.vision_output_slices
-    self.policy_output_slices = self.adapter.policy_output_slices
-    self._policy_slices_list = self.adapter._policy_slices_list
-    self._combined_model_type = self.adapter._combined_model_type
-    self._vision_input_names = self.adapter._vision_input_names
-    self.numpy_inputs = self.adapter.numpy_inputs
-    self._policy_keys = self.adapter._policy_keys
-    self._has_on_policy = self.adapter._has_on_policy
-    self._desire_key = self.adapter._desire_key
-    self._road_key = self.adapter._road_key
-    self._wide_key = self.adapter._wide_key
-    self.frame_buf_params = self.adapter.frame_buf_params
+    self.is_run_model = 'run_model' in jits
+
+    nv12_info = get_nv12_info(cam_w, cam_h)
+    self.frame_copy_size = nv12_copy_size(*nv12_info[:3])
+    self.full_frames: dict = {}
+    self._blob_cache: dict = {}
+    self.frame_buffers: dict = {}
+
+    if self.is_run_model or 'model' in metadata:
+      model_metadata = metadata.get('model', metadata)
+      self.input_shapes = model_metadata['input_shapes']
+      self.vision_output_slices = model_metadata['output_slices']
+      self.policy_output_slices = {}
+      self._policy_slices_list = []
+      self._combined_model_type = 'supercombo'
+      self._vision_input_names = [key for key in self.input_shapes if 'img' in key]
+      self.frame_skip = derive_frame_skip({}, self.input_shapes)
+      if self.is_run_model:
+        self.input_queues, self.numpy_inputs, self.frame_buffers = make_stock_input_queues(
+          self.input_shapes, self.frame_skip, device=self.DEV, frame_copy_size=self.frame_copy_size)
+        self.frame_views, self.npy = self.frame_buffers, self.numpy_inputs
+        self.run_model, self.run_policy, self.warp = jits['run_model'][(cam_w, cam_h)], None, None
+      else:
+        self.input_queues, self.numpy_inputs = make_supercombo_input_queues(self.input_shapes, self.frame_skip, device=self.QUEUE_DEV)
+        self.run_model, self.run_policy, self.warp = None, jits['run_policy'], jits[(cam_w, cam_h)]
+    else:
+      self.run_model, self.run_policy, self.warp = None, jits['run_policy'], jits[(cam_w, cam_h)]
+      vision_metadata = metadata['vision']
+      policy_keys = [k for k in metadata if k not in ('vision', 'warp_dev')]
+      self._combined_model_type = 'split' if policy_keys == ['policy'] else 'multi_policy'
+      self.vision_output_slices = vision_metadata['output_slices']
+      self._policy_keys = policy_keys
+      self._policy_slices_list = [metadata[k]['output_slices'] for k in policy_keys]
+      self.policy_output_slices = self._policy_slices_list[0]
+      self._has_on_policy = any('on' in k.lower() for k in policy_keys)
+      self._vision_input_names = [key for key in vision_metadata['input_shapes'] if 'img' in key]
+      first_policy_meta = metadata[policy_keys[0]]
+      frame_skip = derive_frame_skip(vision_metadata['input_shapes'], first_policy_meta['input_shapes'])
+      self.input_queues, self.numpy_inputs = make_split_input_queues(vision_metadata['input_shapes'],
+                                                                     first_policy_meta['input_shapes'],
+                                                                     frame_skip, device=self.QUEUE_DEV)
+
+    self._desire_key = next(key for key in self.numpy_inputs if key.startswith('desire'))
+    self._road_key = next(key for key in self._vision_input_names if 'big' not in key)
+    self._wide_key = next(key for key in self._vision_input_names if 'big' in key)
+    self.frame_buf_params = dict.fromkeys(self._vision_input_names, nv12_info)
 
     is_20hz = bundle.is20hz if bundle else self._combined_model_type in ('split', 'multi_policy')
     if is_20hz:
@@ -139,10 +179,26 @@ class ModelState(ModelStateBase):
     self.parser = Parser()
     self.prev_desire = np.zeros(self.constants.DESIRE_LEN, dtype=np.float32)
 
+    if self.warp is not None:
+      self.full_frames = {k: Tensor(np.zeros(nv12_info[3], dtype=np.uint8), device=self.WARP_DEV).contiguous().realize() for k in self._vision_input_names}
+      self.warp(**{k: self.input_queues[k] for k in WARP_INPUTS}, frame=self.full_frames[self._road_key], big_frame=self.full_frames[self._wide_key])
+
   def warmup(self) -> None:
-    dummy_frames, transforms, dummy_inputs = self.adapter.get_dummy_inputs()
+    dummy_size = self.frame_copy_size if self.is_run_model else self.frame_buf_params[self._road_key][3]
+    dummy_frames = {k: np.zeros(dummy_size, dtype=np.uint8) for k in self._vision_input_names}
+    transforms = {k: np.eye(3, dtype=np.float32) for k in [self._road_key, self._wide_key] if k}
+    dummy_inputs = {k: np.zeros(v.shape, dtype=v.dtype) for k, v in self.numpy_inputs.items() if k not in ['tfm', 'big_tfm', 'prev_feat']}
     self.run(dummy_frames, transforms, dummy_inputs)
-    self.adapter.reset_warmup_buffers()
+    if self.is_run_model:
+      self.input_queues, self.numpy_inputs, self.frame_buffers = make_stock_input_queues(
+        self.input_shapes, self.frame_skip, device=self.DEV, frame_copy_size=self.frame_copy_size)
+      self.frame_views = self.frame_buffers
+      self.npy = self.numpy_inputs
+    else:
+      for v in self.numpy_inputs.values():
+        v[:] = 0
+      self.full_frames.clear()
+      self._blob_cache.clear()
     self.prev_desire[:] = 0
 
   @property
@@ -160,12 +216,21 @@ class ModelState(ModelStateBase):
   def run(self, bufs: dict[str, VisionBuf], transforms: dict[str, np.ndarray],
           inputs: dict[str, np.ndarray],
           after_enqueue: Callable[[], None] | None = None) -> dict[str, np.ndarray] | None:
-    self.adapter.copy_frames(bufs)
+    if self.is_run_model:
+      for key, buf in bufs.items():
+        data = buf.data if hasattr(buf, 'data') else buf
+        np.copyto(self.frame_buffers[key], np.frombuffer(data, dtype=np.uint8, count=self.frame_copy_size))
+    else:
+      for key, buf in bufs.items():
+        ptr = np.frombuffer(buf.data, dtype=np.uint8).ctypes.data
+        cache_key = (key, ptr)
+        if cache_key not in self._blob_cache:
+          self._blob_cache[cache_key] = Tensor.from_blob(ptr, (self.frame_buf_params[key][3],), dtype='uint8', device=self.WARP_DEV)
+        self.full_frames[key] = self._blob_cache[cache_key]
 
     desire_key = self.desire_key
     inputs[desire_key][0] = 0
-    (self.numpy_inputs[desire_key].flat if self.adapter.is_native else self.numpy_inputs[desire_key])[:] = \
-      np.where(inputs[desire_key] - self.prev_desire > .99, inputs[desire_key], 0)
+    self.numpy_inputs[desire_key][:] = np.where(inputs[desire_key] - self.prev_desire > .99, inputs[desire_key], 0)
     self.prev_desire[:] = inputs[desire_key]
 
     for key in ('traffic_convention', 'lateral_control_params', 'action_t'):
@@ -175,7 +240,13 @@ class ModelState(ModelStateBase):
     self.numpy_inputs['tfm'][:, :] = transforms[self._road_key].reshape(3, 3)
     self.numpy_inputs['big_tfm'][:, :] = transforms[self._wide_key].reshape(3, 3)
 
-    raw_outputs = self.adapter.run()
+    if self.run_model is not None:
+      outs, = self.run_model(**{k: self.input_queues[k] for k in MODELD_INPUTS})
+      raw_outputs = outs
+    else:
+      assert self.warp is not None and self.run_policy is not None
+      warped = self.warp(**{k: self.input_queues[k] for k in WARP_INPUTS}, frame=self.full_frames[self._road_key], big_frame=self.full_frames[self._wide_key])
+      raw_outputs = self.run_policy(**{k: self.input_queues[k] for k in POLICY_INPUTS if k in self.input_queues}, warped=warped)
 
     if after_enqueue is not None:
       after_enqueue()
@@ -187,16 +258,14 @@ class ModelState(ModelStateBase):
       sliced = {k: model_output[np.newaxis, v] for k, v in self.vision_output_slices.items()}
       outputs = self.parser.parse_outputs(sliced)
       if 'prev_feat' in self.numpy_inputs and 'hidden_state' in self.vision_output_slices:
-        (self.numpy_inputs['prev_feat'].flat if self.adapter.is_native else self.numpy_inputs['prev_feat'])[:] = \
-          model_output[self.vision_output_slices['hidden_state']]
+        self.numpy_inputs['prev_feat'][:] = model_output[self.vision_output_slices['hidden_state']]
     else:
       vision_output = raw_outputs[0].numpy().flatten()
       vision_sliced = {k: vision_output[np.newaxis, v] for k, v in self.vision_output_slices.items()}
       outputs = self.parser.parse_vision_outputs(vision_sliced)
 
       if 'prev_feat' in self.numpy_inputs and 'hidden_state' in self.vision_output_slices:
-        (self.numpy_inputs['prev_feat'].flat if self.adapter.is_native else self.numpy_inputs['prev_feat'])[:] = \
-          vision_output[self.vision_output_slices['hidden_state']]
+        self.numpy_inputs['prev_feat'][:] = vision_output[self.vision_output_slices['hidden_state']]
 
       for i, policy_slices in enumerate(self._policy_slices_list):
         policy_output = raw_outputs[i + 1].numpy().flatten()
@@ -303,7 +372,11 @@ def main(demo=False):
     loader.start()
     loader.join(BIG_MODEL_TIMEOUT)
     model = big_model
+    if model is None:
+      params.put_bool("ChestnutModelError", True)
     params.put_bool("ChestnutActive", model is not None)
+    if model is not None:
+      params.remove("ChestnutModelError")
 
   small_model = ModelState(cam_w=vipc_client_main.width, cam_h=vipc_client_main.height, chestnut=False) if model is None or CHESTNUT else None
   if model is None:
@@ -445,7 +518,8 @@ def main(demo=False):
     except Exception:
       if not params.get_bool("ChestnutActive"):
         raise
-      cloudlog.exception("big model failed, fall back to small")
+      cloudlog.exception("chestnut failed, falling back to small")
+      params.put_bool("ChestnutModelError", True)
       params.put_bool("ChestnutActive", False)
       assert small_model is not None
       model = small_model

openpilot/sunnypilot/modeld_v2/stock_dependencies.py

@@ -1,167 +0,0 @@
-"""
-Copyright (c) 2021-, Haibin Wen, sunnypilot, and a number of other contributors.
-
-This file is part of sunnypilot and is licensed under the MIT License.
-See the LICENSE.md file in the root directory for more details.
-"""
-import math
-import numpy as np
-from collections import namedtuple
-from tinygrad.tensor import Tensor
-from tinygrad.helpers import Context
-from tinygrad.device import Device
-
-"""
-Frozen in time compile_modeld dependencies to support all models prior to transition to tinygrad compilation.
-This file is not meant to be modified.
-"""
-NV12Frame = namedtuple("NV12Frame", ['width', 'height', 'stride', 'y_height', 'uv_height', 'size'])
-MODELD_INPUTS = ['img_q', 'big_img_q', 'feat_q', 'desire_q', 'packed_npy_inputs']
-
-
-def nv12_copy_size(stride: int, y_height: int, uv_height: int) -> int:
-  return stride * (y_height + uv_height)
-
-
-def warp_perspective_tinygrad(src_flat, M_inv, dst_shape, src_shape, stride_pad, border_fill_val=None):
-  w_dst, h_dst = dst_shape
-  h_src, w_src = src_shape
-
-  x = Tensor.arange(w_dst).reshape(1, w_dst).expand(h_dst, w_dst).reshape(-1)
-  y = Tensor.arange(h_dst).reshape(h_dst, 1).expand(h_dst, w_dst).reshape(-1)
-
-  src_x = M_inv[0, 0] * x + M_inv[0, 1] * y + M_inv[0, 2]
-  src_y = M_inv[1, 0] * x + M_inv[1, 1] * y + M_inv[1, 2]
-  src_w = M_inv[2, 0] * x + M_inv[2, 1] * y + M_inv[2, 2]
-
-  src_x = src_x / src_w
-  src_y = src_y / src_w
-
-  x_round = Tensor.round(src_x)
-  y_round = Tensor.round(src_y)
-  x_nn_clipped = x_round.clip(0, w_src - 1).cast('int')
-  y_nn_clipped = y_round.clip(0, h_src - 1).cast('int')
-  idx = y_nn_clipped * (w_src + stride_pad) + x_nn_clipped
-  sampled = src_flat[idx]
-
-  if border_fill_val is None:
-    return sampled
-
-  in_bounds = ((x_round >= 0) & (x_round <= w_src - 1) &
-               (y_round >= 0) & (y_round <= h_src - 1)).cast(sampled.dtype)
-  return sampled * in_bounds + Tensor(border_fill_val, dtype=sampled.dtype) * (1 - in_bounds)
-
-
-def frames_to_tensor(frames):
-  H = (frames.shape[0] * 2) // 3
-  W = frames.shape[1]
-  in_img1 = Tensor.cat(frames[0:H:2, 0::2],
-                       frames[1:H:2, 0::2],
-                       frames[0:H:2, 1::2],
-                       frames[1:H:2, 1::2],
-                       frames[H:H+H//4].reshape((H//2, W//2)),
-                       frames[H+H//4:H+H//2].reshape((H//2, W//2)), dim=0).reshape((6, H//2, W//2))
-  return in_img1
-
-
-def make_frame_prepare(nv12: NV12Frame, model_w, model_h):
-  cam_w, cam_h, stride, y_height, uv_height, _ = nv12
-  uv_offset = stride * y_height
-  stride_pad = stride - cam_w
-
-  def frame_prepare_tinygrad(input_frame, M_inv):
-    M_inv_uv = M_inv * Tensor([[1.0, 1.0, 0.5], [1.0, 1.0, 0.5], [2.0, 2.0, 1.0]], device=Device.DEFAULT)
-    uv = input_frame[uv_offset:uv_offset + uv_height * stride].reshape(uv_height, stride)
-    with Context(SPLIT_REDUCEOP=0):
-      y = warp_perspective_tinygrad(input_frame[:cam_h*stride],
-                                    M_inv, (model_w, model_h),
-                                    (cam_h, cam_w), stride_pad).realize()
-      u = warp_perspective_tinygrad(uv[:cam_h//2, :cam_w:2].flatten(),
-                                    M_inv_uv, (model_w//2, model_h//2),
-                                    (cam_h//2, cam_w//2), 0).realize()
-      v = warp_perspective_tinygrad(uv[:cam_h//2, 1:cam_w:2].flatten(),
-                                    M_inv_uv, (model_w//2, model_h//2),
-                                    (cam_h//2, cam_w//2), 0).realize()
-    yuv = y.cat(u).cat(v).reshape((model_h * 3 // 2, model_w))
-    tensor = frames_to_tensor(yuv)
-    return tensor
-  return frame_prepare_tinygrad
-
-
-def get_policy_npy_shapes(input_shapes):
-  dp = input_shapes['desire_pulse']
-  tc = input_shapes['traffic_convention']
-  at = input_shapes['action_t']
-  fb = input_shapes['features_buffer']
-  feat_dim = math.prod(fb[2:])
-  shapes = {'desire': (dp[2],), 'traffic_convention': tuple(tc), 'action_t': tuple(at), 'prev_feat': (fb[0], feat_dim)}
-  return shapes, [math.prod(s) for s in shapes.values()]
-
-
-def make_input_queues(input_shapes, frame_skip, device, frame_copy_size):
-  img = input_shapes['img']
-  fb = input_shapes['features_buffer']
-  feat_dim = math.prod(fb[2:])
-  dp = input_shapes['desire_pulse']
-  n_frames = img[1] // 6
-  img_buf_shape = (frame_skip * (n_frames - 1) + 1, 6, img[2], img[3])
-
-  policy_shapes, _ = get_policy_npy_shapes(input_shapes)
-  shapes = {'tfm': (3, 3), 'big_tfm': (3, 3)} | policy_shapes
-  sizes = [math.prod(s) for s in shapes.values()]
-  packed_npy_size = sum(sizes) * np.dtype(np.float32).itemsize
-  packed_input = np.zeros(packed_npy_size + 2 * frame_copy_size, dtype=np.uint8)
-  packed_npy_inputs = packed_input[:packed_npy_size].view(np.float32)
-  frames = packed_input[packed_npy_size:]
-  frame_views = {'img': frames[:frame_copy_size], 'big_img': frames[frame_copy_size:]}
-  npy = {k: v.reshape(s) for (k, s), v in zip(shapes.items(), np.split(packed_npy_inputs, np.cumsum(sizes[:-1])), strict=True)}
-  input_queues = {
-    'img_q': Tensor(np.zeros(img_buf_shape, dtype=np.uint8), device=device).contiguous().realize(),
-    'big_img_q': Tensor(np.zeros(img_buf_shape, dtype=np.uint8), device=device).contiguous().realize(),
-    'feat_q': Tensor(np.zeros((frame_skip * fb[1], fb[0], feat_dim), dtype=np.float32), device=device).contiguous().realize(),
-    'desire_q': Tensor(np.zeros((frame_skip * dp[1], dp[0], dp[2]), dtype=np.float32), device=device).contiguous().realize(),
-    'packed_npy_inputs': Tensor(packed_input, device='NPY').realize(),
-  }
-  return input_queues, npy, frame_views
-
-
-def shift_and_sample(buf, new_val, sample_fn):
-  buf.assign(buf[1:].cat(new_val, dim=0).contiguous())
-  return sample_fn(buf)
-
-
-def sample_skip(buf, frame_skip):
-  return buf[::frame_skip].contiguous().flatten(0, 1).unsqueeze(0)
-
-
-def sample_desire(buf, frame_skip):
-  return buf.reshape(-1, frame_skip, *buf.shape[1:]).max(1).flatten(0, 1).unsqueeze(0)
-
-
-def make_warp(nv12, model_w, model_h):
-  frame_prepare = make_frame_prepare(nv12, model_w, model_h)
-  def warp(tfm, big_tfm, frame, big_frame):
-    tfm = tfm.to(Device.DEFAULT)
-    big_tfm = big_tfm.to(Device.DEFAULT)
-    frame = frame.to(Device.DEFAULT)
-    big_frame = big_frame.to(Device.DEFAULT)
-    Tensor.realize(tfm, big_tfm, frame, big_frame)
-    warped_frame = frame_prepare(frame, tfm).unsqueeze(0)
-    warped_big_frame = frame_prepare(big_frame, big_tfm).unsqueeze(0)
-    return Tensor.cat(warped_frame, warped_big_frame)
-  return warp
-
-
-def make_run_model(warp, run_policy, model_metadata, frame_copy_size):
-  _, policy_sizes = get_policy_npy_shapes(model_metadata['input_shapes'])
-  packed_npy_size = (18 + sum(policy_sizes)) * np.dtype(np.float32).itemsize
-  def run_model(img_q, big_img_q, feat_q, desire_q, packed_npy_inputs):
-    packed_input = packed_npy_inputs.to(Device.DEFAULT)
-    Tensor.realize(packed_input)
-    packed_npy_inputs = packed_input[:packed_npy_size].bitcast('float32')
-    frame = packed_input[packed_npy_size:packed_npy_size + frame_copy_size]
-    big_frame = packed_input[packed_npy_size + frame_copy_size:]
-    tfm, big_tfm, policy_inputs = packed_npy_inputs.split([9, 9, sum(policy_sizes)])
-    warped = warp(tfm.reshape(3, 3), big_tfm.reshape(3, 3), frame, big_frame)
-    return run_policy(warped, img_q, big_img_q, feat_q, desire_q, policy_inputs)
-  return run_model

openpilot/sunnypilot/modeld_v2/tests/helpers.py

@@ -7,8 +7,6 @@ See the LICENSE.md file in the root directory for more details.
 
 import pathlib
 import tempfile
-import codecs
-import pickle
 
 import openpilot.sunnypilot.models.helpers as helpers
 import openpilot.sunnypilot.modeld_v2.modeld as modeld_module
@@ -163,16 +161,6 @@ ARCHETYPES = {
 
 
 def make_pkl_data(archetype):
-  if archetype.expected_model_type == 'supercombo':
-    slices_b64 = codecs.encode(pickle.dumps(archetype.metadata_structure['model']['output_slices']), 'base64').decode()
-    return {
-      'metadata': {
-        'metadata': {'output_slices': slices_b64},
-        'input_shapes': archetype.metadata_structure['model']['input_shapes']
-      },
-      'variants': {f'{CAM_W}x{CAM_H}': {'input_specs': {}, 'packed_specs': {}, 'run': _noop_jit}},
-    }
-
   return {
     'metadata': archetype.metadata_structure,
     'run_policy': _noop_jit,

openpilot/sunnypilot/modeld_v2/tests/test_combined_pkl_loader.py

@@ -12,7 +12,7 @@ import openpilot.sunnypilot.modeld_v2.modeld as modeld_module
 from openpilot.sunnypilot.modeld_v2.modeld import _find_driving_pkl
 from openpilot.sunnypilot.modeld_v2.tests import helpers as tests_helpers
 from openpilot.sunnypilot.modeld_v2.tests.helpers import DummyModel, DummyBundle, ARCHETYPES, CAM_W, CAM_H, \
-  SPLIT_VISION_INPUT_SHAPES
+  SPLIT_VISION_INPUT_SHAPES, SPLIT_POLICY_INPUT_SHAPES
 from openpilot.common.test import OpenpilotTestCase
 
 # resolved by name from this module when a test asks for them
@@ -66,6 +66,21 @@ class TestModelStateCombinedInit(OpenpilotTestCase):
 
 
 class TestStockEquivalence(OpenpilotTestCase):
+
+  def test_split_queue_keys_match_stock(self, model_state_factory):
+    from openpilot.selfdrive.modeld.compile_modeld import make_input_queues
+    from openpilot.sunnypilot.modeld_v2.compile_modeld import derive_frame_skip
+
+    state = model_state_factory(ARCHETYPES['vision_policy_split'])
+
+    frame_skip = derive_frame_skip(SPLIT_VISION_INPUT_SHAPES, SPLIT_POLICY_INPUT_SHAPES)
+    stock_shapes = {**SPLIT_VISION_INPUT_SHAPES, **SPLIT_POLICY_INPUT_SHAPES, 'action_t': (1, 2)}
+    stock_queues, stock_npy, _frame_views = make_input_queues(stock_shapes, frame_skip, device='NPY', frame_copy_size=49152)
+
+    # sunnypilot split pipeline has tfm/big_tfm as queues (stock has them in npy only)
+    assert set(stock_queues.keys()) <= set(state.input_queues.keys())
+    assert {'desire', 'traffic_convention'} <= set(state.numpy_inputs.keys())
+
   def test_split_queue_keys_work_with_desire_key(self, model_state_factory):
     from openpilot.sunnypilot.modeld_v2.compile_modeld import derive_frame_skip, make_split_input_queues
 
@@ -93,11 +108,8 @@ class TestStockEquivalence(OpenpilotTestCase):
     from openpilot.selfdrive.modeld.helpers import dump_oob
     shapes = {'img': (1, 12, 128, 256), 'big_img': (1, 12, 128, 256), 'features_buffer': (1, 24, 32, 512),
               'desire_pulse': (1, 25, 8), 'traffic_convention': (1, 2), 'action_t': (1, 2)}
-    import codecs
-    import pickle
-    slices_b64 = codecs.encode(pickle.dumps({}), 'base64').decode()
-    pkl_data = {'metadata': {'metadata': {'output_slices': slices_b64}},
-                'variants': {f'{CAM_W}x{CAM_H}': {'input_specs': {}, 'packed_specs': {}, 'run': tests_helpers._noop_jit}}}
+    pkl_data = {'metadata': {'model': {'input_shapes': shapes, 'output_slices': {}}},
+                'run_model': {(CAM_W, CAM_H): tests_helpers._noop_jit}}
     with open(tmp_path / 'driving_test_tinygrad.pkl', 'wb') as f:
       dump_oob(pkl_data, f)
     bundle = DummyBundle(models=[DummyModel('supercombo', 'driving_test_tinygrad.pkl')])

openpilot/sunnypilot/modeld_v2/tests/test_models.py

@@ -1,43 +0,0 @@
-"""
-Copyright (c) 2021-, Haibin Wen, sunnypilot, and a number of other contributors.
-
-This file is part of sunnypilot and is licensed under the MIT License.
-See the LICENSE.md file in the root directory for more details.
-"""
-
-import os
-import unittest
-from unittest.mock import patch
-from openpilot.common.file_chunker import open_file_chunked
-from openpilot.sunnypilot.modeld_v2.helpers import load_oob
-from tinygrad.device import Device
-
-
-class TestLegacyModels(unittest.TestCase):
-  def test_legacy_model_load(self):
-    base_name = os.environ.get("MODEL_BASE_NAME")
-    if not base_name:
-      raise unittest.SkipTest("MODEL_BASE_NAME env var not set, skipping integration test.")
-    chunk_dir = os.environ.get("MODEL_CHUNK_DIR", "/tmp/model_chunks")
-    base_path = os.path.join(chunk_dir, base_name)
-
-    try:
-      f = open_file_chunked(base_path)
-    except Exception as error:
-      self.fail(f"Failed to open chunked file {base_path}: {error}")
-    self.addCleanup(f.close)
-
-    real_getitem = Device.__class__.__getitem__
-
-    def safe_getitem(device_self, ix):
-      if ix == "QCOM" and not os.path.exists("/dev/kgsl-3d0"):
-        return real_getitem(device_self, "CPU")
-      if ix == "AMD" and not os.path.exists("/dev/kfd"):
-        return real_getitem(device_self, "CPU")
-      return real_getitem(device_self, ix)
-
-    with patch.object(Device.__class__, "__getitem__", safe_getitem):
-      obj = load_oob(f)
-
-    assert isinstance(obj, dict), "Parsed object is not a dictionary"
-    assert "metadata" in obj, "Metadata key is missing"

openpilot/sunnypilot/models/tests/test_tinygrad_ref.py

@@ -0,0 +1,24 @@
+import requests
+
+from openpilot.sunnypilot.models.tinygrad_ref import get_tinygrad_ref
+from openpilot.sunnypilot.models.fetcher import ModelFetcher
+from openpilot.common.test import OpenpilotTestCase
+
+def fetch_tinygrad_ref():
+  response = requests.get(ModelFetcher.MODEL_URL, timeout=10)
+  response.raise_for_status()
+  json_data = response.json()
+  return json_data.get("tinygrad_ref")
+
+
+class TestTinygradRef(OpenpilotTestCase):
+  def test_tinygrad_ref(self):
+    current_ref = get_tinygrad_ref()
+    remote_ref = fetch_tinygrad_ref()
+    assert remote_ref == current_ref, (
+      f"""tinygrad_repo ref does not match remote tinygrad_ref of current compiled driving models json.
+    Current: {current_ref}
+    Remote: {remote_ref}
+    Please run build-all workflow to update models."""
+    )
+    print("tinygrad_repo ref matches current compiled driving models json ref.")

openpilot/sunnypilot/sunnylink/athena/local_discovery.py

@@ -0,0 +1,261 @@
+"""
+Copyright (c) 2021-, Haibin Wen, sunnypilot, and a number of other contributors.
+
+This file is part of sunnypilot and is licensed under the MIT License.
+See the LICENSE.md file in the root directory for more details.
+"""
+from __future__ import annotations
+
+import json
+import socket
+import threading
+import time
+from collections.abc import Callable
+from dataclasses import dataclass
+
+from openpilot.common.params import Params
+from openpilot.common.swaglog import cloudlog
+
+from openpilot.sunnypilot.sunnylink.athena.local_pairing import (
+  BEACON_PREFIX,
+  DISCOVERED_APP_KEY,
+  SUNNYLINK_LOCAL_UDP_PORT,
+  format_endpoint,
+  get_local_apps,
+  pairing_requested,
+  update_local_app_endpoint,
+)
+
+LOCAL_BEACON_FRESH_S = 30
+
+@dataclass
+class AppBeacon:
+  """A parsed app beacon — the app announcing it is acting as the local backend."""
+  app_id: str
+  ws_port: int
+  source_ip: str
+
+  @property
+  def endpoint(self) -> str:
+    return format_endpoint(self.source_ip, self.ws_port)
+
+
+def parse_beacon(raw: str | bytes, source_ip: str = "") -> AppBeacon | None:
+  """
+  Parse one UDP beacon line from the app.
+
+  Wire format: `SUNNYLINK1 {"v":1,"role":"app","app_id":"<uuid>","ws_port":8443}`
+  Returns None for anything else. Beacons carry ids + addresses only — no secrets.
+  """
+  if isinstance(raw, bytes):
+    raw = raw.decode("utf-8", errors="replace")
+  raw = raw.strip()
+  if not raw.startswith(BEACON_PREFIX + " "):
+    return None
+  try:
+    data = json.loads(raw[len(BEACON_PREFIX) + 1:])
+  except ValueError:
+    return None
+  if not isinstance(data, dict):
+    return None
+  if data.get("role") != "app" or data.get("v") != 1:
+    return None
+  app_id = data.get("app_id")
+  ws_port = data.get("ws_port")
+  if not isinstance(app_id, str) or not app_id:
+    return None
+  if not isinstance(ws_port, int) or not (0 < ws_port <= 65535):
+    return None
+  return AppBeacon(app_id=app_id, ws_port=ws_port, source_ip=source_ip)
+
+
+class LocalDiscovery(threading.Thread):
+  """
+  Passive UDP listener
+
+  - While a pairing window is armed: track the freshest app beacon so the
+    daemon can offer pairing to a NEW app, and mirror it into a status param
+    for the settings UI.
+  - Independently of any window: a beacon from an app ALREADY in the paired
+    registry refreshes its cached endpoint — IPs are not identity, the app can
+    move between networks.
+  """
+
+  def __init__(self, params: Params | None = None, port: int = SUNNYLINK_LOCAL_UDP_PORT,
+               sock: socket.socket | None = None, write_interval_s: float = 5.0,
+               paired_refresh_cb: Callable[[AppBeacon], None] | None = None):
+    super().__init__(name="local_discovery_listener", daemon=True)
+    self.params = params or Params()
+    self.port = port
+    self._sock = sock
+    self.paired_refresh_cb = paired_refresh_cb
+    self._latest_endpoint: str | None = None
+    self._latest_app_id: str | None = None
+    self._last_seen_monotonic: float = 0.0
+    self._latest_paired_endpoint: str | None = None
+    self._latest_paired_app_id: str | None = None
+    self._last_paired_seen_monotonic: float = 0.0
+    self._lock = threading.Lock()
+    self._stop_event = threading.Event()
+    self.write_interval_s = write_interval_s
+    self._last_write_monotonic = 0.0
+    self._last_written_endpoint: str | None = None
+    self._last_written_app_id: str | None = None
+    self._discovered_cleared = False
+
+  def stop(self) -> None:
+    self._stop_event.set()
+    if self._sock is not None:
+      try:
+        self._sock.close()
+      except OSError:
+        pass
+
+  def latest_endpoint(self) -> str | None:
+    """The most recently announced app endpoint (None outside a pairing window)."""
+    with self._lock:
+      return self._latest_endpoint
+
+  def latest_app_id(self) -> str | None:
+    """The app_id of the most recently announced beacon (None outside a window)."""
+    with self._lock:
+      return self._latest_app_id
+
+  def last_seen_ago(self) -> float | None:
+    """Seconds since the last app beacon was heard (None when none heard yet)."""
+    with self._lock:
+      if self._last_seen_monotonic == 0.0:
+        return None
+      return time.monotonic() - self._last_seen_monotonic
+
+  def latest_paired_endpoint(self) -> str | None:
+    """The freshest beacon endpoint announced by an ALREADY-PAIRED."""
+    with self._lock:
+      return self._latest_paired_endpoint
+
+  def latest_paired_app_id(self) -> str | None:
+    """The app_id of the freshest paired-app beacon (None until one is heard)."""
+    with self._lock:
+      return self._latest_paired_app_id
+
+  def latest_paired_seen_ago(self) -> float | None:
+    """Seconds since the freshest paired-app beacon was heard (None when none)."""
+    with self._lock:
+      if self._last_paired_seen_monotonic == 0.0:
+        return None
+      return time.monotonic() - self._last_paired_seen_monotonic
+
+  def _handle(self, raw: bytes, source_ip: str) -> None:
+    beacon = parse_beacon(raw, source_ip)
+    if beacon is None:
+      return
+    if pairing_requested(self.params):
+      with self._lock:
+        self._latest_endpoint = beacon.endpoint
+        self._latest_app_id = beacon.app_id
+        self._last_seen_monotonic = time.monotonic()
+      self._write_discovered_param(beacon)
+      cloudlog.debug(f"local_discovery.app_found {beacon.app_id} at {beacon.endpoint}")
+    else:
+      with self._lock:
+        self._latest_endpoint = None
+        self._latest_app_id = None
+        self._last_seen_monotonic = 0.0
+      self._clear_discovered_param()
+    self._maybe_refresh_paired_app(beacon)
+
+  def _maybe_refresh_paired_app(self, beacon: AppBeacon) -> None:
+    """Refresh a paired app's registry endpoint from its beacon."""
+
+    if not any(app.app_id == beacon.app_id for app in get_local_apps(self.params)):
+      return
+    with self._lock:
+      self._latest_paired_endpoint = beacon.endpoint
+      self._latest_paired_app_id = beacon.app_id
+      self._last_paired_seen_monotonic = time.monotonic()
+    if update_local_app_endpoint(beacon.app_id, beacon.endpoint, self.params):
+      cloudlog.debug(f"local_discovery.paired_refresh {beacon.app_id} -> {beacon.endpoint}")
+    if self.paired_refresh_cb is not None:
+      try:
+        self.paired_refresh_cb(beacon)
+      except Exception:
+        cloudlog.exception("local_discovery.paired_refresh_cb.exception")
+
+  def _clear_discovered_param(self) -> None:
+    if self._discovered_cleared:
+      return
+    self._discovered_cleared = True
+    try:
+      self.params.remove(DISCOVERED_APP_KEY)
+    except Exception:
+      cloudlog.exception("local_discovery.param_clear.exception")
+
+  def _write_discovered_param(self, beacon: AppBeacon) -> None:
+    """Mirror the freshest beacon into a param the settings UI can read."""
+    now = time.monotonic()
+    changed = beacon.endpoint != self._last_written_endpoint or beacon.app_id != self._last_written_app_id
+    if not changed and now - self._last_write_monotonic < self.write_interval_s:
+      return
+    self._last_write_monotonic = now
+    self._last_written_endpoint = beacon.endpoint
+    self._last_written_app_id = beacon.app_id
+    self._discovered_cleared = False
+    payload = {
+      "endpoint": beacon.endpoint,
+      "app_id": beacon.app_id,
+      "ts": int(time.monotonic()),
+    }
+    try:
+      self.params.put(DISCOVERED_APP_KEY, payload, block=True)
+    except Exception:
+      cloudlog.exception("local_discovery.param_write.exception")
+
+  def _bind(self) -> socket.socket:
+    if self._sock is not None:
+      return self._sock
+    sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+    sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+    sock.bind(("0.0.0.0", self.port))
+    sock.settimeout(0.5)
+    return sock
+
+  def run(self) -> None:
+    sock = self._bind()
+    try:
+      while not self._stop_event.is_set():
+        try:
+          data, addr = sock.recvfrom(4096)
+          self._handle(data, addr[0] if len(addr) > 0 else "")
+        except TimeoutError:
+          continue
+        except OSError:
+          # Socket closed by stop() — exit quietly.
+          if self._stop_event.is_set():
+            break
+          cloudlog.exception("local_discovery.recv.exception")
+          break
+    finally:
+      try:
+        sock.close()
+      except OSError:
+        pass
+
+
+def latest_discovered_app(params: Params | None = None,
+                          fresh_s: float = LOCAL_BEACON_FRESH_S) -> tuple[str, int] | None:
+  params = params or Params()
+  data = params.get(DISCOVERED_APP_KEY)
+  if not isinstance(data, dict):
+    return None
+  endpoint = str(data.get("endpoint", ""))
+  try:
+    ts = int(data.get("ts") or 0)
+  except (ValueError, TypeError):
+    return None
+  if not endpoint or ts <= 0:
+    return None
+  age = time.monotonic() - ts
+  # Negative age = written before the last reboot (monotonic restarts at boot).
+  if age < 0 or age > fresh_s:
+    return None
+  return endpoint, max(0, int(age))

openpilot/sunnypilot/sunnylink/athena/local_pairing.py

@@ -0,0 +1,257 @@
+"""
+Copyright (c) 2021-, Haibin Wen, sunnypilot, and a number of other contributors.
+
+This file is part of sunnypilot and is licensed under the MIT License.
+See the LICENSE.md file in the root directory for more details.
+"""
+from __future__ import annotations
+
+import secrets
+import threading
+import time
+from dataclasses import asdict, dataclass
+from datetime import datetime, UTC
+from typing import Any
+from collections.abc import Callable
+
+from openpilot.common.params import Params
+from openpilot.common.swaglog import cloudlog
+
+SUNNYLINK_LOCAL_UDP_PORT = 53133
+SUNNYLINK_LOCAL_WS_PORT = 8443
+
+LOCAL_APPS_KEY = "SunnylinkLocalApps"
+PAIRING_CODE_KEY = "SunnylinkLocalPairingCode"
+PAIRING_REQUEST_KEY = "SunnylinkLocalPairingRequest"
+DISCOVERED_APP_KEY = "SunnylinkLocalDiscoveredApp"
+
+PAIRING_CODE_LENGTH = 6
+PAIRING_CODE_ALPHABET = "0123456789"
+DEFAULT_CODE_ROTATION_S = 10 * 60  # re-roll the displayed code every 10 min
+PAIRING_WINDOW_S = 5 * 60
+
+BEACON_PREFIX = "SUNNYLINK1"
+
+
+@dataclass
+class LocalApp:
+  """One app paired with this device (the app runs the local "backend")."""
+  app_id: str
+  endpoint: str
+  app_name: str = ""
+  alias: str = ""
+  paired_at: int = 0  # epoch seconds
+
+  @staticmethod
+  def from_dict(data: dict[str, Any]) -> LocalApp:
+    return LocalApp(
+      app_id=str(data.get("app_id", "")),
+      endpoint=str(data.get("endpoint", "")),
+      app_name=str(data.get("app_name", "")),
+      alias=str(data.get("alias", "")),
+      paired_at=int(data.get("paired_at") or 0),
+    )
+
+
+def local_app_display_name(app: LocalApp) -> str:
+  return app.alias or app.app_name or app.app_id
+
+
+def is_locally_paired(params: Params | None = None) -> bool:
+  return len(get_local_apps(params)) > 0
+
+
+def get_local_apps(params: Params | None = None) -> list[LocalApp]:
+  """The paired-app registry (a JSON list persisted in `SunnylinkLocalApps`)."""
+  params = params or Params()
+  data = params.get(LOCAL_APPS_KEY)
+  if not isinstance(data, list):
+    return []
+  return [LocalApp.from_dict(item) for item in data if isinstance(item, dict) and item.get("app_id")]
+
+
+def _save_local_apps(apps: list[LocalApp], params: Params | None = None) -> None:
+  params = params or Params()
+  if apps:
+    params.put(LOCAL_APPS_KEY, [asdict(app) for app in apps], block=True)
+  else:
+    params.remove(LOCAL_APPS_KEY)
+
+
+def add_local_app(app: LocalApp, params: Params | None = None) -> None:
+  """Append (or update by app_id) and persist."""
+  if not app.paired_at:
+    app.paired_at = int(datetime.now(UTC).replace(tzinfo=None).timestamp())
+  apps = [existing for existing in get_local_apps(params) if existing.app_id != app.app_id]
+  apps.append(app)
+  _save_local_apps(apps, params)
+  cloudlog.event("local_pairing.app_paired", app_id=app.app_id, endpoint=app.endpoint)
+
+
+def update_local_app_endpoint(app_id: str, endpoint: str, params: Params | None = None) -> bool:
+  """Refresh a PAIRED app's cached endpoint from its beacon."""
+  apps = get_local_apps(params)
+  for i, app in enumerate(apps):
+    if app.app_id != app_id or app.endpoint == endpoint:
+      continue
+    apps[i] = LocalApp(app_id=app.app_id, endpoint=endpoint,
+                       app_name=app.app_name, alias=app.alias, paired_at=app.paired_at)
+    _save_local_apps(apps, params)
+    cloudlog.event("local_pairing.app_endpoint_refreshed", app_id=app_id, endpoint=endpoint)
+    return True
+  return False
+
+
+def set_local_app_alias(app_id: str, alias: str, params: Params | None = None) -> bool:
+  apps = get_local_apps(params)
+  for i, app in enumerate(apps):
+    if app.app_id != app_id:
+      continue
+    if app.alias == alias:
+      return False
+    apps[i] = LocalApp(app_id=app.app_id, endpoint=app.endpoint,
+                       app_name=app.app_name, alias=alias, paired_at=app.paired_at)
+    _save_local_apps(apps, params)
+    cloudlog.event("local_pairing.app_alias_updated", app_id=app_id, alias=alias)
+    return True
+  return False
+
+
+def remove_local_app(app_id: str, params: Params | None = None) -> bool:
+  """Unpair an app by id. Returns True when an app was removed."""
+  apps = get_local_apps(params)
+  remaining = [app for app in apps if app.app_id != app_id]
+  if len(remaining) == len(apps):
+    return False
+  _save_local_apps(remaining, params)
+  cloudlog.event("local_pairing.app_unpaired", app_id=app_id)
+  return True
+
+
+def remove_all_local_apps(params: Params | None = None) -> None:
+  """Unpair every app."""
+  _save_local_apps([], params)
+  cloudlog.event("local_pairing.all_apps_unpaired")
+
+
+def generate_pairing_code() -> str:
+  """A 6-digit numeric pairing code."""
+  return "".join(secrets.choice(PAIRING_CODE_ALPHABET) for _ in range(PAIRING_CODE_LENGTH))
+
+
+def _write_pairing_code(code: str, params: Params) -> None:
+  """Persist the code with its armed-at monotonic timestamp — the window is derived from it."""
+  params.put(PAIRING_CODE_KEY, {"code": code, "ts": int(time.monotonic())}, block=True)
+
+
+def read_pairing_code(params: Params | None = None) -> str | None:
+  """The stored pairing code, or None when cleared / not yet generated."""
+  params = params or Params()
+  data = params.get(PAIRING_CODE_KEY)
+  if not isinstance(data, dict):
+    return None
+  code = data.get("code")
+  return str(code) if code else None
+
+
+def get_pairing_code(params: Params | None = None) -> str:
+  """The displayed pairing code, generating and persisting one on first use."""
+  params = params or Params()
+  code = read_pairing_code(params)
+  if code is None:
+    code = generate_pairing_code()
+    _write_pairing_code(code, params)
+  return code
+
+
+def pairing_requested(params: Params | None = None) -> bool:
+  """True while the pairing window is armed and fresh.
+
+  Self-expiring: if the code (which carries the armed-at timestamp) is missing
+  or older than PAIRING_WINDOW_S, the flag is dropped here."""
+  params = params or Params()
+  if not params.get_bool(PAIRING_REQUEST_KEY):
+    return False
+  data = params.get(PAIRING_CODE_KEY)
+  ts = data.get("ts") if isinstance(data, dict) else None
+  if not isinstance(ts, (int, float)):
+    clear_pairing_request(params)
+    return False
+  age = time.monotonic() - ts
+  # Negative age = armed before the last reboot (monotonic restarts at boot).
+  if age < 0 or age > PAIRING_WINDOW_S:
+    clear_pairing_request(params)
+    return False
+  return True
+
+
+def arm_pairing(params: Params | None = None) -> str:
+  """Arm a pairing window and return the code for the app.
+
+  Rolls a fresh code first, then sets the flag, so pairing_requested never
+  sees an armed flag without a valid code."""
+  params = params or Params()
+  code = generate_pairing_code()
+  _write_pairing_code(code, params)
+  params.put_bool(PAIRING_REQUEST_KEY, True, block=True)
+  return code
+
+
+def clear_pairing_request(params: Params | None = None) -> None:
+  """Close the pairing window: drop the request flag and the code together."""
+  params = params or Params()
+  params.remove(PAIRING_REQUEST_KEY)
+  params.remove(PAIRING_CODE_KEY)
+
+
+def verify_pairing_code(code: str, params: Params | None = None) -> bool:
+  """Constant-time check of a code typed into the app against the displayed one."""
+  params = params or Params()
+  current = read_pairing_code(params)
+  if current is None:
+    return False
+  return secrets.compare_digest(str(code).strip().upper(), current)
+
+
+class PairingCodeRotator(threading.Thread):
+  """Re-roll the displayed code while a pairing window is armed; clear it
+  otherwise — the code is never generated outside a window."""
+
+  def __init__(self, params: Params | None = None, rotation_s: float = DEFAULT_CODE_ROTATION_S,
+               stop_event: threading.Event | None = None, tick_cb: Callable[[], None] | None = None):
+    super().__init__(name="local_pairing_code_rotator", daemon=True)
+    self.params = params or Params()
+    self.rotation_s = rotation_s
+    self.stop_event = stop_event or threading.Event()
+    # Test seam: invoked once per loop iteration after state is updated.
+    self.tick_cb = tick_cb
+
+  def rotate(self) -> None:
+    """Re-roll the code while the window is armed, clear it otherwise."""
+    if pairing_requested(self.params):
+      _write_pairing_code(generate_pairing_code(), self.params)
+    else:
+      self.params.remove(PAIRING_CODE_KEY)
+
+  def run(self) -> None:
+    self.rotate()
+    while not self.stop_event.wait(self.rotation_s):
+      try:
+        self.rotate()
+        if self.tick_cb is not None:
+          self.tick_cb()
+      except Exception:
+        cloudlog.exception("local_pairing.code_rotator.exception")
+
+
+def format_endpoint(host: str, ws_port: int = SUNNYLINK_LOCAL_WS_PORT) -> str:
+  return f"ws://{host}:{ws_port}"
+
+
+def local_identity(params: Params | None = None) -> str:
+  """Identity claim on local connections. DongleId always exists on comma
+  hardware (SunnylinkDongleId is "UnregisteredDevice" until cloud
+  registration) and is what the app matches against the backend device list
+  to dedupe cloud + local entries."""
+  params = params or Params()
+  return params.get("DongleId") or params.get("HardwareSerial") or ""

openpilot/sunnypilot/sunnylink/athena/sunnylinkd.py

@@ -30,10 +30,26 @@ from websocket import (ABNF, WebSocket, WebSocketException, WebSocketTimeoutExce
 import openpilot.cereal.messaging as messaging
 from openpilot.sunnypilot.models.model_name import DEFAULT_MODEL, DEFAULT_BIG_MODEL
 from openpilot.sunnypilot.selfdrive.car.sync_sunnylink_params import update_car_list_param
+from openpilot.system.athena import rpc as rpc_module
 from openpilot.sunnypilot.sunnylink.api import SunnylinkApi
 from openpilot.sunnypilot.sunnylink.utils import sunnylink_need_register, sunnylink_ready, get_param_as_byte, save_param_from_base64_encoded_string
 from openpilot.sunnypilot.sunnylink.capabilities import generate_capabilities, CAPABILITY_LABELS
 from openpilot.sunnypilot.sunnylink.tools.generate_settings_schema import generate_schema
+from openpilot.sunnypilot.sunnylink.athena.local_discovery import LOCAL_BEACON_FRESH_S, AppBeacon, LocalDiscovery
+from openpilot.sunnypilot.sunnylink.athena.local_pairing import (
+  PAIRING_WINDOW_S,
+  LocalApp,
+  PairingCodeRotator,
+  add_local_app,
+  clear_pairing_request,
+  get_local_apps,
+  is_locally_paired,
+  local_identity,
+  pairing_requested,
+  remove_local_app,
+  set_local_app_alias,
+  verify_pairing_code,
+)
 
 SUNNYLINK_ATHENA_HOST = os.getenv('SUNNYLINK_ATHENA_HOST', 'wss://athena.sunnylink.ai')
 HANDLER_THREADS = int(os.getenv('HANDLER_THREADS', "4"))
@@ -42,6 +58,15 @@ SUNNYLINK_LOG_ATTR_NAME = "user.sunny.upload"
 SUNNYLINK_RECONNECT_TIMEOUT_S = 70  # FYI changing this will also would require a change on sidebar.cc
 DISALLOW_LOG_UPLOAD = threading.Event()
 
+LOCAL_PAIRING_SESSION_TIMEOUT_S = PAIRING_WINDOW_S
+LOCAL_PROBE_INTERVAL_S = 60
+LOCAL_ENDPOINT_BACKOFF_S = 300
+PAIRING_WATCHDOG_INTERVAL_S = 2.0
+
+_active_local_endpoint: str | None = None
+_active_ws: WebSocket | None = None
+_pairing_in_progress = threading.Event()
+
 params = Params()
 
 # Parameters that should never be remotely modified
@@ -266,44 +291,302 @@ def startLocalProxy(global_end_event: threading.Event, remote_ws_uri: str, local
   return start_local_proxy_shim(global_end_event, local_port, ws)
 
 
+@dispatcher.add_method
+def pairLocalApp(code: str, app_id: str = "", app_name: str = "", alias: str = "") -> dict[str, bool | str]:
+  """Complete pairing with the app on the CURRENT local connection."""
+  if _active_local_endpoint is None:
+    return {"success": False, "error": "not connected to a local app"}
+  if not verify_pairing_code(code):
+    cloudlog.warning("sunnylinkd.pairLocalApp.invalid_code")
+    return {"success": False, "error": "invalid code"}
+  add_local_app(LocalApp(app_id=app_id or f"app@{_active_local_endpoint}",
+                         endpoint=_active_local_endpoint, app_name=app_name, alias=alias))
+  clear_pairing_request()
+  return {"success": True}
+
+
+@dispatcher.add_method
+def updateLocalAppAlias(app_id: str, alias: str) -> dict[str, bool | str]:
+  if _active_local_endpoint is None:
+    return {"success": False, "error": "not connected to a local app"}
+  updated = set_local_app_alias(app_id, alias)
+  return {"success": True, "updated": updated}
+
+
+@dispatcher.add_method
+def unpairLocalApp(app_id: str) -> dict[str, bool | str]:
+  if _active_local_endpoint is None:
+    return {"success": False, "error": "not connected to a local app"}
+  removed = remove_local_app(app_id)
+  return {"success": True, "removed": removed}
+
+
+def _auth_header(is_local: bool) -> dict[str, str]:
+  """Bearer header for a dial."""
+  api = SunnylinkApi(params.get("SunnylinkDongleId"))
+  payload = {"identity": local_identity()} if is_local else None
+  return {"Authorization": f"Bearer {api.get_token(payload_extra=payload)}"}
+
+
+def _pairing_session(ws: WebSocket, timeout_s: float = LOCAL_PAIRING_SESSION_TIMEOUT_S) -> bool:
+  """Serve only the pairing RPCs to an app that isn't in the registry yet —
+  everything else is refused. Returns True if pairing completed (the connection may then
+  serve normally)."""
+  cloudlog.info("sunnylinkd.pairing_session.started")
+  ws.settimeout(10)
+  deadline = time.monotonic() + timeout_s
+  try:
+    while time.monotonic() < deadline and pairing_requested():
+      try:
+        raw = ws.recv()  # auto-pongs pings; blocks up to the socket timeout
+      except WebSocketTimeoutException:
+        continue
+      except Exception as e:
+        cloudlog.warning(f"sunnylinkd.pairing_session.{type(e).__name__}")
+        return is_locally_paired()
+      try:
+        msg = rpc_module.loads(raw)
+      except Exception:
+        continue
+      if not rpc_module.is_call(msg):
+        continue
+      if msg.get("method") not in ("pairLocalApp", "unpairLocalApp"):
+        continue  # refuse anything but pairing until paired
+      try:
+        ws.send(rpc_module.handle(msg, dispatcher))
+      except Exception as e:
+        cloudlog.warning(f"sunnylinkd.pairing_session.{type(e).__name__}")
+        return is_locally_paired()
+    return is_locally_paired()
+  finally:
+    ws.settimeout(SUNNYLINK_RECONNECT_TIMEOUT_S)
+
+
+def _pick_ws_uri(discovery: LocalDiscovery, backoffs: dict[str, float]) -> tuple[str, str]:
+  now = time.monotonic()
+  apps = get_local_apps()
+  app_ids = {app.app_id for app in apps}
+  if pairing_requested():
+    latest = discovery.latest_endpoint()
+    seen = discovery.last_seen_ago()
+    app_id = discovery.latest_app_id()
+    if latest is not None and seen is not None and seen <= LOCAL_BEACON_FRESH_S \
+       and app_id is not None and app_id not in app_ids \
+       and backoffs.get(latest, 0.0) <= now:
+      return latest, "pairing_offer"
+    return SUNNYLINK_ATHENA_HOST, "cloud"
+  fresh_endpoint = discovery.latest_paired_endpoint()
+  fresh_seen = discovery.latest_paired_seen_ago()
+  fresh_app_id = discovery.latest_paired_app_id()
+  fresh_ok = (fresh_endpoint is not None and fresh_seen is not None
+              and fresh_seen <= LOCAL_BEACON_FRESH_S and fresh_app_id is not None
+              and fresh_app_id in app_ids)
+  if fresh_ok and backoffs.get(fresh_endpoint, 0.0) <= now:
+    return fresh_endpoint, "paired_local"
+  for app in reversed(apps):
+    if fresh_ok and app.app_id == fresh_app_id:
+      continue
+    if backoffs.get(app.endpoint, 0.0) <= now:
+      return app.endpoint, "paired_local"
+  return SUNNYLINK_ATHENA_HOST, "cloud"
+
+
+def _probe_local_apps(active_ws: WebSocket, discovery: LocalDiscovery,
+                      backoffs: dict[str, float], stop_event: threading.Event) -> None:
+  while not stop_event.wait(LOCAL_PROBE_INTERVAL_S):
+    if pairing_requested():
+      # The pairing watchdog owns an armed window; don't migrate mid-window.
+      continue
+    now = time.monotonic()
+    candidate: str | None = None
+    for app in reversed(get_local_apps()):
+      if backoffs.get(app.endpoint, 0.0) <= now:
+        candidate = app.endpoint
+        break
+    if candidate is None:
+      continue
+    try:
+      probe = create_connection(candidate, header=_auth_header(is_local=True), timeout=10)
+      probe.close()
+    except Exception:
+      backoffs[candidate] = now + LOCAL_ENDPOINT_BACKOFF_S
+      continue
+    cloudlog.event("sunnylinkd.local_probe.reachable", endpoint=candidate)
+    try:
+      active_ws.close()
+    except Exception:
+      pass
+    break
+
+
+def _handle_paired_refresh(backoffs: dict[str, float], force_attempts: dict[str, float],
+                           beacon: AppBeacon) -> None:
+  if not any(app.app_id == beacon.app_id for app in get_local_apps()):
+    return
+  for app in get_local_apps():
+    if app.app_id == beacon.app_id:
+      backoffs.pop(app.endpoint, None)
+  backoffs.pop(beacon.endpoint, None)
+  if _pairing_in_progress.is_set():
+    return
+  if _active_local_endpoint == beacon.endpoint:
+    return
+  if _active_local_endpoint is not None:
+    return
+  now = time.monotonic()
+  if force_attempts.get(beacon.endpoint, 0.0) + LOCAL_BEACON_FRESH_S > now:
+    return
+  force_attempts[beacon.endpoint] = now
+  ws = _active_ws
+  if ws is not None:
+    cloudlog.event("sunnylinkd.paired_refresh.reconnect",
+                   app_id=beacon.app_id, endpoint=beacon.endpoint)
+    try:
+      ws.close()
+    except Exception:
+      pass
+
+
+def _pairing_watchdog(active_ws: WebSocket, backoffs: dict[str, float],
+                      stop_event: threading.Event,
+                      interval_s: float = PAIRING_WATCHDOG_INTERVAL_S) -> None:
+  """Watch for the pairing window being armed mid-session and force a
+  re-selection to the newly-discovered app."""
+  while not stop_event.wait(interval_s):
+    if not pairing_requested():
+      continue
+    cloudlog.event("sunnylinkd.pairing_watchdog.arm_detected")
+    for key in list(backoffs):
+      backoffs.pop(key, None)
+    try:
+      active_ws.close()
+    except Exception:
+      pass
+    break
+
+
 def main(exit_event: threading.Event | None = None):
   try:
     set_core_affinity([0, 1, 2, 3])
   except Exception:
     cloudlog.exception("failed to set core affinity")
 
-  while sunnylink_need_register(params):
-    cloudlog.info("Waiting for sunnylink registration to complete")
-    time.sleep(10)
+  discovery = LocalDiscovery()
+  code_rotator = PairingCodeRotator()
+  discovery.start()
+  code_rotator.start()
+
+  try:
+    _connection_loop(exit_event, discovery)
+  finally:
+    discovery.stop()
+    code_rotator.stop_event.set()
+
+
+def _serviceable(params: Params) -> bool:
+  """sunnylinkd should run when sunnylink is enabled and not on a temporary
+  fault. This deliberately includes the unregistered/unpaired state so a
+  never-registered device can still be discovered and paired over the LAN (the
+  actual session gates — registration/local pairing — are handled per
+  connection inside the loop)."""
+  return params.get_bool("SunnylinkEnabled") and not params.get_bool("SunnylinkTempFault")
+
+
+def _connection_loop(exit_event: threading.Event | None, discovery: LocalDiscovery) -> None:
+  """Local-first, cloud-fallback connection loop: a paired local endpoint
+  first, cloud when unreachable, and a pairing session to a freshly-discovered
+  app when a window is armed."""
+  global _active_local_endpoint, _active_ws
 
-  sunnylink_dongle_id = params.get("SunnylinkDongleId")
-  sunnylink_api = SunnylinkApi(sunnylink_dongle_id)
   UploadQueueCache.initialize(upload_queue)
-
   update_car_list_param()
 
-  ws_uri = f"{SUNNYLINK_ATHENA_HOST}"
   conn_start = None
   conn_retries = 0
-  while (exit_event is None or not exit_event.is_set()) and sunnylink_ready(params):
-    try:
-      if conn_start is None:
-        conn_start = time.monotonic()
+  backoffs: dict[str, float] = {}
+  force_attempts: dict[str, float] = {}
+  discovery.paired_refresh_cb = partial(_handle_paired_refresh, backoffs, force_attempts)
 
-      cloudlog.event("sunnylinkd.main.connecting_ws", ws_uri=ws_uri, retries=conn_retries)
+  while (exit_event is None or not exit_event.is_set()) and _serviceable(params):
+    ws_uri, kind = _pick_ws_uri(discovery, backoffs)
+
+    if kind == "cloud" and pairing_requested():
+      cloudlog.debug("sunnylinkd.main.pairing_waiting_for_beacon")
+      time.sleep(3)
+      continue
+
+    if kind == "cloud" and sunnylink_need_register(params):
+      cloudlog.info("Waiting for sunnylink registration or local pairing to complete")
+      time.sleep(10)
+      continue
+
+    if conn_start is None:
+      conn_start = time.monotonic()
+
+    cloudlog.event("sunnylinkd.main.connecting_ws", ws_uri=ws_uri, kind=kind, retries=conn_retries)
+    try:
       ws = create_connection(
         ws_uri,
-        header={"Authorization": f"Bearer {sunnylink_api.get_token()}"},
+        header=_auth_header(is_local=kind != "cloud"),
         enable_multithread=True,
         sslopt={"cert_reqs": ssl.CERT_NONE if "localhost" in ws_uri else ssl.CERT_REQUIRED},
         timeout=SUNNYLINK_RECONNECT_TIMEOUT_S,
       )
-      cloudlog.event("sunnylinkd.main.connected_ws", ws_uri=ws_uri, retries=conn_retries,
-                     duration=time.monotonic() - conn_start)
-      conn_start = None
+    except Exception as e:
+      if kind != "cloud":
+        backoffs[ws_uri] = time.monotonic() + LOCAL_ENDPOINT_BACKOFF_S
+      conn_retries += 1
+      params.remove("LastSunnylinkPingTime")
+      _log_connection_error(e)
+      time.sleep(backoff(conn_retries))
+      continue
 
-      conn_retries = 0
-      cur_upload_items.clear()
+    cloudlog.event("sunnylinkd.main.connected_ws", ws_uri=ws_uri, kind=kind, retries=conn_retries,
+                   duration=time.monotonic() - conn_start)
+    conn_start = None
+    conn_retries = 0
+    cur_upload_items.clear()
+    _active_ws = ws
+
+    probe_stop: threading.Event | None = None
+    watch_stop: threading.Event | None = None
+    session_endpoint: str | None = ws_uri if kind != "cloud" else None
+    try:
+      if kind == "pairing_offer":
+        _active_local_endpoint = ws_uri
+        _pairing_in_progress.set()
+        try:
+          paired_ok = _pairing_session(ws)
+        finally:
+          _pairing_in_progress.clear()
+        if not paired_ok:
+          backoffs[ws_uri] = time.monotonic() + LOCAL_ENDPOINT_BACKOFF_S
+          conn_retries += 1
+          params.remove("LastSunnylinkPingTime")
+          try:
+            ws.close()
+          except Exception:
+            pass
+          time.sleep(backoff(conn_retries))
+          continue
+        # Paired during the session — this connection may now serve normally.
+        kind = "paired_local"
+
+      if kind == "paired_local":
+        _active_local_endpoint = ws_uri
+      else:
+        _active_local_endpoint = None
+        # While on the cloud link, watch for the local app and migrate back.
+        probe_stop = threading.Event()
+        threading.Thread(target=_probe_local_apps,
+                         args=(ws, discovery, backoffs, probe_stop),
+                         name="sunnylinkd_local_probe", daemon=True).start()
+
+      # Started after any pairing session on this connection, so it can never
+      # close the connection the code is typed over.
+      watch_stop = threading.Event()
+      threading.Thread(target=_pairing_watchdog, args=(ws, backoffs, watch_stop),
+                       name="sunnylinkd_pairing_watchdog", daemon=True).start()
 
       handle_long_poll(ws, exit_event)
     except (KeyboardInterrupt, SystemExit):
@@ -311,23 +594,37 @@ def main(exit_event: threading.Event | None = None):
     except Exception as e:
       conn_retries += 1
       params.remove("LastSunnylinkPingTime")
-
-      if isinstance(e, (ConnectionError, TimeoutError, WebSocketException)):
-        cloudlog.warning(f"sunnylinkd.main.{type(e).__name__}")
-      elif isinstance(e, OSError):
-        name = errno.errorcode.get(e.errno or -1, "UNKNOWN")
-        msg = f"sunnylinkd.main.OSError.{name} ({e.errno})"
-        is_expected_error = e.errno in (errno.ENETDOWN, errno.ENETRESET, errno.ENETUNREACH)
-        cloudlog.warning(msg) if is_expected_error else cloudlog.exception(msg)
-      else:
-        cloudlog.exception("sunnylinkd.main.exception")
+      _log_connection_error(e)
+    finally:
+      if probe_stop is not None:
+        probe_stop.set()
+      if watch_stop is not None:
+        watch_stop.set()
+      if session_endpoint is not None and kind == "paired_local":
+        backoffs.pop(session_endpoint, None)
+      if _active_local_endpoint == session_endpoint:
+        _active_local_endpoint = None
+      if _active_ws is ws:
+        _active_ws = None
 
     time.sleep(backoff(conn_retries))
 
-  if not sunnylink_ready(params):
-    cloudlog.debug("Reached end of sunnylinkd.main while sunnylink is not ready. Waiting 60s before retrying")
+  if not _serviceable(params):
+    cloudlog.debug("Reached end of sunnylinkd.main while sunnylink is not serviceable. Waiting 60s before retrying")
     time.sleep(60)
 
 
+def _log_connection_error(e: Exception) -> None:
+  if isinstance(e, (ConnectionError, TimeoutError, WebSocketException)):
+    cloudlog.warning(f"sunnylinkd.main.{type(e).__name__}")
+  elif isinstance(e, OSError):
+    name = errno.errorcode.get(e.errno or -1, "UNKNOWN")
+    msg = f"sunnylinkd.main.OSError.{name} ({e.errno})"
+    is_expected_error = e.errno in (errno.ENETDOWN, errno.ENETRESET, errno.ENETUNREACH)
+    cloudlog.warning(msg) if is_expected_error else cloudlog.exception(msg)
+  else:
+    cloudlog.exception("sunnylinkd.main.exception")
+
+
 if __name__ == "__main__":
   main()

openpilot/sunnypilot/sunnylink/utils.py

@@ -2,6 +2,7 @@ import base64
 import gzip
 import json
 from openpilot.sunnypilot.sunnylink.api import SunnylinkApi, UNREGISTERED_SUNNYLINK_DONGLE_ID
+from openpilot.sunnypilot.sunnylink.athena.local_pairing import is_locally_paired
 from openpilot.common.params import Params, ParamKeyType
 from openpilot.common.version import is_prebuilt
 
@@ -16,10 +17,11 @@ def get_sunnylink_status(params=None) -> tuple[bool, bool, bool]:
 
 
 def sunnylink_ready(params=None) -> bool:
-  """Check if the device is ready to communicate with Sunnylink. That means it is enabled and registered."""
+  """Enabled and (cloud-registered or locally paired), and not on a temporary
+  fault. Local pairing makes never-registered devices usable over the LAN."""
   params = params or Params()
   is_sunnylink_enabled, is_registered, is_on_temporary_fault = get_sunnylink_status(params)
-  return is_sunnylink_enabled and is_registered and not is_on_temporary_fault
+  return is_sunnylink_enabled and (is_registered or is_locally_paired(params)) and not is_on_temporary_fault
 
 
 def use_sunnylink_uploader(params) -> bool:
@@ -28,10 +30,11 @@ def use_sunnylink_uploader(params) -> bool:
 
 
 def sunnylink_need_register(params=None) -> bool:
-  """Check if the device needs to be registered with Sunnylink."""
+  """Enabled, unregistered, and not locally paired — a locally paired device
+  works without cloud registration and must not be blocked."""
   params = params or Params()
   is_sunnylink_enabled, is_registered, is_on_temporary_fault = get_sunnylink_status(params)
-  return is_sunnylink_enabled and not is_registered and not is_on_temporary_fault
+  return is_sunnylink_enabled and not is_registered and not is_locally_paired(params) and not is_on_temporary_fault
 
 
 def register_sunnylink():