| #!/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. |
| |
| """A simplified and cleaner wrapper for ADB commands.""" |
| from __future__ import annotations |
| from typing import Dict, Any, List, Optional |
| |
| import os |
| import subprocess |
| import time |
| |
| from utils_terminal import TerminalPrintColors |
| |
| |
| class AdbError(Exception): |
| """Custom exception for ADB errors.""" |
| |
| |
| class Adb: |
| """A class to manage adb and fastboot paths and commands.""" |
| |
| _adb_path: Optional[str] = None |
| _fastboot_path: Optional[str] = None |
| |
| def __init__(self, serial: Optional[str] = None): |
| """Initializes the Adb object.""" |
| self.serial = serial |
| if Adb._adb_path is None: |
| Adb._adb_path = Adb._find_executable("adb") |
| if Adb._fastboot_path is None: |
| Adb._fastboot_path = Adb._find_executable("fastboot") |
| |
| @staticmethod |
| def _run_command( |
| cmd: List[str], |
| *, |
| capture_output: bool = True, |
| check: bool = True, |
| text: bool = True, |
| ) -> subprocess.CompletedProcess: |
| """Runs a command using subprocess.""" |
| try: |
| return subprocess.run( |
| cmd, capture_output=capture_output, check=check, text=text |
| ) |
| except FileNotFoundError as e: |
| raise AdbError(f"Command '{cmd[0]}' not found.") from e |
| except subprocess.CalledProcessError as e: |
| stderr = e.stderr.strip() if e.stderr else "No stderr output." |
| raise AdbError( |
| f"Command '{' '.join(cmd)}' failed with exit code " |
| f"{e.returncode}.\n" |
| f"Stderr: {stderr}" |
| ) from e |
| |
| @staticmethod |
| def _find_executable(name: str) -> str: |
| """Checks if an executable is available, prompts for a path if not.""" |
| try: |
| Adb._run_command(["which", name]) |
| return name |
| except AdbError: |
| # The adb tool is also present in host tools in the Android tree. |
| username = os.environ.get("USER") |
| android_host_out = os.environ.get("ANDROID_HOST_OUT") |
| standard_paths = [] |
| if username: |
| standard_paths.append( |
| f"/usr/local/google/home/{username}/Android/Sdk/" |
| f"platform-tools/{name}" |
| ) |
| if android_host_out: |
| standard_paths.append(f"{android_host_out}/bin/{name}") |
| |
| for path in standard_paths: |
| if os.path.isfile(path) and os.access(path, os.X_OK): |
| use_standard = input( |
| f"The '{name}' command was not found in your PATH. " |
| "However, it was found at the standard location " |
| f"'{path}'.\n\nUse this path? (Y/n): " |
| ).lower() |
| if use_standard in ["y", "yes", ""]: |
| return path |
| |
| while True: |
| path = input( |
| f"Please provide the full path to your {name} binary: ") |
| if os.path.isfile(path) and os.access(path, os.X_OK): |
| return path |
| |
| print( |
| f"\nERROR: The path '{path}' is not a valid executable " |
| "file. Please try again.\n" |
| ) |
| |
| def adb_command( |
| self, |
| command: List[str], **kwargs, |
| ) -> subprocess.CompletedProcess: |
| """Runs an adb command for the specified device.""" |
| if Adb._adb_path is None: |
| Adb._adb_path = Adb._find_executable("adb") |
| cmd = [Adb._adb_path] |
| if self.serial: |
| cmd.extend(["-s", self.serial]) |
| cmd.extend(command) |
| return Adb._run_command(cmd, **kwargs) |
| |
| def fastboot_command( |
| self, |
| command: List[str], **kwargs, |
| ) -> subprocess.CompletedProcess: |
| """Runs a fastboot command.""" |
| if Adb._fastboot_path is None: |
| Adb._fastboot_path = Adb._find_executable("fastboot") |
| cmd = [Adb._fastboot_path] |
| cmd.extend(command) |
| return Adb._run_command(cmd, **kwargs) |
| |
| def root(self, *, quiet_terminal_output = False) -> None: |
| """Restart adbd with root permissions.""" |
| if not self.serial: |
| raise AdbError("Device serial not set to run root command.") |
| if not quiet_terminal_output: |
| print("Attempting to run ADB as root...") |
| result = self.adb_command(["root"], check=False) |
| if result.returncode != 0: |
| raise AdbError( |
| f"Failed to restart in root mode: {result.stderr.strip()}") |
| if "restarting" in result.stdout: |
| print("ADB is running as root.") |
| print("Waiting for device to reconnect...") |
| self.adb_command(["wait-for-device"]) |
| print("Device reconnected.") |
| elif "timeout expired" in result.stdout: |
| raise AdbError("`adb root` timed out.") |
| |
| def get_state(self) -> str: |
| """Gets the device state.""" |
| if not self.serial: |
| raise AdbError("Device serial not set to get device state.") |
| return self.adb_command(["get-state"]).stdout.strip() |
| |
| def wait_for_device_with_serial(self, timeout: int = 300): |
| """Waits for the device with the given serial to be available.""" |
| if not self.serial: |
| raise AdbError("Device serial not set.") |
| |
| print(f"Waiting for device {self.serial} to be online...") |
| start_time = time.time() |
| while time.time() - start_time < timeout: |
| try: |
| state = self.get_state() |
| if state == 'device': |
| print(f"Device {self.serial} is online.") |
| return |
| except AdbError: |
| # Ignore errors while polling |
| pass |
| time.sleep(5) |
| raise AdbError( |
| f"Timeout waiting for device {self.serial} to be online.") |
| |
| @staticmethod |
| def list_devices() -> Dict[str, Any]: |
| """Fetches available android devices and their page sizes.""" |
| if Adb._adb_path is None: |
| Adb._adb_path = Adb._find_executable("adb") |
| try: |
| result = Adb._run_command([Adb._adb_path, "devices"]) |
| except AdbError: |
| return {} |
| |
| lines = result.stdout.strip().split("\n") |
| devices = {} |
| for line in lines[1:]: |
| if "device" in line: |
| device_id = line.split('\t')[0] |
| try: |
| pagesize_result = Adb._run_command( |
| [ |
| Adb._adb_path, |
| "-s", |
| device_id, |
| "shell", |
| "getconf", |
| "PAGESIZE", |
| ] |
| ) |
| pagesize = pagesize_result.stdout.strip() |
| devices[device_id] = {"page_size": pagesize} |
| except AdbError: |
| devices[device_id] = {"page_size": "unknown"} |
| return devices |
| |
| |
| def prompt_for_device_serial() -> str: |
| """Prompts the user to select a device and returns the serial.""" |
| while True: |
| serials = Adb.list_devices() |
| devices = list(serials.keys()) |
| |
| title = "Select the serial" |
| if not devices: |
| title = "No devices found. Please connect a device and refresh." |
| |
| print( |
| f"{TerminalPrintColors.BOLD}{TerminalPrintColors.GREEN}{title}" |
| f"{TerminalPrintColors.RESET}" |
| ) |
| |
| for i, device_id in enumerate(devices): |
| pagesize = serials.get( |
| device_id, {}).get("page_size", "N/A") |
| print(f" {i + 1}: {device_id} (page size: {pagesize})") |
| |
| print("\n r: Refresh device list") |
| print(" q: Quit") |
| |
| choice = input("Enter your choice: ").strip().lower() |
| |
| if choice == "q": |
| raise SystemExit("User cancelled device selection.") |
| |
| if choice == "r": |
| print("Refreshing device list...") |
| continue |
| |
| try: |
| choice_index = int(choice) - 1 |
| if 0 <= choice_index < len(devices): |
| return devices[choice_index] |
| |
| print("Invalid choice. Please try again.\n") |
| except ValueError: |
| print("Invalid input. Please enter a number, 'r', or 'q'.\n") |
