From 901b93ac57df5ac8b5e7e781e955c12621ec95f5 Mon Sep 17 00:00:00 2001 From: firestar5683 <168790843+firestar5683@users.noreply.github.com> Date: Fri, 4 Sep 2026 10:46:11 -0500 Subject: [PATCH] yeetit --- scripts/model_rebuild_pipeline.py | 29 +- scripts/overnight_model_rebuild_upload.py | 249 ++++++++++++++++++ selfdrive/modeld/modeld.py | 17 +- selfdrive/modeld/tests/test_usbgpu_helpers.py | 35 +++ starpilot/assets/model_manager.py | 5 + starpilot/assets/tests/test_model_pipeline.py | 7 + 6 files changed, 334 insertions(+), 8 deletions(-) create mode 100755 scripts/overnight_model_rebuild_upload.py diff --git a/scripts/model_rebuild_pipeline.py b/scripts/model_rebuild_pipeline.py index 38bc79cc71..4a728edf66 100644 --- a/scripts/model_rebuild_pipeline.py +++ b/scripts/model_rebuild_pipeline.py @@ -22,7 +22,13 @@ DEFAULT_SOURCE_MAP = REPO_ROOT / "scripts/model_source_map_v25.json" DEFAULT_MANIFEST = DEFAULT_WORKSPACE / "manifests/model_names_v25.json" REMOTE = os.environ.get("STAR_PILOT_MODEL_REMOTE", "comma@192.168.3.110") REMOTE_ROOT = Path("/data/openpilot") -SSH_OPTIONS = ("-o", "ConnectTimeout=10", "-o", "ConnectionAttempts=1") +SSH_OPTIONS = ( + "-o", "ConnectTimeout=10", + "-o", "ConnectionAttempts=1", + "-o", "ServerAliveInterval=30", + "-o", "ServerAliveCountMax=600", +) +RSYNC_SSH = "ssh -o ConnectTimeout=10 -o ConnectionAttempts=1 -o ServerAliveInterval=30 -o ServerAliveCountMax=600" MODEL_FILENAMES = ( "driving_supercombo.onnx", @@ -238,7 +244,7 @@ def pull_remote_artifact(remote_output: str, local_output: Path) -> None: local_output.parent.joinpath(f"{local_output.name}.chunkmanifest").unlink(missing_ok=True) local_output.parent.joinpath(f"{local_output.name}.sha256").unlink(missing_ok=True) local_output.parent.mkdir(parents=True, exist_ok=True) - run(["rsync", "-az", "-e", "ssh -o ConnectTimeout=10 -o ConnectionAttempts=1", f"{REMOTE}:{remote_output}*", f"{local_output.parent}/"]) + run(["rsync", "-az", "-e", RSYNC_SSH, f"{REMOTE}:{remote_output}*", f"{local_output.parent}/"]) parts = sorted(local_output.parent.glob(f"{local_output.name}.chunk[0-9][0-9]of[0-9][0-9]")) if local_output.is_file(): @@ -280,13 +286,13 @@ def compile_model(model_id: str, source: dict, version: str, workspace: Path, fo remote_input = f"{REMOTE_ROOT}/uncompiledmodels/{model_id}" remote_output = f"{REMOTE_ROOT}/compiledmodels/{model_id}_driving_tinygrad.pkl" remote(f"rm -rf {remote_input} && mkdir -p {remote_input} {REMOTE_ROOT}/compiledmodels") - run(["rsync", "-az", "-e", "ssh -o ConnectTimeout=10 -o ConnectionAttempts=1", "--exclude=._*", f"{source_dir}/", f"{REMOTE}:{remote_input}/"]) + run(["rsync", "-az", "-e", RSYNC_SSH, "--exclude=._*", f"{source_dir}/", f"{REMOTE}:{remote_input}/"]) log_path = workspace / "logs" / f"{model_id}.log" command_parts = [ f"cd {REMOTE_ROOT} && ./models --model {model_id}", f"--input-dir {remote_input} --output-dir {REMOTE_ROOT}/compiledmodels", - f"--input-format {source['input_format']} --version {version}", + f"--input-format auto --version {version}", ] if source.get("uses_external_gpu"): command_parts.append("--external-gpu") @@ -319,7 +325,7 @@ def validate_model(model_id: str, version: str, workspace: Path) -> dict: artifact = workspace / "compiled" / f"{model_id}_driving_tinygrad.pkl" if not artifact.is_file(): raise FileNotFoundError(artifact) - run(["rsync", "-az", "-e", "ssh -o ConnectTimeout=10 -o ConnectionAttempts=1", str(artifact), f"{REMOTE}:/data/models/{artifact.name}"]) + run(["rsync", "-az", "-e", RSYNC_SSH, str(artifact), f"{REMOTE}:/data/models/{artifact.name}"]) run([ "rsync", "-az", @@ -403,6 +409,19 @@ def main() -> int: update_manifest(args.base_manifest, args.workspace, source_map) return 0 + if args.command == "compile" and args.model and args.model not in source_map: + uses_external_gpu = False + if args.base_manifest: + base = load_json(args.base_manifest) + models = base.get("models", base) + uses_external_gpu = any( + model.get("id") == args.model and model.get("uses_external_gpu", False) + for model in models + ) + source_map[args.model] = { + "input_format": "auto", + "uses_external_gpu": uses_external_gpu, + } model_ids = [args.model] if args.model else list(source_map) versions = {} if args.base_manifest: diff --git a/scripts/overnight_model_rebuild_upload.py b/scripts/overnight_model_rebuild_upload.py new file mode 100755 index 0000000000..4fc65c5de4 --- /dev/null +++ b/scripts/overnight_model_rebuild_upload.py @@ -0,0 +1,249 @@ +#!/usr/bin/env python3 +"""Compile staged v25 models on the desktop comma and publish their artifacts.""" + +from __future__ import annotations + +import json +import os +import subprocess +import sys +import time +from pathlib import Path + + +REPO_ROOT = Path(__file__).resolve().parents[1] +WORKSPACE = Path("/Volumes/agnos/StarPilot-v25-rebuild-2026-09-03") +SOURCE_MAP = REPO_ROOT / "scripts/model_source_map_v25.json" +BASE_MANIFEST = WORKSPACE / "manifests/model_names_v25.json" +READY_DIR = WORKSPACE / "ready-for-resources" +RESULTS_DIR = WORKSPACE / "results" +LOG_DIR = WORKSPACE / "logs" +REMOTE = "comma@192.168.3.111" +HF = Path.home() / ".local/bin/hf" +HF_BUCKET = "StarPilot-Driving/StarPilot-Resources" +MANIFEST_VERSION = "v25" + +SKIP_COMPILE = {"bmrlnapv6"} + + +def write_json(path: Path, payload: object) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + temporary = path.with_suffix(path.suffix + ".tmp") + temporary.write_text(json.dumps(payload, indent=2, ensure_ascii=False) + "\n") + temporary.replace(path) + + +def artifact_paths(model_id: str) -> list[Path]: + prefix = f"{model_id}_driving_tinygrad.pkl" + return sorted(path for path in READY_DIR.glob(f"{prefix}*") if path.is_file()) + + +def upload_file(path: Path, remote_path: str) -> None: + destination = f"hf://buckets/{HF_BUCKET}/{remote_path}" + for attempt in range(1, 4): + result = subprocess.run( + [str(HF), "buckets", "cp", str(path), destination, "--format", "quiet"], + text=True, + capture_output=True, + check=False, + ) + if result.returncode == 0: + return + if attempt == 3: + detail = (result.stderr or result.stdout).strip() + raise RuntimeError(f"HF upload failed for {path.name}: {detail}") + time.sleep(attempt * 10) + + +def upload_model(model_id: str) -> list[str]: + paths = artifact_paths(model_id) + if not paths: + raise FileNotFoundError(f"No ready artifact files for {model_id}") + uploaded = [] + for path in paths: + upload_file(path, f"models/{MANIFEST_VERSION}/{model_id}/{path.name}") + uploaded.append(path.name) + return uploaded + + +def run_compile(model_id: str) -> None: + environment = os.environ.copy() + environment["STAR_PILOT_MODEL_REMOTE"] = REMOTE + environment["PYTHONUNBUFFERED"] = "1" + command = [ + sys.executable, + str(REPO_ROOT / "scripts/model_rebuild_pipeline.py"), + "compile", + "--model", + model_id, + "--workspace", + str(WORKSPACE), + "--source-map", + str(SOURCE_MAP), + "--base-manifest", + str(BASE_MANIFEST), + ] + log_path = LOG_DIR / f"overnight-{model_id}.log" + with log_path.open("ab") as log: + log.write(f"\n=== START {time.strftime('%Y-%m-%dT%H:%M:%SZ', time.gmtime())} ===\n".encode()) + result = subprocess.run(command, cwd=REPO_ROOT, env=environment, stdout=log, stderr=subprocess.STDOUT) + if result.returncode: + raise RuntimeError(f"Compilation failed; see {log_path}") + + +def remote_compile_ids() -> set[str]: + """Return model IDs currently compiling on the device. + + SSH can time out while the remote process survives. Checking the device + before starting another model prevents two compiles from sharing the GPU. + """ + result = subprocess.run( + [ + "ssh", + "-o", + "ConnectTimeout=10", + "-o", + "ConnectionAttempts=1", + "-o", + "ServerAliveInterval=30", + "-o", + "ServerAliveCountMax=600", + REMOTE, + "pgrep -af '[c]ompile_modeld.py' || true", + ], + text=True, + capture_output=True, + timeout=20, + check=False, + ) + if result.returncode and not result.stdout: + raise RuntimeError(f"Could not inspect remote compiler: {result.stderr.strip()}") + model_ids = set() + for line in result.stdout.splitlines(): + marker = "/compiledmodels/" + suffix = "_driving_tinygrad.pkl" + if marker in line and suffix in line: + model_ids.add(line.split(marker, 1)[1].split(suffix, 1)[0].split()[0]) + return model_ids + + +def wait_for_remote_idle() -> set[str]: + active = remote_compile_ids() + while active: + print(f" waiting for remote compiler(s): {', '.join(sorted(active))}", flush=True) + time.sleep(30) + active = remote_compile_ids() + return active + + +def recover_remote_artifact(model_id: str) -> bool: + """Pull a valid artifact left behind after an SSH disconnect.""" + local_output = WORKSPACE / "compiled" / f"{model_id}_driving_tinygrad.pkl" + os.environ["STAR_PILOT_MODEL_REMOTE"] = REMOTE + from model_rebuild_pipeline import pull_remote_artifact, stage_ready_artifact + + try: + pull_remote_artifact( + f"/data/openpilot/compiledmodels/{model_id}_driving_tinygrad.pkl", + local_output, + ) + except (FileNotFoundError, ValueError, subprocess.CalledProcessError, OSError): + return False + local_output.chmod(0o644) + stage_ready_artifact(local_output, WORKSPACE) + return True + + +def prepare_model(model_id: str) -> None: + """Recover a completed orphan first, otherwise wait before compiling.""" + active = remote_compile_ids() + if not active: + if recover_remote_artifact(model_id): + print(f" recovered completed remote artifact for {model_id}", flush=True) + return + print(f" remote compile detected before {model_id}: {', '.join(sorted(active))}", flush=True) + wait_for_remote_idle() + if model_id in active and recover_remote_artifact(model_id): + print(f" recovered {model_id} after remote SSH disconnect", flush=True) + + +def recover_after_failure(model_id: str) -> bool: + """Recover a valid remote result instead of launching a concurrent retry.""" + active = remote_compile_ids() + if active: + print(f" compile session ended locally; waiting on remote: {', '.join(sorted(active))}", flush=True) + wait_for_remote_idle() + return recover_remote_artifact(model_id) + + +def main() -> int: + if not BASE_MANIFEST.is_file(): + raise FileNotFoundError(f"Expected current HF manifest at {BASE_MANIFEST}") + source_map = json.loads(SOURCE_MAP.read_text()) + manifest = json.loads(BASE_MANIFEST.read_text()) + manifest_models = manifest.get("models", manifest) + manifest_by_id = {model["id"]: model for model in manifest_models} + manifest_ids = set(manifest_by_id) + staged_ids = sorted( + path.name for path in (WORKSPACE / "onnx").iterdir() + if path.is_dir() and any(path.glob("*.onnx")) + ) + requested_ids = { + model_id.strip() for model_id in os.environ.get("STAR_PILOT_MODEL_IDS", "").split(",") + if model_id.strip() + } + if requested_ids: + staged_ids = [model_id for model_id in staged_ids if model_id in requested_ids] + for model_id in staged_ids: + source_map.setdefault(model_id, { + "input_format": "auto", + "uses_external_gpu": bool(manifest_by_id.get(model_id, {}).get("uses_external_gpu")), + }) + missing_from_manifest = sorted(set(staged_ids) - manifest_ids) + if missing_from_manifest: + raise ValueError(f"Staged IDs missing from v25 manifest: {', '.join(missing_from_manifest)}") + + status_path = RESULTS_DIR / "overnight_status.json" + status = { + "remote": REMOTE, + "manifest": f"models/{MANIFEST_VERSION}", + "started_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), + "models": {}, + "unresolved_sources": ["berightthere", "rdf53", "rdf63"], + } + write_json(status_path, status) + + for model_id in staged_ids: + if model_id in SKIP_COMPILE: + status["models"][model_id] = {"status": "skipped_existing_artifact"} + write_json(status_path, status) + continue + try: + artifact = WORKSPACE / "compiled" / f"{model_id}_driving_tinygrad.pkl" + if not artifact.is_file(): + prepare_model(model_id) + if not artifact.is_file(): + try: + run_compile(model_id) + except Exception: + if not recover_after_failure(model_id): + raise + uploaded = upload_model(model_id) + status["models"][model_id] = {"status": "compiled_uploaded", "files": uploaded} + except Exception as error: + status["models"][model_id] = {"status": "failed", "error": str(error)} + write_json(status_path, status) + + try: + upload_file(BASE_MANIFEST, f"manifests/model_names_{MANIFEST_VERSION}.json") + status["manifest_status"] = "uploaded" + except Exception as error: + status["manifest_status"] = "failed" + status["manifest_error"] = str(error) + status["finished_at"] = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()) + write_json(status_path, status) + return 0 if status.get("manifest_status") == "uploaded" else 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/selfdrive/modeld/modeld.py b/selfdrive/modeld/modeld.py index cf18f20529..a832a25cdb 100755 --- a/selfdrive/modeld/modeld.py +++ b/selfdrive/modeld/modeld.py @@ -313,10 +313,15 @@ def _select_builtin_model(params: Params) -> None: def _close_tinygrad_disk_cache_connection() -> None: - """Drop tinygrad's process-global cache connection before loading the next model.""" + """Close tinygrad's cache connection without replacing its thread-local holder.""" import tinygrad.helpers as tinygrad_helpers - connection = getattr(tinygrad_helpers, "_db_connection", None) + holder = getattr(tinygrad_helpers, "_db_connection", None) + if holder is None: + return + + has_thread_local_connection = hasattr(holder, "conn") + connection = getattr(holder, "conn", holder if hasattr(holder, "close") else None) if connection is None: return @@ -325,7 +330,13 @@ def _close_tinygrad_disk_cache_connection() -> None: except Exception: cloudlog.exception("failed to close tinygrad disk cache connection") finally: - tinygrad_helpers._db_connection = None + if has_thread_local_connection: + try: + del holder.conn + except AttributeError: + pass + else: + tinygrad_helpers._db_connection = None def get_action_from_model(model_output: dict[str, np.ndarray], prev_action: log.ModelDataV2.Action, diff --git a/selfdrive/modeld/tests/test_usbgpu_helpers.py b/selfdrive/modeld/tests/test_usbgpu_helpers.py index 80e1858700..f7ca932532 100644 --- a/selfdrive/modeld/tests/test_usbgpu_helpers.py +++ b/selfdrive/modeld/tests/test_usbgpu_helpers.py @@ -171,6 +171,41 @@ def test_tinygrad_disk_cache_connection_is_closed_between_models(monkeypatch): assert tinygrad_helpers._db_connection is None +def test_tinygrad_thread_local_cache_holder_survives_cleanup(monkeypatch): + import threading + import tinygrad.helpers as tinygrad_helpers + + class FakeConnection: + def __init__(self): + self.closed = False + + def close(self): + self.closed = True + + holder = threading.local() + connection = FakeConnection() + holder.conn = connection + monkeypatch.setattr(tinygrad_helpers, "_db_connection", holder) + + modeld._close_tinygrad_disk_cache_connection() + + assert connection.closed + assert tinygrad_helpers._db_connection is holder + assert not hasattr(holder, "conn") + + +def test_tinygrad_empty_thread_local_cache_holder_is_safe(monkeypatch): + import threading + import tinygrad.helpers as tinygrad_helpers + + holder = threading.local() + monkeypatch.setattr(tinygrad_helpers, "_db_connection", holder) + + modeld._close_tinygrad_disk_cache_connection() + + assert tinygrad_helpers._db_connection is holder + + def test_external_gpu_load_finishes_before_native_model_can_start(monkeypatch): calls = [] diff --git a/starpilot/assets/model_manager.py b/starpilot/assets/model_manager.py index 006d7c9fc4..2a66b636ca 100644 --- a/starpilot/assets/model_manager.py +++ b/starpilot/assets/model_manager.py @@ -36,6 +36,10 @@ ARTIFACT_URLS_CACHE = ".model_artifact_urls.json" ARTIFACT_METADATA_CACHE = ".model_artifacts.json" MODEL_KEY_CANONICAL_MAP = { "sc": "sc2", + "napv1": "remove-avgpoolv1", + "napv2": "remove-avgpoolv2", + "napv3": "remove-avgpoolv3", + "napv4": "remove-avgpoolv4", # The original bundled RDF key remains valid after the bundled default moves # to the v23 RDF V4 artifact. "rdf": DEFAULT_MODEL_KEY, @@ -738,6 +742,7 @@ class ModelManager: def _download_model(self, model_to_download: str, allow_gpu_without_gpu: bool): self.downloading_model = True + model_to_download = self._canonical_model_key(model_to_download) if is_builtin_model_key(model_to_download): self.params_memory.put(DOWNLOAD_PROGRESS_PARAM, "Built-in model already downloaded.") diff --git a/starpilot/assets/tests/test_model_pipeline.py b/starpilot/assets/tests/test_model_pipeline.py index c3b5fd5f86..7776e0f178 100644 --- a/starpilot/assets/tests/test_model_pipeline.py +++ b/starpilot/assets/tests/test_model_pipeline.py @@ -57,6 +57,13 @@ def test_old_manifest_ids_resolve_to_v23_namespace(): assert manager._resolve_manifest_model_key("missing") == "missing" +def test_old_remove_avgpool_ids_resolve_to_artifact_ids(): + manager = object.__new__(ModelManager) + manager.available_models = ["remove-avgpoolv4"] + assert manager._resolve_manifest_model_key("napv4") == "remove-avgpoolv4" + assert model_manager.canonical_model_key("napv4") == "remove-avgpoolv4" + + def test_model_cleanup_matches_legacy_split_artifacts(): assert model_manager.is_driving_artifact_file("pop223_driving_tinygrad.pkl") assert model_manager.is_driving_artifact_file("driving_vision_tinygrad.pkl")