|
| 1 | +"""ADB wrapper — works across Android 5+ without extra Python packages.""" |
| 2 | + |
| 3 | +from __future__ import annotations |
| 4 | + |
| 5 | +import logging |
| 6 | +import shutil |
| 7 | +import subprocess |
| 8 | +import time |
| 9 | +from dataclasses import dataclass |
| 10 | +from typing import Callable, Iterable |
| 11 | + |
| 12 | +log = logging.getLogger("fiver") |
| 13 | + |
| 14 | + |
| 15 | +class ADBError(RuntimeError): |
| 16 | + pass |
| 17 | + |
| 18 | + |
| 19 | +@dataclass |
| 20 | +class Device: |
| 21 | + serial: str |
| 22 | + state: str |
| 23 | + |
| 24 | + @property |
| 25 | + def online(self) -> bool: |
| 26 | + return self.state == "device" |
| 27 | + |
| 28 | + @property |
| 29 | + def wireless(self) -> bool: |
| 30 | + return ":" in self.serial |
| 31 | + |
| 32 | + |
| 33 | +class ADB: |
| 34 | + def __init__(self, binary: str = "adb") -> None: |
| 35 | + self.binary = binary |
| 36 | + |
| 37 | + def available(self) -> bool: |
| 38 | + return shutil.which(self.binary) is not None or self.binary not in {"adb", ""} |
| 39 | + |
| 40 | + def path(self) -> str | None: |
| 41 | + if self.binary not in {"adb", ""} and PathExists(self.binary): |
| 42 | + return self.binary |
| 43 | + return shutil.which(self.binary) |
| 44 | + |
| 45 | + def run(self, *args: str, timeout: float = 30.0, check: bool = True) -> str: |
| 46 | + cmd = [self.binary, *args] |
| 47 | + try: |
| 48 | + proc = subprocess.run( |
| 49 | + cmd, |
| 50 | + capture_output=True, |
| 51 | + text=True, |
| 52 | + timeout=timeout, |
| 53 | + check=False, |
| 54 | + ) |
| 55 | + except FileNotFoundError as exc: |
| 56 | + raise ADBError( |
| 57 | + "adb not found. Install Android platform-tools, then re-run: fiver --doctor" |
| 58 | + ) from exc |
| 59 | + except subprocess.TimeoutExpired as exc: |
| 60 | + raise ADBError(f"adb timed out: {' '.join(args)}") from exc |
| 61 | + |
| 62 | + out = (proc.stdout or "").strip() |
| 63 | + err = (proc.stderr or "").strip() |
| 64 | + if check and proc.returncode != 0: |
| 65 | + msg = err or out or f"exit {proc.returncode}" |
| 66 | + raise ADBError(f"adb {' '.join(args)}: {msg}") |
| 67 | + return out if out else err |
| 68 | + |
| 69 | + def start_server(self) -> None: |
| 70 | + try: |
| 71 | + self.run("start-server", timeout=20.0) |
| 72 | + except ADBError as exc: |
| 73 | + log.warning("%s", exc) |
| 74 | + |
| 75 | + def version(self) -> str: |
| 76 | + try: |
| 77 | + out = self.run("version", check=False) |
| 78 | + return out.splitlines()[0] if out else "unknown" |
| 79 | + except ADBError: |
| 80 | + return "unavailable" |
| 81 | + |
| 82 | + def devices(self) -> list[Device]: |
| 83 | + out = self.run("devices", check=False) |
| 84 | + result: list[Device] = [] |
| 85 | + for line in out.splitlines()[1:]: |
| 86 | + line = line.strip() |
| 87 | + if not line: |
| 88 | + continue |
| 89 | + parts = line.split() |
| 90 | + if len(parts) >= 2: |
| 91 | + result.append(Device(serial=parts[0], state=parts[1])) |
| 92 | + return result |
| 93 | + |
| 94 | + def online(self) -> list[Device]: |
| 95 | + return [d for d in self.devices() if d.online] |
| 96 | + |
| 97 | + def shell(self, serial: str, script: str, timeout: float = 15.0) -> str: |
| 98 | + args: list[str] = [] |
| 99 | + if serial: |
| 100 | + args.extend(["-s", serial]) |
| 101 | + args.extend(["shell", script]) |
| 102 | + return self.run(*args, timeout=timeout, check=False) |
| 103 | + |
| 104 | + def getprop(self, serial: str, prop: str) -> str: |
| 105 | + return self.shell(serial, f"getprop {prop}").replace("\r", "").strip() |
| 106 | + |
| 107 | + def info(self, serial: str) -> str: |
| 108 | + model = self.getprop(serial, "ro.product.model") or "unknown" |
| 109 | + rel = self.getprop(serial, "ro.build.version.release") or "?" |
| 110 | + api = self.getprop(serial, "ro.build.version.sdk") or "?" |
| 111 | + return f"{model} (Android {rel}, API {api})" |
| 112 | + |
| 113 | + def phone_ip(self, serial: str) -> str | None: |
| 114 | + scripts = [ |
| 115 | + r"ip -f inet addr show wlan0 2>/dev/null | awk '/inet /{print $2}' | cut -d/ -f1 | head -n1", |
| 116 | + r"ip -f inet addr show wlan1 2>/dev/null | awk '/inet /{print $2}' | cut -d/ -f1 | head -n1", |
| 117 | + r"ip -f inet addr show rndis0 2>/dev/null | awk '/inet /{print $2}' | cut -d/ -f1 | head -n1", |
| 118 | + r"ip route show default 2>/dev/null | awk '{for(i=1;i<=NF;i++) if($i==\"src\"){print $(i+1); exit}}'", |
| 119 | + r"ip route 2>/dev/null | awk '/src /{for(i=1;i<=NF;i++) if($i==\"src\"){print $(i+1); exit}}'", |
| 120 | + r"ip -f inet addr show 2>/dev/null | awk '/inet / && $2 !~ /^(127\.|169\.254\.)/ {print $2}' | cut -d/ -f1 | head -n1", |
| 121 | + r"ifconfig wlan0 2>/dev/null | awk '/inet addr:/{print $2}' | cut -d: -f2 | head -n1", |
| 122 | + r"ifconfig wlan0 2>/dev/null | awk '/inet /{print $2}' | head -n1", |
| 123 | + ] |
| 124 | + for script in scripts: |
| 125 | + out = self.shell(serial, script).replace("\r", "").strip() |
| 126 | + out = out.removeprefix("addr:") |
| 127 | + if out.count(".") == 3 and not out.startswith("127."): |
| 128 | + return out |
| 129 | + return None |
| 130 | + |
| 131 | + def tcpip(self, serial: str, port: int) -> None: |
| 132 | + args: list[str] = [] |
| 133 | + if serial and ":" not in serial: |
| 134 | + args.extend(["-s", serial]) |
| 135 | + args.extend(["tcpip", str(port)]) |
| 136 | + self.run(*args, timeout=20.0) |
| 137 | + |
| 138 | + def connect(self, hostport: str, timeout: float = 7.0) -> bool: |
| 139 | + try: |
| 140 | + out = self.run("connect", hostport, timeout=timeout, check=False).lower() |
| 141 | + except ADBError: |
| 142 | + return False |
| 143 | + if any(x in out for x in ("unable", "failed", "cannot", "error", "timed out")): |
| 144 | + log.debug("adb connect %s -> %s", hostport, out) |
| 145 | + return False |
| 146 | + # confirm listed |
| 147 | + return any(d.serial == hostport and d.online for d in self.devices()) or "connected" in out |
| 148 | + |
| 149 | + def enable_wireless(self, serial: str, port: int, fixed_ip: str = "") -> str: |
| 150 | + ip = fixed_ip or self.phone_ip(serial) |
| 151 | + if not ip: |
| 152 | + raise ADBError("could not detect phone IP address. Please ensure phone Wi-Fi is turned ON and connected to the same local network.") |
| 153 | + log.info("setting phone ADB port to %s...", port) |
| 154 | + self.tcpip(serial, port) |
| 155 | + time.sleep(1.5) |
| 156 | + target = f"{ip}:{port}" |
| 157 | + log.info("connecting wirelessly to %s...", target) |
| 158 | + if not self.connect(target, timeout=6.0): |
| 159 | + time.sleep(1.0) |
| 160 | + if not self.connect(target, timeout=8.0): |
| 161 | + raise ADBError(f"wireless connect failed for {target}. Ensure your PC and phone are on the same Wi-Fi network.") |
| 162 | + return target |
| 163 | + |
| 164 | + def send_phone_notification(self, serial: str, title: str, message: str) -> bool: |
| 165 | + """Send a notification / alert message to the Android phone screen via adb.""" |
| 166 | + try: |
| 167 | + # Try posting native notification via cmd notification (Android 11+) |
| 168 | + cmd = f"cmd notification post -S inbox -t '{title}' 'fiver_connection' '{message}'" |
| 169 | + res = self.shell(serial, cmd) |
| 170 | + if "error" not in res.lower(): |
| 171 | + return True |
| 172 | + except Exception: |
| 173 | + pass |
| 174 | + return False |
| 175 | + |
| 176 | + def wait_online( |
| 177 | + self, |
| 178 | + timeout: float, |
| 179 | + poll: float = 1.5, |
| 180 | + on_pending: Callable[[str], None] | None = None, |
| 181 | + stop_flag: Callable[[], bool] | None = None, |
| 182 | + ) -> Device: |
| 183 | + deadline = time.monotonic() + timeout |
| 184 | + last_msg = "" |
| 185 | + while time.monotonic() < deadline: |
| 186 | + if stop_flag and stop_flag(): |
| 187 | + raise ADBError("stopped while waiting for device") |
| 188 | + online = self.online() |
| 189 | + if online: |
| 190 | + return online[0] |
| 191 | + for d in self.devices(): |
| 192 | + if d.state in {"unauthorized", "offline"}: |
| 193 | + msg = ( |
| 194 | + f"{d.serial} is {d.state} — unlock the phone and allow USB debugging" |
| 195 | + ) |
| 196 | + if on_pending and msg != last_msg: |
| 197 | + on_pending(msg) |
| 198 | + last_msg = msg |
| 199 | + time.sleep(max(0.4, poll)) |
| 200 | + raise ADBError( |
| 201 | + "no authorized phone found.\n" |
| 202 | + " 1) Enable Developer options + USB debugging (one-time on the phone)\n" |
| 203 | + " 2) Unlock phone, plug USB, tap Allow\n" |
| 204 | + " See: fiver --help-android" |
| 205 | + ) |
| 206 | + |
| 207 | + def pick( |
| 208 | + self, |
| 209 | + prefer_serial: str = "", |
| 210 | + phone_ip: str = "", |
| 211 | + port: int = 5555, |
| 212 | + prefer_wireless: bool = True, |
| 213 | + ) -> Device | None: |
| 214 | + if prefer_serial: |
| 215 | + return Device(serial=prefer_serial, state="device") |
| 216 | + if phone_ip: |
| 217 | + return Device(serial=f"{phone_ip}:{port}", state="device") |
| 218 | + online = self.online() |
| 219 | + if not online: |
| 220 | + return None |
| 221 | + usb = [d for d in online if not d.wireless] |
| 222 | + wifi = [d for d in online if d.wireless] |
| 223 | + if prefer_wireless and wifi: |
| 224 | + return wifi[0] |
| 225 | + if usb: |
| 226 | + return usb[0] |
| 227 | + return online[0] |
| 228 | + |
| 229 | + |
| 230 | +def PathExists(path: str) -> bool: |
| 231 | + from pathlib import Path |
| 232 | + |
| 233 | + return Path(path).exists() |
| 234 | + |
| 235 | + |
| 236 | +def install_hints() -> Iterable[str]: |
| 237 | + import platform |
| 238 | + |
| 239 | + system = platform.system().lower() |
| 240 | + if system == "linux": |
| 241 | + return [ |
| 242 | + "Arch / CachyOS / Manjaro: sudo pacman -S scrcpy android-tools", |
| 243 | + "Debian / Ubuntu / Mint: sudo apt update && sudo apt install -y scrcpy adb", |
| 244 | + "Fedora / RHEL: sudo dnf install -y scrcpy android-tools", |
| 245 | + "openSUSE: sudo zypper install scrcpy android-tools", |
| 246 | + "Alpine: sudo apk add scrcpy android-tools", |
| 247 | + ] |
| 248 | + if system == "darwin": |
| 249 | + return [ |
| 250 | + "macOS (Homebrew): brew install scrcpy android-platform-tools", |
| 251 | + ] |
| 252 | + if system == "windows": |
| 253 | + return [ |
| 254 | + "Windows (winget): winget install Genymobile.scrcpy Google.PlatformTools", |
| 255 | + "Windows (choco): choco install scrcpy adb", |
| 256 | + ] |
| 257 | + return ["Install scrcpy and Android platform-tools (adb) for your OS."] |
0 commit comments