Artemis-Eapil/artemis/core/diagnostics/probes/adb_probe.py

# Copyright 2026 Google LLC
#
# 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.

"""Android ADB & Device Connectivity Probe with AVD Detection."""

import asyncio
import os
from pathlib import Path
import shutil
import subprocess
from typing import Any

from artemis.core.diagnostics.probes.base import BaseProbe
from artemis.core.diagnostics.schema import (
    DeviceInfo,
    ProbeAction,
    ProbeCategory,
    ProbeResult,
    ProbeStatus,
)
from artemis.platform import OSType, platform
from artemis.toolchain import toolchain
from artemis.utils.logger import get_logger

logger = get_logger(__name__)


class AdbDeviceProbe(BaseProbe):
    """Deep inspection probe for Android Debug Bridge, connected mobile devices, and local AVD emulators."""

    def __init__(self, target_serial: str | None = None):
        self._target_serial = target_serial

    @property
    def probe_id(self) -> str:
        return "android_adb"

    @property
    def category(self) -> ProbeCategory:
        return ProbeCategory.DEVICE

    @property
    def is_blocker(self) -> bool:
        return True

    def set_target_serial(self, serial: str | None) -> None:
        """Set or update preferred active device serial."""
        self._target_serial = serial

    def _locate_adb(self) -> str | None:
        """Find the adb binary path from ToolchainResolver."""
        return toolchain.resolve("adb", force_refresh=True)

    def _locate_emulator(self) -> str | None:
        """Find the emulator binary path from PATH or standard SDK environments."""
        emu_path = shutil.which("emulator")
        if emu_path:
            return emu_path

        sdk_candidates = [
            os.environ.get("ANDROID_HOME"),
            os.environ.get("ANDROID_SDK_ROOT"),
            str(Path.home() / "Android" / "Sdk"),
            str(Path.home() / "Android" / "sdk"),
            str(Path.home() / "Library" / "Android" / "sdk"),
            os.getenv("LOCALAPPDATA", "") + "/Android/Sdk" if os.getenv("LOCALAPPDATA") else None,
            "/usr/lib/android-sdk",
            "/opt/android-sdk",
        ]
        for base in sdk_candidates:
            if base:
                cand = (
                    Path(base)
                    / "emulator"
                    / ("emulator.exe" if platform.os_type == OSType.WINDOWS else "emulator")
                )
                if cand.is_file():
                    return str(cand)
        return None

    def _list_installed_avds(self, emulator_path: str | None = None) -> list[str]:
        """Discover all locally created Android Virtual Devices (AVDs)."""
        avd_names: list[str] = []

        # Method 1: Scan ~/.android/avd/*.ini
        avd_dir = Path.home() / ".android" / "avd"
        if avd_dir.is_dir():
            for ini_file in avd_dir.glob("*.ini"):
                name = ini_file.stem
                if name and name not in avd_names:
                    avd_names.append(name)

        return avd_names

    async def _get_adb_version(self, adb_path: str) -> str:
        """Query ADB version string."""
        try:
            proc = await asyncio.create_subprocess_exec(
                adb_path,
                "version",
                stdout=subprocess.PIPE,
                stderr=subprocess.PIPE,
            )
            stdout, _ = await proc.communicate()
            if stdout:
                first_line = stdout.decode(errors="replace").strip().splitlines()[0]
                return first_line
        except Exception as e:
            logger.debug(f"Failed to read adb version: {e}")
        return "Installed"

    async def _parse_adb_devices(self, adb_path: str) -> list[DeviceInfo]:
        """Execute `adb devices -l` and extract structured device metadata."""
        devices: list[DeviceInfo] = []
        try:
            proc = await asyncio.create_subprocess_exec(
                adb_path,
                "devices",
                "-l",
                stdout=subprocess.PIPE,
                stderr=subprocess.PIPE,
            )
            stdout, _ = await proc.communicate()
            if proc.returncode != 0 or not stdout:
                return devices

            lines = [
                line.strip()
                for line in stdout.decode(errors="replace").splitlines()
                if line.strip()
            ]
            for line in lines[1:]:  # Skip "List of devices attached"
                parts = line.split()
                if len(parts) < 2:
                    continue

                serial = parts[0]
                state = parts[1]
                extra_tokens = parts[2:]

                model = None
                product = None
                for token in extra_tokens:
                    if token.startswith("model:"):
                        model = token.split(":", 1)[1].replace("_", " ")
                    elif token.startswith("product:"):
                        product = token.split(":", 1)[1]

                is_emulator = (
                    serial.startswith("emulator-") or "127.0.0.1" in serial or "localhost" in serial
                )
                if is_emulator and not model:
                    model = "Android Emulator"

                device_info = DeviceInfo(
                    serial=serial,
                    state=state,
                    model=model,
                    product=product,
                    is_emulator=is_emulator,
                )
                devices.append(device_info)

        except Exception as e:
            logger.error(f"Error querying adb devices: {e}")

        # For authorized devices, attempt non-blocking quick property enrichment
        for dev in devices:
            if dev.state == "device":
                try:
                    p_prop = await asyncio.create_subprocess_exec(
                        adb_path,
                        "-s",
                        dev.serial,
                        "shell",
                        "getprop ro.build.version.release",
                        stdout=subprocess.PIPE,
                        stderr=subprocess.PIPE,
                    )
                    p_size = await asyncio.create_subprocess_exec(
                        adb_path,
                        "-s",
                        dev.serial,
                        "shell",
                        "wm size",
                        stdout=subprocess.PIPE,
                        stderr=subprocess.PIPE,
                    )
                    out_prop, _ = await p_prop.communicate()
                    out_size, _ = await p_size.communicate()

                    if out_prop:
                        dev.android_version = out_prop.decode(errors="replace").strip()
                    if out_size:
                        size_str = out_size.decode(errors="replace").strip()
                        if "Physical size:" in size_str:
                            dev.screen_resolution = size_str.split("Physical size:")[-1].strip()

                    # Query installed application packages with timeout protection
                    try:
                        p_pkg = await asyncio.create_subprocess_exec(
                            adb_path,
                            "-s",
                            dev.serial,
                            "shell",
                            "pm list packages",
                            stdout=subprocess.PIPE,
                            stderr=subprocess.PIPE,
                        )
                        out_pkg, _ = await asyncio.wait_for(p_pkg.communicate(), timeout=2.5)
                        if out_pkg:
                            pkgs = [
                                line.split("package:", 1)[1].strip()
                                for line in out_pkg.decode(errors="replace").splitlines()
                                if line.startswith("package:")
                                and line.split("package:", 1)[1].strip()
                            ]
                            dev.installed_packages = pkgs
                    except Exception as e_pkg:
                        logger.debug(f"Failed to query packages for {dev.serial}: {e_pkg}")
                except Exception:
                    pass

        return devices

    async def probe(self) -> ProbeResult:
        """Run complete ADB, device, and local AVD connectivity probe."""
        adb_path = self._locate_adb()
        emulator_path = self._locate_emulator()
        installed_avds = self._list_installed_avds(emulator_path)

        # 1. ADB binary check
        if not adb_path:
            actions = []
            if platform.os_type == OSType.WINDOWS:
                actions.append(
                    ProbeAction(
                        action_type="command",
                        label="Install via WinGet",
                        payload="winget install Google.PlatformTools",
                    )
                )
                actions.append(
                    ProbeAction(
                        action_type="command",
                        label="Run PowerShell Setup",
                        payload="powershell -ExecutionPolicy Bypass -File scripts/install_deps.ps1",
                    )
                )
            elif platform.os_type == OSType.MACOS:
                actions.append(
                    ProbeAction(
                        action_type="command",
                        label="Install via Homebrew",
                        payload="brew install android-platform-tools",
                    )
                )
            else:
                actions.append(
                    ProbeAction(
                        action_type="command",
                        label="Linux Quick Install",
                        payload="sudo apt-get install -y adb",
                    )
                )
                actions.append(
                    ProbeAction(
                        action_type="command",
                        label="One-Click Install All",
                        payload="bash scripts/install_deps.sh",
                    )
                )

            actions.append(
                ProbeAction(
                    action_type="link",
                    label="Download Android Platform Tools",
                    payload="https://developer.android.com/tools/releases/platform-tools",
                )
            )

            return ProbeResult(
                id=self.probe_id,
                category=self.category,
                title="Device / Emulator Connected",
                status=ProbeStatus.FAIL,
                is_blocker=self.is_blocker,
                summary="ADB Not Found",
                description="Android Debug Bridge (adb) is not installed or not in PATH.",
                metadata={"installed": False},
                actions=actions,
            )

        adb_ver = await self._get_adb_version(adb_path)
        devices = await self._parse_adb_devices(adb_path)

        # Normalize emulator command display
        is_emu_in_path = shutil.which("emulator") is not None
        emu_display_cmd = "emulator" if is_emu_in_path else (emulator_path or "emulator")

        metadata: dict[str, Any] = {
            "installed": True,
            "adb_path": adb_path,
            "adb_version": adb_ver,
            "emulator_path": emulator_path,
            "is_emulator_in_path": is_emu_in_path,
            "installed_avds": installed_avds,
            "device_count": len(devices),
            "devices": [d.model_dump() for d in devices],
        }

        # 2. Case: No devices found
        if not devices:
            actions: list[ProbeAction] = []

            # If user has installed AVDs, generate exact launch commands for each
            if installed_avds:
                for avd in installed_avds:
                    actions.append(
                        ProbeAction(
                            action_type="command",
                            label=f"Launch {avd}",
                            payload=f"{emu_display_cmd} -avd {avd}",
                        )
                    )
            else:
                actions.append(
                    ProbeAction(
                        action_type="command",
                        label="Start Default Emulator",
                        payload=f"{emu_display_cmd} -avd Pixel_8_API_34",
                    )
                )

            actions.append(
                ProbeAction(
                    action_type="hint",
                    label="Connect via USB",
                    payload="Connect your Android phone via USB cable and enable Developer Options -> USB Debugging.",
                )
            )

            desc = "ADB is ready, but no active Android device or emulator was detected."
            if installed_avds:
                desc += f" Found {len(installed_avds)} installed emulator(s): {', '.join(installed_avds)}."

            return ProbeResult(
                id=self.probe_id,
                category=self.category,
                title="Device / Emulator Connected",
                status=ProbeStatus.WARN,
                is_blocker=self.is_blocker,
                summary="No Device Found",
                description=desc,
                metadata=metadata,
                actions=actions,
            )

        # 3. Check for authorized and ready devices
        ready_devices = [d for d in devices if d.state == "device"]
        unauthorized_devices = [d for d in devices if d.state == "unauthorized"]

        if not ready_devices:
            if unauthorized_devices:
                unauth_serial = unauthorized_devices[0].serial
                return ProbeResult(
                    id=self.probe_id,
                    category=self.category,
                    title="Device / Emulator Connected",
                    status=ProbeStatus.WARN,
                    is_blocker=self.is_blocker,
                    summary="Device Unauthorized",
                    description=f"Device detected ({unauth_serial}), but USB debugging is not yet authorized.",
                    metadata=metadata,
                    actions=[
                        ProbeAction(
                            action_type="hint",
                            label="Authorize USB Debugging",
                            payload="Please unlock your Android phone and tap 'Allow' on the USB Debugging permission prompt.",
                        ),
                        ProbeAction(
                            action_type="command",
                            label="Restart ADB Server",
                            payload=f"{adb_path} kill-server && {adb_path} start-server",
                        ),
                    ],
                )
            else:
                first_serial = devices[0].serial
                first_state = devices[0].state
                return ProbeResult(
                    id=self.probe_id,
                    category=self.category,
                    title="Device / Emulator Connected",
                    status=ProbeStatus.WARN,
                    is_blocker=self.is_blocker,
                    summary="Device Booting",
                    description=f"Device detected ({first_serial}, state: {first_state}). It is currently booting up or offline. Please wait a few seconds...",
                    metadata=metadata,
                    actions=[
                        ProbeAction(
                            action_type="hint",
                            label="Device Booting",
                            payload=f"Device {first_serial} is initializing. It will be ready automatically in a few seconds.",
                        )
                    ],
                )

        # 4. Ready device available
        active_dev = ready_devices[0]
        if self._target_serial:
            matched = next((d for d in ready_devices if d.serial == self._target_serial), None)
            if matched:
                active_dev = matched

        display_name = active_dev.model or active_dev.serial
        if active_dev.screen_resolution:
            display_name = f"{display_name} ({active_dev.screen_resolution})"

        metadata["active_device"] = active_dev.model_dump()

        return ProbeResult(
            id=self.probe_id,
            category=self.category,
            title="Device / Emulator Connected",
            status=ProbeStatus.PASS,
            is_blocker=self.is_blocker,
            summary="Connected",
            description=f"ADB connected to {display_name}. Ready for perception and touch automation.",
            metadata=metadata,
            actions=[
                ProbeAction(
                    action_type="hint",
                    label="Device Ready",
                    payload=f"Serial: {active_dev.serial} | Android OS: {active_dev.android_version or 'Unknown'}",
                )
            ],
        )