From 2f02c93b3b15d7c890ea93be76d5812543701ccd Mon Sep 17 00:00:00 2001 From: firestar5683 <168790843+firestar5683@users.noreply.github.com> Date: Fri, 3 Jul 2026 12:24:16 -0500 Subject: [PATCH] Revert "The Power Within" This reverts commit 7f314d9f288bb718b9b54ded183c227a83c49b8d. --- launch_env.sh | 4 +- .../build_detector_preservation_dataset.py | 478 ---------------- .../controls/lib/latcontrol_vehicle_tunes.py | 22 +- system/hardware/tici/agnos.json | 9 +- tools/agnos/flash_desktop_system_to_comma.sh | 138 ----- tools/agnos/patch_system_reset_image.py | 520 +++++++++--------- 6 files changed, 283 insertions(+), 888 deletions(-) delete mode 100644 scripts/speed_limit_vision/build_detector_preservation_dataset.py delete mode 100755 tools/agnos/flash_desktop_system_to_comma.sh diff --git a/launch_env.sh b/launch_env.sh index d32295607..46d2b8b6a 100755 --- a/launch_env.sh +++ b/launch_env.sh @@ -21,11 +21,11 @@ fi export QCOM_PRIORITY=12 if [ -z "$AGNOS_VERSION" ]; then - export AGNOS_VERSION="12.8.25" + export AGNOS_VERSION="12.8.18" fi if [ -z "$AGNOS_ACCEPTED_VERSIONS" ]; then - export AGNOS_ACCEPTED_VERSIONS="$AGNOS_VERSION" + export AGNOS_ACCEPTED_VERSIONS="$AGNOS_VERSION 12.8.17" fi export STAGING_ROOT="/data/safe_staging" diff --git a/scripts/speed_limit_vision/build_detector_preservation_dataset.py b/scripts/speed_limit_vision/build_detector_preservation_dataset.py deleted file mode 100644 index 08c16376c..000000000 --- a/scripts/speed_limit_vision/build_detector_preservation_dataset.py +++ /dev/null @@ -1,478 +0,0 @@ -#!/usr/bin/env python3 -from __future__ import annotations - -import argparse -import csv -import re - -from collections import Counter, defaultdict -from pathlib import Path - -import cv2 - -import starpilot.system.speed_limit_vision as slv - -if __package__ in (None, ""): - import sys - sys.path.insert(0, str(Path(__file__).resolve().parent)) - from common import DEFAULT_WORKSPACE, DETECTOR_CLASS_NAMES, REPO_ASSET_DIR, ensure_dir, resolve_workspace # type: ignore - from rebalance_detector_dataset import link_or_copy, remove_appledouble_files, safe_unlink, write_dataset_yaml, visible_file_count # type: ignore -else: - from .common import DEFAULT_WORKSPACE, DETECTOR_CLASS_NAMES, REPO_ASSET_DIR, ensure_dir, resolve_workspace - from .rebalance_detector_dataset import link_or_copy, remove_appledouble_files, safe_unlink, write_dataset_yaml, visible_file_count - - -DEFAULT_OLD_QUEUES = ( - "manual_review_queue_v1_initial10_fast", - "manual_review_queue_v2_classifier_v1", - "manual_review_queue_v3_diverse", - "manual_review_queue_v4_diverse", - "manual_review_queue_v5_diverse", - "manual_review_queue_v6_clearer", - "manual_review_queue_v7_regulatory_clear_filtered", -) -DEFAULT_BOX_QUEUE = "manual_detector_box_review_v1" -TREE_EVAL_NAME = "tree_promoted_threshold07_runtime_eval.csv" -RUNTIME_MANIFEST_NAME = "manual_review_runtime_eval_manifest.csv" -MANUAL_QUEUE_NAME = "manual_review_queue.csv" -MANUAL_LABELS_NAME = "manual_review_labels.csv" -IMPORTANT_SPEEDS = set(range(30, 70, 5)) - -SIGN_TYPE_TO_CLASS_ID = { - "regulatory": 0, - "regulatory_speed_limit": 0, - "advisory": 1, - "advisory_speed_limit": 1, - "school": 2, - "school_zone": 2, - "school_zone_speed_limit": 2, -} - - -def parse_args() -> argparse.Namespace: - parser = argparse.ArgumentParser(description="Build a detector dataset that preserves current-correct reviews while adding manual box fixes.") - parser.add_argument("--workspace", type=Path, default=DEFAULT_WORKSPACE, help="Training workspace root.") - parser.add_argument("--base-dataset", type=Path, help="Existing YOLO dataset to include before preservation samples.") - parser.add_argument("--output-root", type=Path, help="Output YOLO dataset root.") - parser.add_argument("--models-dir", type=Path, default=REPO_ASSET_DIR, help="Current tree ONNX model directory.") - parser.add_argument("--old-queue", action="append", dest="old_queues", default=[], help="Reviewed queue name to preserve. Repeatable.") - parser.add_argument("--box-queue", default=DEFAULT_BOX_QUEUE, help="Manual detector-box review queue name.") - parser.add_argument("--manual-queue", action="append", dest="manual_queues", default=[], help="Manual detector-box review queue name to include. Repeatable.") - parser.add_argument("--manual-repeat", type=int, default=2, help="Train-time repeats for manual box positives outside 30-65 mph.") - parser.add_argument("--manual-important-repeat", type=int, default=3, help="Train-time repeats for manual box positives in 30-65 mph.") - parser.add_argument("--pseudo-repeat", type=int, default=1, help="Train-time repeats for current-tree exact pseudo positives.") - parser.add_argument("--negative-repeat", type=int, default=1, help="Train-time repeats for old reviewed negatives.") - parser.add_argument("--copy", action="store_true", help="Copy images/labels instead of symlinking.") - return parser.parse_args() - - -def load_csv(path: Path) -> list[dict[str, str]]: - if not path.is_file(): - return [] - with path.open("r", encoding="utf-8", newline="") as csv_file: - return list(csv.DictReader(csv_file)) - - -def first_present(row: dict[str, str], keys: tuple[str, ...]) -> str: - for key in keys: - value = row.get(key, "").strip() - if value: - return value - return "" - - -def int_value(text: str) -> int | None: - text = text.strip() - if not text: - return None - try: - return int(float(text)) - except ValueError: - return None - - -def expected_value(row: dict[str, str]) -> int | None: - value = int_value(first_present(row, ("speed_limit_mph", "review_speed_limit_mph", "expected_speed_limit_mph", "dominant_value"))) - if value is not None: - return value - for key in ("full_detection", "model_read", "ocr_read"): - read_text = row.get(key, "").strip() - if "@" in read_text: - value = int_value(read_text.split("@", 1)[0]) - if value is not None: - return value - return None - - -def is_negative(row: dict[str, str]) -> bool: - sample_type = row.get("sample_type", "").lower() - if "negative" in sample_type: - return True - return expected_value(row) is None - - -def parse_bool(text: str) -> bool: - return text.strip().lower() in ("1", "true", "yes") - - -def class_id_from_row(row: dict[str, str]) -> int: - sign_type = first_present(row, ("review_sign_type", "detector_class")) - return SIGN_TYPE_TO_CLASS_ID.get(sign_type.strip().lower(), 0) - - -def parse_bbox(text: str) -> tuple[int, int, int, int] | None: - parts = [part.strip() for part in text.split(",")] - if len(parts) != 4: - return None - try: - x1, y1, x2, y2 = (int(round(float(part))) for part in parts) - except ValueError: - return None - if x2 <= x1 or y2 <= y1: - return None - return x1, y1, x2, y2 - - -def clamp_bbox(bbox: tuple[int, int, int, int], width: int, height: int) -> tuple[int, int, int, int] | None: - x1, y1, x2, y2 = bbox - x1 = max(min(x1, width - 1), 0) - y1 = max(min(y1, height - 1), 0) - x2 = max(min(x2, width), 0) - y2 = max(min(y2, height), 0) - if x2 <= x1 or y2 <= y1: - return None - return x1, y1, x2, y2 - - -def yolo_label(class_id: int, bbox: tuple[int, int, int, int], image_width: int, image_height: int) -> str: - x1, y1, x2, y2 = bbox - center_x = ((x1 + x2) / 2) / image_width - center_y = ((y1 + y2) / 2) / image_height - width = (x2 - x1) / image_width - height = (y2 - y1) / image_height - return f"{class_id} {center_x:.6f} {center_y:.6f} {width:.6f} {height:.6f}\n" - - -def sanitize_name(text: str) -> str: - sanitized = re.sub(r"[^A-Za-z0-9_.-]+", "_", text.strip()) - return sanitized[:180] or "sample" - - -def prepare_output(root: Path) -> None: - for split in ("train", "val"): - image_dir = ensure_dir(root / "images" / split) - label_dir = ensure_dir(root / "labels" / split) - for existing in image_dir.glob("*"): - safe_unlink(existing) - for existing in label_dir.glob("*"): - safe_unlink(existing) - - -def add_sample( - output_root: Path, - split: str, - name: str, - image_path: Path, - label_text: str, - copy_files: bool, -) -> bool: - if split not in ("train", "val"): - split = "train" - if not image_path.is_file(): - return False - - suffix = image_path.suffix or ".jpg" - image_name = f"{name}{suffix}" - label_name = f"{name}.txt" - image_dst = output_root / "images" / split / image_name - label_dst = output_root / "labels" / split / label_name - link_or_copy(image_path, image_dst, copy_files) - ensure_dir(label_dst.parent) - label_dst.write_text(label_text, encoding="utf-8") - return True - - -def include_base_dataset(base_dataset: Path, output_root: Path, copy_files: bool, stats: Counter[str]) -> None: - for split in ("train", "val"): - source_image_dir = base_dataset / "images" / split - source_label_dir = base_dataset / "labels" / split - if not source_image_dir.is_dir() or not source_label_dir.is_dir(): - continue - for image_path in sorted(source_image_dir.glob("*")): - if not image_path.is_file() or image_path.name.startswith("._"): - continue - label_path = source_label_dir / f"{image_path.stem}.txt" - if not label_path.is_file(): - continue - link_or_copy(image_path, output_root / "images" / split / image_path.name, copy_files) - link_or_copy(label_path, output_root / "labels" / split / label_path.name, copy_files) - stats[f"base_{split}"] += 1 - - -def classify_expanded_proposal( - daemon: slv.SpeedLimitVisionDaemon, - frame_bgr, - bbox: tuple[int, int, int, int], -) -> set[int]: - frame_height, frame_width = frame_bgr.shape[:2] - x1, y1, x2, y2 = bbox - box_width = x2 - x1 - box_height = y2 - y1 - reads: set[int] = set() - for expand_left, expand_top, expand_right, expand_bottom, _expansion_weight in slv.DETECTOR_CLASSIFIER_EXPANSIONS: - expanded_x1 = max(int(x1 - box_width * expand_left), 0) - expanded_y1 = max(int(y1 - box_height * expand_top), 0) - expanded_x2 = min(int(x2 + box_width * expand_right), frame_width) - expanded_y2 = min(int(y2 + box_height * expand_bottom), frame_height) - crop = frame_bgr[expanded_y1:expanded_y2, expanded_x1:expanded_x2] - if crop.size == 0: - continue - model_read = daemon._classify_speed_limit_from_model(crop) - ocr_read = daemon._read_speed_limit_from_crop(crop) - if model_read is not None: - reads.add(int(model_read[0])) - if ocr_read is not None: - reads.add(int(ocr_read[0])) - return reads - - -def select_current_tree_bbox( - daemon: slv.SpeedLimitVisionDaemon, - frame_bgr, - expected_speed: int, - target_class_id: int, -) -> tuple[int, int, int, int] | None: - proposals = daemon._collect_detector_classifier_proposals(frame_bgr) - if not proposals: - return None - - same_class = [] - exact_read = [] - for confidence, class_id, bbox in proposals: - proposal = (float(confidence), int(class_id), bbox) - if int(class_id) == target_class_id: - same_class.append(proposal) - reads = classify_expanded_proposal(daemon, frame_bgr, bbox) - if expected_speed in reads: - exact_read.append(proposal) - - for pool in ( - [proposal for proposal in exact_read if proposal[1] == target_class_id], - exact_read, - same_class, - [(float(confidence), int(class_id), bbox) for confidence, class_id, bbox in proposals], - ): - if pool: - return max(pool, key=lambda proposal: proposal[0])[2] - return None - - -def load_eval_by_record(eval_path: Path) -> dict[str, dict[str, str]]: - return {row.get("record_key", ""): row for row in load_csv(eval_path)} - - -def add_old_queue_samples( - workspace: Path, - output_root: Path, - queue_name: str, - daemon: slv.SpeedLimitVisionDaemon, - copy_files: bool, - pseudo_repeat: int, - negative_repeat: int, - stats: Counter[str], - class_stats: Counter[str], -) -> None: - queue_root = workspace / "review" / queue_name - manifest_rows = load_csv(queue_root / RUNTIME_MANIFEST_NAME) - eval_by_record = load_eval_by_record(queue_root / TREE_EVAL_NAME) - if not manifest_rows: - stats[f"{queue_name}_missing_manifest"] += 1 - return - - for row_index, row in enumerate(manifest_rows): - record_key = row.get("record_key", "") - eval_row = eval_by_record.get(record_key, {}) - split = row.get("split", "train").strip() or "train" - image_text = first_present(row, ("dataset_image", "frame_path", "source_frame")) - if not image_text: - stats[f"{queue_name}_missing_image"] += 1 - continue - image_path = Path(image_text).expanduser().resolve() - - if is_negative(row): - repeats = max(negative_repeat, 1) if split == "train" else 1 - for repeat_index in range(repeats): - name = sanitize_name(f"preserve_negative_{queue_name}_{row_index:05d}_{repeat_index}_{record_key}") - if add_sample(output_root, split, name, image_path, "", copy_files): - stats[f"old_negative_{split}"] += 1 - continue - - expected = expected_value(row) - predicted = int_value(eval_row.get("predicted_speed_limit_mph", "")) - negative_eval = parse_bool(eval_row.get("negative", "False")) - if expected is None or predicted != expected or negative_eval: - stats[f"{queue_name}_positive_not_current_exact"] += 1 - continue - - frame_bgr = cv2.imread(str(image_path)) - if frame_bgr is None: - stats[f"{queue_name}_unreadable_image"] += 1 - continue - image_height, image_width = frame_bgr.shape[:2] - class_id = class_id_from_row(row) - bbox = select_current_tree_bbox(daemon, frame_bgr, expected, class_id) - if bbox is None: - stats[f"{queue_name}_missing_pseudo_bbox"] += 1 - continue - bbox = clamp_bbox(bbox, image_width, image_height) - if bbox is None: - stats[f"{queue_name}_invalid_pseudo_bbox"] += 1 - continue - - label_text = yolo_label(class_id, bbox, image_width, image_height) - repeats = max(pseudo_repeat, 1) if split == "train" else 1 - for repeat_index in range(repeats): - name = sanitize_name(f"preserve_exact_{queue_name}_{row_index:05d}_{repeat_index}_{record_key}") - if add_sample(output_root, split, name, image_path, label_text, copy_files): - stats[f"old_pseudo_positive_{split}"] += 1 - class_stats[f"old_pseudo_{DETECTOR_CLASS_NAMES[class_id]}_{split}"] += 1 - - -def add_manual_box_samples( - workspace: Path, - output_root: Path, - queue_name: str, - copy_files: bool, - manual_repeat: int, - manual_important_repeat: int, - stats: Counter[str], - class_stats: Counter[str], -) -> None: - queue_root = workspace / "review" / queue_name - queue_by_record = {row.get("record_key", ""): row for row in load_csv(queue_root / MANUAL_QUEUE_NAME)} - manifest_by_record = {row.get("record_key", ""): row for row in load_csv(queue_root / RUNTIME_MANIFEST_NAME)} - label_rows = load_csv(queue_root / MANUAL_LABELS_NAME) - if not label_rows: - stats[f"{queue_name}_missing_labels"] += 1 - return - - for row_index, label_row in enumerate(label_rows): - record_key = label_row.get("record_key", "") - queue_row = queue_by_record.get(record_key, {}) - manifest_row = manifest_by_record.get(record_key, {}) - merged = dict(queue_row) - merged.update(manifest_row) - merged.update({key: value for key, value in label_row.items() if value.strip()}) - - if merged.get("review_status", "").strip().lower() == "ignored": - stats["manual_ignored"] += 1 - continue - - speed = expected_value(merged) - if speed is None: - stats["manual_missing_speed"] += 1 - continue - - image_text = first_present(merged, ("frame_path", "dataset_image", "source_frame")) - if not image_text: - stats["manual_missing_image"] += 1 - continue - image_path = Path(image_text).expanduser().resolve() - frame_bgr = cv2.imread(str(image_path)) - if frame_bgr is None: - stats["manual_unreadable_image"] += 1 - continue - image_height, image_width = frame_bgr.shape[:2] - - bbox = parse_bbox(first_present(merged, ("review_bbox", "bbox", "crop_bbox"))) - if bbox is None: - stats["manual_missing_bbox"] += 1 - continue - bbox = clamp_bbox(bbox, image_width, image_height) - if bbox is None: - stats["manual_invalid_bbox"] += 1 - continue - - split = merged.get("split", "train").strip() or "train" - class_id = class_id_from_row(merged) - label_text = yolo_label(class_id, bbox, image_width, image_height) - repeats = 1 - if split == "train": - repeats = max(manual_important_repeat if speed in IMPORTANT_SPEEDS else manual_repeat, 1) - for repeat_index in range(repeats): - name = sanitize_name(f"manual_box_{queue_name}_{row_index:05d}_{repeat_index}_{record_key}") - if add_sample(output_root, split, name, image_path, label_text, copy_files): - stats[f"manual_positive_{split}"] += 1 - class_stats[f"manual_{DETECTOR_CLASS_NAMES[class_id]}_{split}"] += 1 - if speed in IMPORTANT_SPEEDS: - stats[f"manual_important_{split}"] += 1 - - -def print_stats(output_root: Path, stats: Counter[str], class_stats: Counter[str]) -> None: - print(f"Created preservation detector dataset at {output_root}") - print(f"Dataset YAML: {output_root / 'dataset.yaml'}") - print(f"Train images: {visible_file_count(output_root / 'images' / 'train')}") - print(f"Val images: {visible_file_count(output_root / 'images' / 'val')}") - print("Sample stats:") - for key, count in sorted(stats.items()): - print(f" {key}: {count}") - print("Class stats:") - for key, count in sorted(class_stats.items()): - print(f" {key}: {count}") - - -def main() -> int: - args = parse_args() - workspace = resolve_workspace(args.workspace) - output_root = args.output_root.expanduser().resolve() if args.output_root else workspace / "detector_preserve_box_v1" - base_dataset = args.base_dataset.expanduser().resolve() if args.base_dataset else workspace / "detector_rebalanced_box_v1" - old_queues = tuple(args.old_queues) if args.old_queues else DEFAULT_OLD_QUEUES - - prepare_output(output_root) - stats: Counter[str] = Counter() - class_stats: Counter[str] = Counter() - - if base_dataset.is_dir(): - include_base_dataset(base_dataset, output_root, args.copy, stats) - else: - stats["missing_base_dataset"] += 1 - - models_dir = args.models_dir.expanduser().resolve() - slv.US_DETECTOR_MODEL_PATH = models_dir / "speed_limit_us_detector.onnx" - slv.US_CLASSIFIER_MODEL_PATH = models_dir / "speed_limit_us_value_classifier.onnx" - daemon = slv.SpeedLimitVisionDaemon(use_runtime=False) - - for queue_name in old_queues: - add_old_queue_samples( - workspace, - output_root, - queue_name, - daemon, - args.copy, - args.pseudo_repeat, - args.negative_repeat, - stats, - class_stats, - ) - - manual_queues = tuple(args.manual_queues) if args.manual_queues else (args.box_queue,) - for queue_name in manual_queues: - add_manual_box_samples( - workspace, - output_root, - queue_name, - args.copy, - args.manual_repeat, - args.manual_important_repeat, - stats, - class_stats, - ) - - remove_appledouble_files(output_root) - write_dataset_yaml(output_root) - print_stats(output_root, stats, class_stats) - return 0 - - -if __name__ == "__main__": - raise SystemExit(main()) diff --git a/selfdrive/controls/lib/latcontrol_vehicle_tunes.py b/selfdrive/controls/lib/latcontrol_vehicle_tunes.py index 7d1c0080b..630ae35fe 100644 --- a/selfdrive/controls/lib/latcontrol_vehicle_tunes.py +++ b/selfdrive/controls/lib/latcontrol_vehicle_tunes.py @@ -434,13 +434,6 @@ IONIQ_5_CENTER_TAPER_LAT = 0.16 IONIQ_5_CENTER_TAPER_LAT_WIDTH = 0.04 IONIQ_5_CENTER_TAPER_SPEED = 15.0 IONIQ_5_CENTER_TAPER_SPEED_WIDTH = 2.2 -IONIQ_5_SUSTAINED_TURN_IN_FF_BOOST_LEFT = 0.10 -IONIQ_5_SUSTAINED_TURN_IN_FF_BOOST_RIGHT = 0.16 -IONIQ_5_SUSTAINED_TURN_IN_FF_SPEED = 13.5 -IONIQ_5_SUSTAINED_TURN_IN_FF_SPEED_WIDTH = 1.8 -IONIQ_5_SUSTAINED_TURN_IN_FF_LAT_START = 1.10 -IONIQ_5_SUSTAINED_TURN_IN_FF_LAT_END = 3.60 -IONIQ_5_SUSTAINED_TURN_IN_FF_LAT_WIDTH = 0.30 IONIQ_EV_OLD_BASE_LAT_ACCEL_FACTOR_MULT = 1.16 IONIQ_EV_OLD_FF_REDUCTION_LEFT = 0.16 @@ -1640,26 +1633,13 @@ def get_ioniq_5_ff_scale(desired_lateral_accel: float, desired_lateral_jerk: flo turn_in_weight = max(phase, 0.0) unwind_weight = max(-phase, 0.0) low_speed_factor = _ioniq_5_low_speed_factor(v_ego) - abs_lateral_accel = abs(desired_lateral_accel) base_reduction = _ioniq_5_side_value(desired_lateral_accel, IONIQ_5_FF_REDUCTION_LEFT, IONIQ_5_FF_REDUCTION_RIGHT) * envelope turn_in_boost = 1.0 + (_ioniq_5_side_value(desired_lateral_accel, IONIQ_5_TURN_IN_BOOST_LEFT, IONIQ_5_TURN_IN_BOOST_RIGHT) * turn_in_weight * (0.35 + 0.65 * low_speed_factor)) unwind_taper = 1.0 - (_ioniq_5_side_value(desired_lateral_accel, IONIQ_5_UNWIND_TAPER_LEFT, IONIQ_5_UNWIND_TAPER_RIGHT) * unwind_weight * (0.35 + 0.65 * low_speed_factor)) - sustained_turn_in_scale = 0.0 - if desired_lateral_accel * desired_lateral_jerk > 0.0: - sustained_speed_weight = _ioniq_5_sigmoid((max(v_ego, 0.0) - IONIQ_5_SUSTAINED_TURN_IN_FF_SPEED) / - IONIQ_5_SUSTAINED_TURN_IN_FF_SPEED_WIDTH) - sustained_lat_onset = _ioniq_5_sigmoid((abs_lateral_accel - IONIQ_5_SUSTAINED_TURN_IN_FF_LAT_START) / - IONIQ_5_SUSTAINED_TURN_IN_FF_LAT_WIDTH) - sustained_lat_cutoff = _ioniq_5_sigmoid((IONIQ_5_SUSTAINED_TURN_IN_FF_LAT_END - abs_lateral_accel) / - IONIQ_5_SUSTAINED_TURN_IN_FF_LAT_WIDTH) - sustained_turn_in_scale = (_ioniq_5_side_value(desired_lateral_accel, - IONIQ_5_SUSTAINED_TURN_IN_FF_BOOST_LEFT, - IONIQ_5_SUSTAINED_TURN_IN_FF_BOOST_RIGHT) * - sustained_speed_weight * sustained_lat_onset * sustained_lat_cutoff) - return (1.0 + sustained_turn_in_scale) * (1.0 - base_reduction) * turn_in_boost * max(unwind_taper, 0.0) + return (1.0 - base_reduction) * turn_in_boost * max(unwind_taper, 0.0) def get_ioniq_5_friction_threshold(v_ego: float, desired_lateral_accel: float = 0.0, desired_lateral_jerk: float = 0.0) -> float: diff --git a/system/hardware/tici/agnos.json b/system/hardware/tici/agnos.json index 1620dc9ba..17d5f1d51 100644 --- a/system/hardware/tici/agnos.json +++ b/system/hardware/tici/agnos.json @@ -67,12 +67,13 @@ }, { "name": "system", - "url": "https://www.dropbox.com/scl/fi/1c8m8oxxbry7f3muqgymx/system5.img.xz?rlkey=81nx8one4t9912izceskbk3fg&st=z870ynme&dl=1", - "hash": "23280e1073340afe23452627a53269fbdad7251620f722df502de1a071ffd88b", - "hash_raw": "23280e1073340afe23452627a53269fbdad7251620f722df502de1a071ffd88b", + "url": "https://www.dropbox.com/scl/fi/3f8viuu3f0t29wxs8vvkj/system4.img.xz?rlkey=rtynoyi38nnbguj23ar0km42f&st=cn7crvdd&dl=1", + "hash": "ced84edefbb2e917cb60e192701920a8c5fd01014b5da640191092985804a287", + "hash_raw": "ced84edefbb2e917cb60e192701920a8c5fd01014b5da640191092985804a287", "size": 5368709120, "sparse": false, "full_check": false, - "has_ab": true + "has_ab": true, + "ondevice_hash": "242aa5adad1c04e1398e00e2440d1babf962022eb12b89adf2e60ee3068946e7" } ] diff --git a/tools/agnos/flash_desktop_system_to_comma.sh b/tools/agnos/flash_desktop_system_to_comma.sh deleted file mode 100755 index f610ce39b..000000000 --- a/tools/agnos/flash_desktop_system_to_comma.sh +++ /dev/null @@ -1,138 +0,0 @@ -#!/usr/bin/env bash -set -euo pipefail - -HOST="${1:-comma@192.168.3.110}" -IMAGE="${2:-/Users/dominickthompson/Desktop/system-23280e1073340afe23452627a53269fbdad7251620f722df502de1a071ffd88b.img.xz}" - -SCRIPT_DIR="$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")" && pwd)" -REPO_ROOT="$(cd -- "${SCRIPT_DIR}/../.." && pwd)" -SSH_KEY="${SSH_KEY:-${REPO_ROOT}/system/hardware/tici/id_rsa}" -SSH_OPTS=(-i "$SSH_KEY" -o BatchMode=yes -o StrictHostKeyChecking=no -o UserKnownHostsFile=/dev/null) - -SESSION="local_agnos_flash" -REMOTE_DIR="/data/local_agnos_flash" -REMOTE_MANIFEST="${REMOTE_DIR}/agnos-local-system.json" -REMOTE_RUNNER="${REMOTE_DIR}/run_flash.sh" -REMOTE_AGNOS="${REMOTE_DIR}/agnos.py" -PORT="8989" - -EXPECTED_VERSION="12.8.25" -RAW_HASH="23280e1073340afe23452627a53269fbdad7251620f722df502de1a071ffd88b" -RAW_SIZE="5368709120" - -if [[ ! -f "$IMAGE" ]]; then - echo "missing image: $IMAGE" >&2 - exit 1 -fi - -IMAGE_NAME="$(basename "$IMAGE")" -REMOTE_IMAGE="${REMOTE_DIR}/${IMAGE_NAME}" - -ssh "${SSH_OPTS[@]}" "$HOST" "mkdir -p '$REMOTE_DIR'" -scp "${SSH_OPTS[@]}" "$IMAGE" "$HOST:$REMOTE_IMAGE" - -LOCAL_AGNOS="$(mktemp "${TMPDIR:-/tmp}/agnos-local.XXXXXX.py")" -python3 - "$REPO_ROOT/system/hardware/tici/agnos.py" "$LOCAL_AGNOS" <<'PY' -import sys -from pathlib import Path - -src, dst = map(Path, sys.argv[1:]) -data = src.read_text(encoding="utf-8") -data = data.replace( - "import openpilot.system.updated.casync.casync as casync", - """class _UnusedCasync: - def __getattr__(self, name): - raise RuntimeError("casync support is unavailable in local AGNOS flash runner") -casync = _UnusedCasync()""", -) -dst.write_text(data, encoding="utf-8") -PY -scp "${SSH_OPTS[@]}" "$LOCAL_AGNOS" "$HOST:$REMOTE_AGNOS" -rm -f "$LOCAL_AGNOS" - -ssh "${SSH_OPTS[@]}" "$HOST" "cat > '$REMOTE_MANIFEST'" < '$REMOTE_RUNNER' && chmod +x '$REMOTE_RUNNER'" <<'REMOTE_RUNNER' -#!/usr/bin/env bash -set -euo pipefail - -: "${REMOTE_DIR:?}" -: "${REMOTE_MANIFEST:?}" -: "${REMOTE_AGNOS:?}" -: "${PORT:?}" -: "${IMAGE_NAME:?}" -: "${EXPECTED_VERSION:?}" - -exec > >(tee -a "${REMOTE_DIR}/flash.log") 2>&1 - -echo "[STEP] Local AGNOS system flash" -echo "[CHECK] Installed AGNOS: $(cat /VERSION 2>/dev/null || echo unknown)" -echo "[CHECK] Target AGNOS: ${EXPECTED_VERSION}" - -if [[ ! -f "$REMOTE_AGNOS" ]]; then - echo "[ERROR] $REMOTE_AGNOS not found" >&2 - exit 1 -fi - -if [[ -x /usr/local/venv/bin/python3 ]]; then - PYTHON_BIN="/usr/local/venv/bin/python3" -else - PYTHON_BIN="python3" -fi - -pkill -f "http.server ${PORT}.*${REMOTE_DIR}" >/dev/null 2>&1 || true -"$PYTHON_BIN" -m http.server "$PORT" --bind 127.0.0.1 --directory "$REMOTE_DIR" >"${REMOTE_DIR}/http.log" 2>&1 & -http_pid="$!" -trap 'kill "$http_pid" >/dev/null 2>&1 || true' EXIT - -http_ready=0 -for _ in $(seq 1 20); do - if "$PYTHON_BIN" - "${PORT}" "${IMAGE_NAME}" <<'PY' -import sys -import urllib.request - -port, image_name = sys.argv[1], sys.argv[2] -with urllib.request.urlopen(f"http://127.0.0.1:{port}/{image_name}", timeout=2) as resp: - resp.read(1) -PY - then - http_ready=1 - break - fi - sleep 0.25 -done - -if [[ "$http_ready" != "1" ]]; then - echo "[ERROR] Local image HTTP server did not become ready" >&2 - cat "${REMOTE_DIR}/http.log" >&2 || true - exit 1 -fi - -echo "[FLASH] Flashing local system image to inactive AGNOS slot" -PYTHONPATH="$(dirname "$REMOTE_AGNOS")" "$PYTHON_BIN" "$REMOTE_AGNOS" --swap "$REMOTE_MANIFEST" - -echo "[DONE] AGNOS flashed and slot swapped" -echo "[REBOOT] Rebooting now" -sudo reboot -REMOTE_RUNNER - -ssh "${SSH_OPTS[@]}" "$HOST" "tmux kill-session -t '$SESSION' >/dev/null 2>&1 || true" -ssh "${SSH_OPTS[@]}" "$HOST" "rm -f '$REMOTE_DIR/flash.log' '$REMOTE_DIR/http.log'" -ssh "${SSH_OPTS[@]}" "$HOST" \ - "tmux new-session -d -s '$SESSION' \"REMOTE_DIR='$REMOTE_DIR' REMOTE_MANIFEST='$REMOTE_MANIFEST' REMOTE_AGNOS='$REMOTE_AGNOS' PORT='$PORT' IMAGE_NAME='$IMAGE_NAME' EXPECTED_VERSION='$EXPECTED_VERSION' bash '$REMOTE_RUNNER'\"" - -echo "Started remote tmux session: $SESSION" -echo "Watch it with: ssh $HOST 'tmux attach -t $SESSION'" diff --git a/tools/agnos/patch_system_reset_image.py b/tools/agnos/patch_system_reset_image.py index 1e037107a..353d4ad9a 100644 --- a/tools/agnos/patch_system_reset_image.py +++ b/tools/agnos/patch_system_reset_image.py @@ -15,12 +15,8 @@ from pathlib import Path RESET_PATH_IN_IMAGE = "/usr/comma/reset" -COMMA_SH_PATH_IN_IMAGE = "/usr/comma/comma.sh" -MAGIC_PATH_IN_IMAGE = "/usr/comma/magic.py" SETUP_PATH_IN_IMAGE = "/usr/comma/setup" UPDATER_PATH_IN_IMAGE = "/usr/comma/updater" -BG_PATH_IN_IMAGE = "/usr/comma/bg.jpg" -WESTON_SERVICE_PATH_IN_IMAGE = "/lib/systemd/system/weston.service" RESET_ENTRY_IN_ZIPAPP = "openpilot/system/ui/reset.py" MICI_RESET_ENTRY_IN_ZIPAPP = "openpilot/system/ui/mici_reset.py" TICI_RESET_ENTRY_IN_ZIPAPP = "openpilot/system/ui/tici_reset.py" @@ -31,18 +27,13 @@ TICI_SETUP_ENTRY_IN_SETUP_ZIPAPP = "openpilot/system/ui/tici_setup.py" MICI_SETUP_ENTRY_IN_SETUP_ZIPAPP = "openpilot/system/ui/mici_setup.py" UPDATER_ENTRY_IN_ZIPAPP = "openpilot/system/ui/updater.py" VERSION_PATH_IN_IMAGE = "/VERSION" -PYTHON_SITE_PACKAGES_PATH_IN_IMAGE = "/usr/local/venv/lib/python3.12/site-packages" PATCH_MARKER = "STARPILOT_C4_RESET_LAYOUT_V1" +MICI_RESET_PATCH_MARKER = "STARPILOT_C4_MICI_RESET_LAYOUT_V1" +TICI_RESET_PATCH_MARKER = "STARPILOT_C4_TICI_RESET_LAYOUT_V1" APP_PATCH_MARKER = "STARPILOT_C4_RESET_APP_DIMENSIONS_V1" SETUP_WIFI_PATCH_MARKER = "JEEPNY_AVAILABLE = True" SETUP_BRANDING_PATCH_MARKER = "STARPILOT_SETUP_BRANDING_V1" SETUP_SSH_RESTORE_PATCH_MARKER = "STARPILOT_SETUP_SSH_RESTORE_V1" -WESTON_BG_PATCH_MARKER = "STARPILOT_WESTON_BG_ORIENTATION_V2" -JEEPNY_VERSION = "0.9.0" -JEEPNY_WHEEL_URL = "https://files.pythonhosted.org/packages/b2/a3/e137168c9c44d18eff0376253da9f1e9234d0239e0ee230d2fee6cea8e55/jeepney-0.9.0-py3-none-any.whl" -JEEPNY_WHEEL_SHA256 = "97e5714520c16fc0a45695e5365a2e11b81ea79bba796e26f9f1d178cb182683" -JEEPNY_PACKAGE_DIR = "jeepney" -JEEPNY_DIST_INFO_DIR = f"jeepney-{JEEPNY_VERSION}.dist-info" ANDROID_SPARSE_MAGIC = 0xED26FF3A CHUNK_TYPE_RAW = 0xCAC1 CHUNK_TYPE_FILL = 0xCAC2 @@ -67,15 +58,184 @@ DEFAULT_SYNC_COMMA_FILES = [ ] INODE_MODE_TYPE_PREFIX = { - "regular": "100", + "regular": "010", "directory": "040", - "symlink": "120", - "character": "20", - "block": "60", - "fifo": "10", - "socket": "140", + "symlink": "012", + "character": "020", + "block": "060", + "fifo": "010", + "socket": "014", } +PATCHED_RESET_SCRIPT = f"""#!/usr/bin/env python3 +import os +import sys +import threading +import time +from enum import IntEnum + +import pyray as rl + +from openpilot.system.hardware import PC +from openpilot.system.ui.lib.application import gui_app, FontWeight +from openpilot.system.ui.widgets import Widget +from openpilot.system.ui.widgets.button import Button, ButtonStyle +from openpilot.system.ui.widgets.label import gui_label, gui_text_box + +# {PATCH_MARKER} + +NVME = "/dev/nvme0n1" +USERDATA = "/dev/disk/by-partlabel/userdata" +TIMEOUT = 3 * 60 + + +class ResetMode(IntEnum): + USER_RESET = 0 + RECOVER = 1 + FORMAT = 2 + + +class ResetState(IntEnum): + NONE = 0 + CONFIRM = 1 + RESETTING = 2 + FAILED = 3 + + +def _device_type() -> str: + model_path = "/sys/firmware/devicetree/base/model" + try: + with open(model_path, "r", encoding="utf-8", errors="ignore") as f: + model = f.read().replace("\\x00", "") + return model.split("comma ")[-1].strip().lower() + except Exception: + return "" + + +def _small_layout() -> bool: + dt = _device_type() + return dt not in ("tici", "tizi") + + +class Reset(Widget): + def __init__(self, mode): + super().__init__() + self._mode = mode + self._previous_reset_state = None + self._reset_state = ResetState.NONE + self._cancel_button = Button("Cancel", self._cancel_callback) + self._confirm_button = Button("Confirm", self._confirm, button_style=ButtonStyle.PRIMARY) + self._reboot_button = Button("Reboot", lambda: os.system("sudo reboot")) + self._render_status = True + self._small = _small_layout() + + def _cancel_callback(self): + self._render_status = False + + def _do_erase(self): + if PC: + return + + os.system(f"sudo umount {{NVME}}") + os.system(f"yes | sudo mkfs.ext4 {{NVME}}") + rm = os.system("sudo rm -rf /data/*") + os.system(f"sudo umount {{USERDATA}}") + fmt = os.system(f"yes | sudo mkfs.ext4 {{USERDATA}}") + + if rm == 0 or fmt == 0: + os.system("sudo reboot") + else: + self._reset_state = ResetState.FAILED + + def start_reset(self): + self._reset_state = ResetState.RESETTING + threading.Timer(0.1, self._do_erase).start() + + def _update_state(self): + if self._reset_state != self._previous_reset_state: + self._previous_reset_state = self._reset_state + self._timeout_st = time.monotonic() + elif self._reset_state != ResetState.RESETTING and (time.monotonic() - self._timeout_st) > TIMEOUT: + exit(0) + + def _render(self, rect: rl.Rectangle): + self._update_state() + + margin = 24 if self._small else 140 + title_font = 48 if self._small else 100 + body_font = 34 if self._small else 90 + button_height = 100 if self._small else 160 + button_spacing = 20 if self._small else 50 + top_margin = 24 if self._small else 0 + body_gap = 24 if self._small else 40 + bottom_margin = 24 if self._small else 0 + + label_rect = rl.Rectangle(rect.x + margin, rect.y + top_margin, rect.width - margin * 2, title_font + 8) + gui_label(label_rect, "System Reset", title_font, font_weight=FontWeight.BOLD) + + body_bottom = rect.y + rect.height - button_height - bottom_margin - body_gap + body_top = label_rect.y + label_rect.height + body_gap + text_rect = rl.Rectangle(rect.x + margin, body_top, rect.width - margin * 2, max(0, body_bottom - body_top)) + gui_text_box(text_rect, self._get_body_text(), body_font) + + button_top = rect.y + rect.height - button_height - bottom_margin + button_width = (rect.width - button_spacing) / 2.0 + + if self._reset_state != ResetState.RESETTING: + if self._mode == ResetMode.RECOVER: + self._reboot_button.render(rl.Rectangle(rect.x, button_top, button_width, button_height)) + elif self._mode == ResetMode.USER_RESET: + self._cancel_button.render(rl.Rectangle(rect.x, button_top, button_width, button_height)) + + if self._reset_state != ResetState.FAILED: + self._confirm_button.render(rl.Rectangle(rect.x + button_width + button_spacing, button_top, button_width, button_height)) + else: + self._reboot_button.render(rl.Rectangle(rect.x, button_top, rect.width, button_height)) + + return self._render_status + + def _confirm(self): + if self._reset_state == ResetState.CONFIRM: + self.start_reset() + else: + self._reset_state = ResetState.CONFIRM + + def _get_body_text(self): + if self._reset_state == ResetState.CONFIRM: + return "Are you sure you want to reset your device?" + if self._reset_state == ResetState.RESETTING: + return "Resetting device...\\nThis may take up to a minute." + if self._reset_state == ResetState.FAILED: + return "Reset failed. Reboot to try again." + if self._mode == ResetMode.RECOVER: + return "Unable to mount data partition. Press confirm to erase and reset your device." + return "System reset triggered. Press confirm to erase all content and settings. Press cancel to resume boot." + + +def main(): + mode = ResetMode.USER_RESET + if len(sys.argv) > 1: + if sys.argv[1] == "--recover": + mode = ResetMode.RECOVER + elif sys.argv[1] == "--format": + mode = ResetMode.FORMAT + + gui_app.init_window("System Reset") + reset = Reset(mode) + + if mode == ResetMode.FORMAT: + reset.start_reset() + + for _ in gui_app.render(): + if not reset.render(rl.Rectangle(0, 0, gui_app.width, gui_app.height)): + break + + +if __name__ == "__main__": + main() +""" + + def patch_application_script(original: bytes) -> bytes: if APP_PATCH_MARKER.encode("utf-8") in original: return original @@ -130,51 +290,6 @@ def patch_setup_wifi_manager() -> bytes: return data -def patched_weston_bg_python() -> str: - som_id_path = "/sys/devices/platform/vendor/vendor:gpio-som-id/som_id" - return ( - "from PIL import Image; " - f"som=open(\"{som_id_path}\").read().strip(); " - "img=Image.open(\"/usr/comma/bg.jpg\").convert(\"RGB\"); " - "img=img.rotate(180) if som == \"1\" else img; " - "mask=img.convert(\"L\").point(lambda p: 255 if p > 16 else 0); " - "bbox=mask.getbbox(); " - "logo=img.crop(bbox) if bbox else img; " - # Pillow positive degrees are counter-clockwise; -90 pre-rotates the source clockwise. - "logo=logo.rotate(-90, expand=True); " - "resample=Image.Resampling.LANCZOS if hasattr(Image, \"Resampling\") else Image.LANCZOS; " - "logo=logo.resize((max(1, logo.width//3), max(1, logo.height//3)), resample); " - "canvas=Image.new(\"RGB\", img.size, (0, 0, 0)); " - "canvas.paste(logo, ((img.width - logo.width)//2, (img.height - logo.height)//2)); " - "canvas.save(\"/tmp/bg.jpg\")" - ) - - -def patched_weston_bg_exec_line() -> str: - python_cmd = patched_weston_bg_python().replace('"', '\\"') - return f"ExecStartPre=/bin/bash -c \"/usr/local/venv/bin/python -c '{python_cmd}'\"" - - -def patch_weston_service(original: bytes) -> bytes: - text = original.decode("utf-8") - if WESTON_BG_PATCH_MARKER in text: - return original - - old = ( - "ExecStartPre=/bin/bash -c \"/usr/local/venv/bin/python -c 'from PIL import Image; " - "img=Image.open(\\\"/usr/comma/bg.jpg\\\"); " - "(img.rotate(180) if open(\\\"/sys/devices/platform/vendor/vendor:gpio-som-id/som_id\\\").read().strip() == \\\"1\\\" else img).save(\\\"/tmp/bg.jpg\\\")'\"" - ) - new = ( - f"# {WESTON_BG_PATCH_MARKER}: displayed boot logo was 90 degrees counter-clockwise.\n" - f"{patched_weston_bg_exec_line()}" - ) - - if old not in text: - raise RuntimeError("Unable to find weston.service background image generation line") - return text.replace(old, new, 1).encode("utf-8") - - def patch_setup_branding_script(original: bytes, entry_name: str) -> bytes: text = original.decode("utf-8") if SETUP_BRANDING_PATCH_MARKER in text: @@ -308,6 +423,48 @@ def patch_reset_script() -> bytes: return data.encode("utf-8") +def patch_mici_reset_script() -> bytes: + """ + Use repo mici_reset.py (slider UX) for C4/tappity reset flow. + """ + repo_root = Path(__file__).resolve().parents[2] + src = repo_root / "system/ui/mici_reset.py" + if not src.is_file(): + raise RuntimeError(f"Unable to find repo mici_reset source: {src}") + data = src.read_text(encoding="utf-8") + if MICI_RESET_PATCH_MARKER not in data: + if data.startswith("#!"): + first_nl = data.find("\n") + if first_nl != -1: + data = data[:first_nl + 1] + f"# {MICI_RESET_PATCH_MARKER}\n" + data[first_nl + 1:] + else: + data = data + f"\n# {MICI_RESET_PATCH_MARKER}\n" + else: + data = f"# {MICI_RESET_PATCH_MARKER}\n" + data + return data.encode("utf-8") + + +def patch_tici_reset_script() -> bytes: + """ + Use repo tici_reset.py so C3/C3X reset behavior remains in sync with repo fixes. + """ + repo_root = Path(__file__).resolve().parents[2] + src = repo_root / "system/ui/tici_reset.py" + if not src.is_file(): + raise RuntimeError(f"Unable to find repo tici_reset source: {src}") + data = src.read_text(encoding="utf-8") + if TICI_RESET_PATCH_MARKER not in data: + if data.startswith("#!"): + first_nl = data.find("\n") + if first_nl != -1: + data = data[:first_nl + 1] + f"# {TICI_RESET_PATCH_MARKER}\n" + data[first_nl + 1:] + else: + data = data + f"\n# {TICI_RESET_PATCH_MARKER}\n" + else: + data = f"# {TICI_RESET_PATCH_MARKER}\n" + data + return data.encode("utf-8") + + def parse_args() -> argparse.Namespace: p = argparse.ArgumentParser(description="Patch AGNOS system image with minimal C4 reset/setup/updater fixes") p.add_argument("--manifest", default="system/hardware/tici/agnos.json", help="Path to AGNOS manifest JSON") @@ -415,20 +572,6 @@ def download(url: str, dst: Path) -> None: tmp.replace(dst) -def download_with_sha256(url: str, dst: Path, expected_sha256: str) -> None: - if not dst.exists(): - download(url, dst) - - actual_sha256 = sha256_file(dst) - if actual_sha256 != expected_sha256: - dst.unlink(missing_ok=True) - download(url, dst) - actual_sha256 = sha256_file(dst) - - if actual_sha256 != expected_sha256: - raise RuntimeError(f"Downloaded file hash mismatch for {dst}: got {actual_sha256}, expected {expected_sha256}") - - def run_cmd(cmd: list[str], check: bool = True, capture: bool = True) -> subprocess.CompletedProcess[str]: proc = subprocess.run(cmd, text=True, capture_output=capture) if check and proc.returncode != 0: @@ -574,8 +717,12 @@ def patch_reset_zipapp(original: bytes) -> bytes: changed = False replacement_reset = patch_reset_script() + replacement_tici_reset = patch_tici_reset_script() + replacement_mici_reset = patch_mici_reset_script() reset_replacements = { RESET_ENTRY_IN_ZIPAPP: replacement_reset, + TICI_RESET_ENTRY_IN_ZIPAPP: replacement_tici_reset, + MICI_RESET_ENTRY_IN_ZIPAPP: replacement_mici_reset, } with zipfile.ZipFile(src_io, "r") as src, zipfile.ZipFile(dst_io, "w", compression=zipfile.ZIP_DEFLATED) as dst: @@ -603,7 +750,7 @@ def patch_reset_zipapp(original: bytes) -> bytes: new_info.create_system = info.create_system dst.writestr(new_info, payload) - # Some reference images may not include reset.py; add missing entry explicitly. + # Some reference images may not include both reset variants; add missing entries explicitly. default_external_attr = 0o100644 << 16 for entry, payload in reset_replacements.items(): if entry in seen_entries: @@ -637,6 +784,11 @@ def patch_updater_zipapp(original: bytes) -> bytes: if payload != replacement: payload = replacement changed = True + elif info.filename == APPLICATION_ENTRY_IN_ZIPAPP: + patched_payload = patch_application_script(payload) + if patched_payload != payload: + payload = patched_payload + changed = True new_info = zipfile.ZipInfo(info.filename, info.date_time) new_info.compress_type = zipfile.ZIP_DEFLATED @@ -713,10 +865,8 @@ def zipapp_has_markers(data: bytes) -> bool: PATCH_MARKER.encode() in reset_script and b"_device_tree_device_type" in reset_script and b"gui_app.big_ui()" not in reset_script - and b"mici_setup" not in mici_reset_script - and b"jeepney" not in mici_reset_script - and b"mici_setup" not in tici_reset_script - and b"jeepney" not in tici_reset_script + and TICI_RESET_PATCH_MARKER.encode() in tici_reset_script + and MICI_RESET_PATCH_MARKER.encode() in mici_reset_script and APP_PATCH_MARKER.encode() in app_script ) @@ -748,20 +898,10 @@ def updater_zipapp_has_expected_content(data: bytes) -> bool: with zipfile.ZipFile(BytesIO(zip_payload), "r") as z: try: updater_script = z.read(UPDATER_ENTRY_IN_ZIPAPP) + app_script = z.read(APPLICATION_ENTRY_IN_ZIPAPP) except KeyError: return False - return updater_script == patch_updater_module() - - -def weston_service_has_expected_content(data: bytes) -> bool: - return ( - WESTON_BG_PATCH_MARKER.encode("utf-8") in data - and b"displayed boot logo was 90 degrees counter-clockwise" in data - and b"logo=img.crop(bbox) if bbox else img" in data - and b"logo=logo.rotate(-90, expand=True)" in data - and b"logo=logo.resize((max(1, logo.width//3), max(1, logo.height//3)), resample)" in data - and b"canvas.save(\\\"/tmp/bg.jpg\\\")" in data - ) + return updater_script == patch_updater_module() and APP_PATCH_MARKER.encode() in app_script def parse_inode(debugfs_output: str) -> int: @@ -786,115 +926,6 @@ def write_regular_file_to_image(debugfs: str, image: Path, image_path: str, loca run_debugfs(debugfs, image, f"set_inode_field <{inode}> gid {gid}", write=True) -def ensure_directory_in_image(debugfs: str, image: Path, image_path: str, mode_octal: str = "040755", uid: int = 0, gid: int = 0) -> None: - try: - run_debugfs(debugfs, image, f"mkdir {image_path}", write=True) - except Exception as e: - err = str(e).lower() - if "already exists" not in err and "file exists" not in err: - raise - - stat_out = run_debugfs(debugfs, image, f"stat {image_path}", write=False) - inode = parse_inode(stat_out) - run_debugfs(debugfs, image, f"set_inode_field <{inode}> mode {mode_octal}", write=True) - run_debugfs(debugfs, image, f"set_inode_field <{inode}> uid {uid}", write=True) - run_debugfs(debugfs, image, f"set_inode_field <{inode}> gid {gid}", write=True) - - -def extract_wheel_subset(wheel_path: Path, extract_dir: Path, roots: set[str]) -> dict[Path, str]: - if extract_dir.exists(): - shutil.rmtree(extract_dir) - extract_dir.mkdir(parents=True) - file_modes: dict[Path, str] = {} - - with zipfile.ZipFile(wheel_path, "r") as wheel: - for info in wheel.infolist(): - parts = Path(info.filename).parts - if not parts or parts[0] not in roots: - continue - if any(part == ".." for part in parts): - raise RuntimeError(f"Unsafe wheel entry path: {info.filename}") - - local_path = extract_dir.joinpath(*parts) - if info.is_dir(): - local_path.mkdir(parents=True, exist_ok=True) - continue - - local_path.parent.mkdir(parents=True, exist_ok=True) - with wheel.open(info, "r") as src, open(local_path, "wb") as dst: - shutil.copyfileobj(src, dst) - - perms = (info.external_attr >> 16) & 0o777 - if not perms: - perms = 0o644 - os.chmod(local_path, perms) - file_modes[local_path] = f"100{perms:03o}" - - if not (extract_dir / JEEPNY_PACKAGE_DIR / "__init__.py").is_file(): - raise RuntimeError("jeepney wheel extraction did not produce jeepney/__init__.py") - if not (extract_dir / JEEPNY_DIST_INFO_DIR / "METADATA").is_file(): - raise RuntimeError("jeepney wheel extraction did not produce dist-info/METADATA") - - return file_modes - - -def install_python_package_tree(debugfs: str, image: Path, source_dir: Path, image_root: str, file_modes: dict[Path, str]) -> None: - dirs = sorted((p for p in source_dir.rglob("*") if p.is_dir()), key=lambda p: len(p.relative_to(source_dir).parts)) - for local_dir in dirs: - rel = local_dir.relative_to(source_dir).as_posix() - ensure_directory_in_image(debugfs, image, f"{image_root}/{rel}", "040755", 0, 0) - - files = sorted((p for p in source_dir.rglob("*") if p.is_file()), key=lambda p: p.relative_to(source_dir).as_posix()) - for local_file in files: - rel = local_file.relative_to(source_dir).as_posix() - mode_octal = file_modes.get(local_file, "100644") - write_regular_file_to_image(debugfs, image, f"{image_root}/{rel}", local_file, mode_octal, 0, 0) - - -def install_jeepney_into_image(debugfs: str, image: Path, work_dir: Path) -> None: - wheel_dir = work_dir / "python_wheels" - wheel_path = wheel_dir / f"jeepney-{JEEPNY_VERSION}-py3-none-any.whl" - download_with_sha256(JEEPNY_WHEEL_URL, wheel_path, JEEPNY_WHEEL_SHA256) - - extract_dir = work_dir / "jeepney_wheel" - file_modes = extract_wheel_subset(wheel_path, extract_dir, {JEEPNY_PACKAGE_DIR, JEEPNY_DIST_INFO_DIR}) - - print(f"Installing jeepney {JEEPNY_VERSION} into AGNOS Python venv", flush=True) - install_python_package_tree(debugfs, image, extract_dir, PYTHON_SITE_PACKAGES_PATH_IN_IMAGE, file_modes) - - -def image_has_jeepney(debugfs: str, image: Path, work_dir: Path) -> bool: - verify_dir = work_dir / "jeepney_verify" - verify_dir.mkdir(parents=True, exist_ok=True) - init_file = verify_dir / "__init__.py" - wrappers_file = verify_dir / "wrappers.py" - metadata_file = verify_dir / "METADATA" - init_file.unlink(missing_ok=True) - wrappers_file.unlink(missing_ok=True) - metadata_file.unlink(missing_ok=True) - try: - for image_path in ( - f"{PYTHON_SITE_PACKAGES_PATH_IN_IMAGE}/{JEEPNY_PACKAGE_DIR}/__init__.py", - f"{PYTHON_SITE_PACKAGES_PATH_IN_IMAGE}/{JEEPNY_PACKAGE_DIR}/wrappers.py", - f"{PYTHON_SITE_PACKAGES_PATH_IN_IMAGE}/{JEEPNY_DIST_INFO_DIR}/METADATA", - ): - file_type, _mode, _uid, _gid = parse_debugfs_stat(run_debugfs(debugfs, image, f"stat {image_path}", write=False)) - if file_type != "regular": - raise RuntimeError(f"{image_path} has inode type {file_type}, expected regular") - - run_debugfs(debugfs, image, f"dump -p {PYTHON_SITE_PACKAGES_PATH_IN_IMAGE}/{JEEPNY_PACKAGE_DIR}/__init__.py {init_file}", write=False) - run_debugfs(debugfs, image, f"dump -p {PYTHON_SITE_PACKAGES_PATH_IN_IMAGE}/{JEEPNY_PACKAGE_DIR}/wrappers.py {wrappers_file}", write=False) - run_debugfs(debugfs, image, f"dump -p {PYTHON_SITE_PACKAGES_PATH_IN_IMAGE}/{JEEPNY_DIST_INFO_DIR}/METADATA {metadata_file}", write=False) - except Exception: - return False - - return ( - b"from .wrappers import *" in init_file.read_bytes() - and b"class DBusAddress" in wrappers_file.read_bytes() - and f"Version: {JEEPNY_VERSION}".encode("utf-8") in metadata_file.read_bytes() - ) - - def parse_debugfs_stat(debugfs_output: str) -> tuple[str, str, int, int]: type_match = re.search(r"Type:\s+([A-Za-z]+)", debugfs_output) mode_match = re.search(r"Mode:\s+([0-7]+)", debugfs_output) @@ -912,7 +943,11 @@ def inode_mode_from_type_and_perms(file_type: str, perms_octal: str) -> str: perms = perms_octal.strip() if not perms: raise RuntimeError("Empty permissions value in inode stat") - return f"{prefix}{int(perms, 8):03o}" + if not perms.startswith("0"): + perms = "0" + perms + if len(perms) < 4: + perms = perms.rjust(4, "0") + return f"{prefix}{perms}" def sync_files_from_reference_image(debugfs: str, reference_img: Path, patched_img: Path, sync_paths: list[str], work_dir: Path) -> list[str]: @@ -1066,43 +1101,48 @@ def main() -> int: synced_files = sync_files_from_reference_image(debugfs, reference_raw, patched_img, sync_paths, work_dir) print(f"Synced {len(synced_files)} /usr/comma files from reference image", flush=True) - preserved_paths = { - RESET_PATH_IN_IMAGE: "comma_reset", - SETUP_PATH_IN_IMAGE: "comma_setup", - COMMA_SH_PATH_IN_IMAGE: "comma_sh", - MAGIC_PATH_IN_IMAGE: "comma_magic", - BG_PATH_IN_IMAGE: "comma_bg", - } - preserved_hashes: dict[str, str] = {} - for image_path, label in preserved_paths.items(): - preserved_file = work_dir / f"{label}.preserved" - print(f"Recording existing {image_path} for preservation", flush=True) - run_debugfs(debugfs, patched_img, f"dump -p {image_path} {preserved_file}", write=False) - preserved_hashes[image_path] = sha256_file(preserved_file) - + original_reset = work_dir / "comma_reset.orig" + patched_reset = work_dir / "comma_reset.patched" + verify_reset = work_dir / "comma_reset.verify" + original_setup = work_dir / "comma_setup.orig" + patched_setup = work_dir / "comma_setup.patched" + verify_setup = work_dir / "comma_setup.verify" original_updater = work_dir / "comma_updater.orig" patched_updater = work_dir / "comma_updater.patched" verify_updater = work_dir / "comma_updater.verify" - original_weston = work_dir / "weston_service.orig" - patched_weston = work_dir / "weston_service.patched" - verify_weston = work_dir / "weston_service.verify" + print("Extracting /usr/comma/reset from image", flush=True) + run_debugfs(debugfs, patched_img, f"dump -p {RESET_PATH_IN_IMAGE} {original_reset}", write=False) - print("Extracting weston.service from image", flush=True) - run_debugfs(debugfs, patched_img, f"dump -p {WESTON_SERVICE_PATH_IN_IMAGE} {original_weston}", write=False) + original_data = original_reset.read_bytes() + patched_data = patch_reset_zipapp(original_data) + if patched_data == original_data: + print("Reset zipapp already contains patch marker; continuing", flush=True) + patched_reset.write_bytes(patched_data) - weston_original_data = original_weston.read_bytes() - weston_patched_data = patch_weston_service(weston_original_data) - if weston_patched_data == weston_original_data: - print("weston.service already contains the expected boot-logo patch; continuing", flush=True) - patched_weston.write_bytes(weston_patched_data) + print("Writing patched /usr/comma/reset back into image", flush=True) + write_regular_file_to_image(debugfs, patched_img, RESET_PATH_IN_IMAGE, patched_reset, "0100775", 0, 0) - print("Writing patched weston.service back into image", flush=True) - write_regular_file_to_image(debugfs, patched_img, WESTON_SERVICE_PATH_IN_IMAGE, patched_weston, "100644", 0, 0) + run_debugfs(debugfs, patched_img, f"dump -p {RESET_PATH_IN_IMAGE} {verify_reset}", write=False) + verify_data = verify_reset.read_bytes() + if not zipapp_has_markers(verify_data): + raise RuntimeError("Patched markers missing after writing reset file into image") - run_debugfs(debugfs, patched_img, f"dump -p {WESTON_SERVICE_PATH_IN_IMAGE} {verify_weston}", write=False) - verify_weston_data = verify_weston.read_bytes() - if not weston_service_has_expected_content(verify_weston_data): - raise RuntimeError("weston.service verification failed after writing weston.service file into image") + print("Extracting /usr/comma/setup from image", flush=True) + run_debugfs(debugfs, patched_img, f"dump -p {SETUP_PATH_IN_IMAGE} {original_setup}", write=False) + + setup_original_data = original_setup.read_bytes() + setup_patched_data = patch_setup_zipapp(setup_original_data) + if setup_patched_data == setup_original_data: + print("Setup zipapp already contains the expected compat patches; continuing", flush=True) + patched_setup.write_bytes(setup_patched_data) + + print("Writing patched /usr/comma/setup back into image", flush=True) + write_regular_file_to_image(debugfs, patched_img, SETUP_PATH_IN_IMAGE, patched_setup, "0100775", 0, 0) + + run_debugfs(debugfs, patched_img, f"dump -p {SETUP_PATH_IN_IMAGE} {verify_setup}", write=False) + verify_setup_data = verify_setup.read_bytes() + if not setup_zipapp_has_expected_content(verify_setup_data): + raise RuntimeError("Setup zipapp verification failed after writing setup file into image") print("Extracting /usr/comma/updater from image", flush=True) run_debugfs(debugfs, patched_img, f"dump -p {UPDATER_PATH_IN_IMAGE} {original_updater}", write=False) @@ -1114,28 +1154,18 @@ def main() -> int: patched_updater.write_bytes(updater_patched_data) print("Writing patched /usr/comma/updater back into image", flush=True) - write_regular_file_to_image(debugfs, patched_img, UPDATER_PATH_IN_IMAGE, patched_updater, "100775", 0, 0) + write_regular_file_to_image(debugfs, patched_img, UPDATER_PATH_IN_IMAGE, patched_updater, "0100775", 0, 0) run_debugfs(debugfs, patched_img, f"dump -p {UPDATER_PATH_IN_IMAGE} {verify_updater}", write=False) verify_updater_data = verify_updater.read_bytes() if not updater_zipapp_has_expected_content(verify_updater_data): raise RuntimeError("Updater zipapp verification failed after writing updater file into image") - install_jeepney_into_image(debugfs, patched_img, work_dir) - if not image_has_jeepney(debugfs, patched_img, work_dir): - raise RuntimeError("jeepney verification failed after installing package into image") - - for image_path, label in preserved_paths.items(): - verify_file = work_dir / f"{label}.verify" - run_debugfs(debugfs, patched_img, f"dump -p {image_path} {verify_file}", write=False) - if sha256_file(verify_file) != preserved_hashes[image_path]: - raise RuntimeError(f"{image_path} changed unexpectedly; protected StarPilot payload files must stay byte-identical") - if args.set_version: version_file = work_dir / "VERSION.patched" version_file.write_text(args.set_version.strip() + "\n", encoding="utf-8") print(f"Writing {VERSION_PATH_IN_IMAGE}={args.set_version.strip()}", flush=True) - write_regular_file_to_image(debugfs, patched_img, VERSION_PATH_IN_IMAGE, version_file, "100644", 0, 0) + write_regular_file_to_image(debugfs, patched_img, VERSION_PATH_IN_IMAGE, version_file, "0100644", 0, 0) version_raw = run_debugfs(debugfs, patched_img, f"cat {VERSION_PATH_IN_IMAGE}", write=False) version_lines = [ln.strip() for ln in version_raw.splitlines() if ln.strip() and not ln.startswith("debugfs ")] version_verify = version_lines[0] if version_lines else ""