#!/usr/bin/env python3 # # Copyright (C) 2025 The Android Open Source Project # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. """Automates boot image profile generation for Android Automotive. This script orchestrates the process of generating a boot image profile by performing the following main stages: 1. Configuring a device using art/tools/boot-image-profile-configure-device.sh to generate a `boot.zip` file. 2. Executing a simple set of activities (Maps, Play Store) on the Android device via ADB shell commands to generate usage profiles. Note: This "Starting Maps and Play activities" is a simple CUJ implementation as a starting point. It may need to be expanded or customized for better profiling by reflecting real usage. 3. Generating boot image profiles based on the generated `boot.zip` file and the collected device profiles, using art/tools/boot-image-profile-generate.sh. The script requires the ANDROID_BUILD_TOP environment variable to be set. Example Usage: $ source build/envsetup.sh # Or equivalent setup $ python3 art/tools/boot-image-profile-aaos-sample-generate.py Example Output Trace: --- Stage 1: Configuring Device --- --- Stage 2: Starting ADB/CUJ Sequence --- --- Stage 3: Running Generation Script --- --- All stages completed successfully! --- """ import os import shlex import subprocess import sys import time from typing import List, Optional # --- Helper Functions --- def print_message(message: str) -> None: """Prints a message to stderr.""" print(f"{message}", file=sys.stderr) def print_separator() -> None: """Prints a separator line.""" print_message("---------------------------------------\n") # --- End Helper Functions --- # --- Configuration --- # Pattern to look for in logcat to indicate the device is ready for profiling. # Set to None to skip the logcat check and only wait for ADB device. LOGCAT_READY_PATTERN: Optional[str] = ( "Displayed com.android.car.carlauncher/.CarLauncher" ) # LOGCAT_READY_PATTERN = None # Example: Disable logcat check # Logcat filters to apply when monitoring for the ready pattern. LOGCAT_FILTER_SPECS: List[str] = [ "ActivityTaskManager:I", "*:S", ] # Ensure ANDROID_BUILD_TOP is set and resolve its absolute path. ANDROID_BUILD_TOP: str = os.getenv("ANDROID_BUILD_TOP", "") if not ANDROID_BUILD_TOP: print_message( "Error: ANDROID_BUILD_TOP environment variable is not set. " "Please run `source build/envsetup.sh` and select a target." ) sys.exit(1) ANDROID_BUILD_TOP = os.path.abspath(ANDROID_BUILD_TOP) print(f"--- Using ANDROID_BUILD_TOP: {ANDROID_BUILD_TOP} ---") # Define paths to required scripts and generated files. _SCRIPT_DIR = "art/tools" # Relative path to scripts within build top _FRAMEWORKS_CONFIG_DIR = "frameworks/base/config" # Relative path to denylist SCRIPT1_PATH: str = os.path.join( ANDROID_BUILD_TOP, _SCRIPT_DIR, "boot-image-profile-configure-device.sh" ) SCRIPT1_ARG: str = "boot.zip" SCRIPT2_PATH: str = os.path.join( ANDROID_BUILD_TOP, _SCRIPT_DIR, "boot-image-profile-generate.sh" ) BOOT_ZIP_ARG: str = os.path.join(ANDROID_BUILD_TOP, "boot.zip") DENYLIST_ARG: str = os.path.join( ANDROID_BUILD_TOP, _FRAMEWORKS_CONFIG_DIR, "preloaded-classes-denylist" ) MAPS_PROF_ARG: str = os.path.join(ANDROID_BUILD_TOP, "maps.prof") VENDING_PROF_ARG: str = os.path.join(ANDROID_BUILD_TOP, "vending.prof") # Arguments for the generation script (SCRIPT2). SCRIPT2_ARGS_LIST: List[str] = [ ANDROID_BUILD_TOP, BOOT_ZIP_ARG, DENYLIST_ARG, MAPS_PROF_ARG, VENDING_PROF_ARG, "--profman-arg", "--upgrade-startup-to-hot=false", ] # Package names and activities for the CUJ steps. MAPS_PACKAGE: str = ( "com.google.android.apps.maps/com.google.android.maps.MapsActivity" ) VENDING_PACKAGE: str = ( "com.android.vending/com.google.android.finsky.activities.MainActivity" ) # Duration to sleep after a CUJ activity is launched, to allow it to run # before taking a snapshot of the profile. CUJ_ACTIVITY_SLEEP_SECONDS: int = 15 # Duration to sleep after starting logcat, to allow it to initialize. LOGCAT_START_SLEEP_SECONDS: float = 0.5 # --- End Configuration --- def run_generic_command( command_list: List[str], check: bool = True, capture: bool = True, cmd_timeout: Optional[int | float] = None, **kwargs, ) -> Optional[subprocess.CompletedProcess]: """Runs a command, handling errors and timeouts. Args: command_list: A list of strings representing the command and its arguments. check: If True, raise CalledProcessError on non-zero exit code. If False, return the CompletedProcess object regardless of exit code. capture: If True, capture stdout/stderr. If False, let them pass through. cmd_timeout: Timeout in seconds for the command. **kwargs: Additional arguments to pass to subprocess.run. Returns: A CompletedProcess object or None on error. Raises: subprocess.CalledProcessError: If check is True and the command fails. subprocess.TimeoutExpired: If the command times out and check is True. """ command_str = " ".join(shlex.quote(part) for part in command_list) print_message(f"--- Running command: {command_str} ---") try: result = subprocess.run( command_list, check=check, stdout=subprocess.PIPE if capture else None, stderr=subprocess.PIPE if capture else None, text=True, # Use text mode for string I/O encoding="utf-8", # Explicitly set encoding errors="replace", # Replace uninterpretable characters timeout=cmd_timeout, **kwargs, ) # If check=True, subprocess.run already raises on error. # If check=False, we just return the result. return result except FileNotFoundError: print_message( f"Error: Command not found: '{command_list[0]}'. Is it in your PATH?" ) except subprocess.TimeoutExpired: print_message( f"Error: Command '{command_str}' timed out after {cmd_timeout} seconds." ) if check: raise except subprocess.CalledProcessError as e: print_message( f"Error: Command '{command_str}' failed with exit code {e.returncode}." ) if e.stdout: print_message(f"--- stdout ---\n{e.stdout.strip()}") if e.stderr: print_message(f"--- stderr ---\n{e.stderr.strip()}") if check: raise except Exception as e: print_message(f"Unexpected Python error running {command_str}: {e}") return None # Return None on any caught exception def run_shell_script( script_path: str, *script_args: str, interpreter: str = "bash" ) -> bool: """Runs a shell script using the specified interpreter. Args: script_path: The path to the shell script. *script_args: Positional arguments to pass to the script. interpreter: The shell interpreter to use (e.g., "bash", "sh"). Returns: True if the script exited with code 0, False otherwise or if an error occurred running the command. """ command = [interpreter, script_path] + list(script_args) print_message(f"--- Running script: {script_path} ---") # Run with check=False so we can inspect the return code manually result = run_generic_command(command, check=False, capture=True) if result is None: # An exception occurred in run_generic_command print_message(f"--- Script {script_path} command execution failed ---") print_separator() return False # Output captured stdout/stderr for user information if result.stdout: print_message(f"--- stdout ---\n{result.stdout.strip()}") if result.stderr: print_message(f"--- stderr ---\n{result.stderr.strip()}") if result.returncode == 0: print_message(f"--- Script {script_path} SUCCESS ---") print_separator() return True print_message( f"--- Script {script_path} FAILED with exit code {result.returncode} ---" ) print_separator() return False def wait_for_device_ready( timeout_seconds: int = 360, check_interval: int = 1, logcat_pattern: Optional[str] = LOGCAT_READY_PATTERN, logcat_filters: List[str] = LOGCAT_FILTER_SPECS, ) -> bool: """Waits for ADB connection and optionally a specific logcat pattern. Args: timeout_seconds: Maximum time in seconds to wait. check_interval: Time in seconds to wait between logcat reads. logcat_pattern: The string or regex pattern to look for in logcat. If None, only waits for ADB device connection. logcat_filters: List of logcat filter specifications (e.g., "TAG:LEVEL"). Returns: True if the device becomes ready within the timeout, False otherwise. """ print_message(f"--- Waiting for ADB device (max {timeout_seconds}s) ---") start_time = time.time() print_message("[Wait Step 1/2] Waiting for ADB connection...") adb_wait_cmd = ["adb", "wait-for-device"] try: # Using run here is simpler as wait-for-device holds until ready or timeout subprocess.run( adb_wait_cmd, timeout=timeout_seconds, capture_output=True, text=True, check=True, # check=True is appropriate here encoding="utf-8", errors="replace", ) elapsed_time = time.time() - start_time print_message( "[Wait Step 1/2] Complete: Device connected. (Elapsed:" f" {elapsed_time:.1f}s)" ) except ( subprocess.TimeoutExpired, subprocess.CalledProcessError, FileNotFoundError, ) as e: print_message(f"ERROR: Initial ADB connection failed: {e}") print_message("Is device connected? Is adb in PATH?") print_separator() return False except Exception as e: print_message(f"ERROR during 'adb wait-for-device': {e}") print_separator() return False if not logcat_pattern: print_message( "[Wait Step 2/2] Skipping logcat pattern check" " (LOGCAT_READY_PATTERN is None)." ) print_message("--- Device ready based on ADB connection ---") print_separator() return True logcat_proc: Optional[subprocess.Popen] = None logcat_found: bool = False logcat_failed: bool = False print_message( "\n[Wait Step 2/2] Starting logcat check for pattern" f" '{logcat_pattern}'..." ) try: print_message(" Clearing logcat buffer...") run_generic_command(["adb", "logcat", "-c"], check=False) logcat_cmd = ["adb", "logcat", "-v", "brief"] + logcat_filters print_message( f" Starting logcat: {' '.join(shlex.quote(p) for p in logcat_cmd)}" ) try: # Use Popen to read line by line logcat_proc = subprocess.Popen( logcat_cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, encoding="utf-8", errors="replace", ) time.sleep(LOGCAT_START_SLEEP_SECONDS) except (FileNotFoundError, Exception) as proc_err: print_message(f"ERROR: Failed to start logcat process: {proc_err}") logcat_failed = True if not logcat_failed: while time.time() - start_time < timeout_seconds: # Check if logcat process died unexpectedly if logcat_proc and logcat_proc.poll() is not None: print_message( " Error: logcat process terminated unexpectedly (Code:" f" {logcat_proc.returncode})." f" Stderr:\n{logcat_proc.stderr.read()}" ) logcat_failed = True break # Attempt to read line. readline() can block, but the outer # timeout loop will eventually catch it. try: # Check if stdout is not None before reading if logcat_proc and logcat_proc.stdout: line = logcat_proc.stdout.readline() if not line: # If readline returns empty string, process might be exiting # or stream is just slow. Check poll() again or wait. if logcat_proc.poll() is not None: print_message(" logcat stream ended.") logcat_failed = True break # Wait a bit if no line was read but process is still running time.sleep(check_interval) continue line = line.strip() if not line: continue # Skip empty lines if logcat_pattern in line: print_message(f' Found target logcat line: "{line}"') print_message( "[Wait Step 2/2] SUCCESS: Found logcat pattern. (Elapsed:" f" {time.time() - start_time:.1f}s)" ) logcat_found = True break except Exception as read_err: print_message(f" Warning: Error reading logcat stream: {read_err}.") # Continue the loop after a short pause time.sleep(check_interval) finally: # Ensure the logcat subprocess is terminated if logcat_proc and logcat_proc.poll() is None: print_message(" Terminating logcat process...") try: # Give it a moment to terminate gracefully logcat_proc.terminate() logcat_proc.wait(timeout=2) except subprocess.TimeoutExpired: # If it doesn't terminate, kill it print_message(" Logcat process did not terminate, killing...") logcat_proc.kill() except Exception as term_err: # Catch any other errors during termination print_message(f" Error during logcat termination: {term_err}") if logcat_found: print_message("--- Device ready based on logcat pattern ---") print_separator() return True else: print_message( f"--- Logcat pattern '{logcat_pattern}' not found within timeout or" " logcat failed. ---" ) print_separator() return False def execute_adb_step( cmd_list: List[str], step_num: int, description: str, allow_fail: bool = False, ) -> bool: """Executes a single ADB shell command step. Args: cmd_list: The list representing the ADB command and its arguments. step_num: The sequential number of this step. description: A brief description of the step. allow_fail: If True, the function returns True even if the command fails (non-zero exit code). If False, it returns False on failure. Returns: True if the step completed successfully or allow_fail is True and the command ran (even if it failed). False if the command execution itself failed (e.g., command not found, timeout) or if allow_fail is False and the command returned a non-zero exit code. """ print_message(f"\n--- ADB Step {step_num}: {description} ---") result = run_generic_command(cmd_list, check=False, capture=True) if result is None: print_message( f"ERROR: Step {step_num} ({description}) command execution failed." ) print_separator() return False # Decode stdout and stderr if they exist cmd_stdout = result.stdout.strip() if result.stdout else "" cmd_stderr = result.stderr.strip() if result.stderr else "" if result.returncode != 0: error_message_base = ( f"Step {step_num} ({description}) failed with exit code" f" {result.returncode}." ) if cmd_stdout: print_message(f"Stdout from failed command:\n{cmd_stdout}") if cmd_stderr: print_message(f"Stderr from failed command:\n{cmd_stderr}") if allow_fail: print_message( f"Warning: {error_message_base} Proceeding as allow_fail is True.", ) print_separator() return True # Command ran, allowed to fail else: print_message(f"ERROR: {error_message_base} Stopping ADB sequence.") print_separator() return False # Command ran, not allowed to fail # Command succeeded (returncode == 0) print_separator() return True def execute_adb_step_or_exit( cmd_list: List[str], step_num: int, description: str, allow_fail: bool = False, ) -> None: """Executes an ADB step and exits if it fails (unless allow_fail is True). Args: cmd_list: The list representing the ADB command and its arguments. step_num: The sequential number of this step. description: A brief description of the step. allow_fail: If True, the function does not exit on failure, just prints a warning. """ if not execute_adb_step(cmd_list, step_num, description, allow_fail): print_message( f"ERROR: ADB Step {step_num} ({description}) failed. Aborting." ) sys.exit(1) def main() -> None: """Main function to orchestrate the profile generation process.""" print_message("-" * 20) # Separator before start print_message(f"Starting script from directory: {os.getcwd()}") print_message(f"Script 1: {SCRIPT1_PATH}") print_message(f"Script 2: {SCRIPT2_PATH}") print_message( f"Wait method: Logcat pattern ('{LOGCAT_READY_PATTERN}')" if LOGCAT_READY_PATTERN else "Wait method: Only ADB connection (Logcat pattern disabled)" ) print_message("-" * 20) # Separator after info print_message("\n--- Stage 1: Configuring Device ---") print_message(f"\n--- Running Configuration Script ---") script1_abs_path = os.path.join(ANDROID_BUILD_TOP, SCRIPT1_PATH) if not os.path.exists(script1_abs_path): print_message( f"Error: Configuration script not found at {script1_abs_path}" ) sys.exit(1) if not run_shell_script(script1_abs_path, SCRIPT1_ARG): print_message("Configuration Script failed. Aborting.") sys.exit(1) print_message("Configuration Script successful.") print_message("\n--- Waiting for Device Readiness ---") if not wait_for_device_ready(): print_message("Device not ready within timeout. Aborting.") sys.exit(1) print_message("Device reported ready.") print_message("\n--- Stage 2: Starting ADB/CUJ Sequence ---") execute_adb_step_or_exit( [ "adb", "shell", "find /data/misc/profiles -name '*.prof' -exec truncate -s 0 {} \\;", ], 1, "Clear existing profiles", allow_fail=True, # Allowed to fail if dir doesn't exist etc. ) execute_adb_step_or_exit( ["adb", "shell", "am", "get-current-user"], 2, "Get current user" ) execute_adb_step_or_exit( [ "adb", "shell", "am", "start", "-S", "-W", "-a", "android.intent.action.MAIN", "-n", MAPS_PACKAGE, ], 3, f"Start Maps activity ({MAPS_PACKAGE})", ) print_message( f"Pausing {CUJ_ACTIVITY_SLEEP_SECONDS} seconds after Maps activity" " launch..." ) time.sleep(CUJ_ACTIVITY_SLEEP_SECONDS) execute_adb_step_or_exit( ["adb", "shell", "cmd", "package", "snapshot-profile", "android"], 4, "Snapshot profile after Maps", ) execute_adb_step_or_exit( ["adb", "pull", "/data/misc/profman/android.prof", MAPS_PROF_ARG], 5, f"Pull profile to {MAPS_PROF_ARG}", ) execute_adb_step_or_exit( ["adb", "shell", "am", "start", "-S", "-W", "-n", VENDING_PACKAGE], 6, f"Start Vending activity ({VENDING_PACKAGE})", ) print_message( f"Pausing {CUJ_ACTIVITY_SLEEP_SECONDS} seconds after Vending activity" " launch..." ) time.sleep(CUJ_ACTIVITY_SLEEP_SECONDS) execute_adb_step_or_exit( ["adb", "shell", "cmd", "package", "snapshot-profile", "android"], 7, "Snapshot profile after Vending", ) execute_adb_step_or_exit( ["adb", "pull", "/data/misc/profman/android.prof", VENDING_PROF_ARG], 8, f"Pull profile to {VENDING_PROF_ARG}", ) print_message("\n--- Stage 3: Running Generation Script ---") print_message( "ADB/CUJ sequence finished (check logs for warnings). Proceeding." ) script2_abs_path = os.path.join(ANDROID_BUILD_TOP, SCRIPT2_PATH) if not os.path.exists(script2_abs_path): print_message(f"Error: Generation script not found at {script2_abs_path}") sys.exit(1) if run_shell_script(script2_abs_path, *SCRIPT2_ARGS_LIST): print_message("\n--- All stages completed successfully! ---") sys.exit(0) else: print_message(f"\n--- Generation Script ({SCRIPT2_PATH}) failed. ---") sys.exit(1) if __name__ == "__main__": main()