Upload core/adb_manager.py with huggingface_hub

core/adb_manager.py

@@ -0,0 +1,535 @@
+"""
+ADB Manager Module
+==================
+
+Handles all ADB (Android Debug Bridge) connections and operations.
+Provides methods for device detection, connection management, and command execution.
+"""
+
+import subprocess
+import logging
+import os
+import time
+import re
+from typing import Optional, List, Dict, Tuple, Any
+from dataclasses import dataclass
+from enum import Enum
+
+logger = logging.getLogger(__name__)
+
+
+class DeviceState(Enum):
+    """Device connection states"""
+    CONNECTED = "device"
+    UNAUTHORIZED = "unauthorized"
+    OFFLINE = "offline"
+    RECOVERY = "recovery"
+    SIDELOAD = "sideload"
+    BOOTLOADER = "bootloader"
+    UNKNOWN = "unknown"
+
+
+@dataclass
+class DeviceInfo:
+    """Data class for device information"""
+    serial: str
+    state: DeviceState
+    model: str = ""
+    manufacturer: str = ""
+    android_version: str = ""
+    sdk_version: str = ""
+    product: str = ""
+    is_rooted: bool = False
+    storage_total: int = 0
+    storage_free: int = 0
+
+
+class ADBManager:
+    """
+    Manages ADB connections and operations for Android devices.
+    
+    This class provides a high-level interface for:
+    - Device detection and connection
+    - File operations (push, pull, list)
+    - Shell command execution
+    - Device information retrieval
+    """
+    
+    def __init__(self, adb_path: Optional[str] = None, timeout: int = 30):
+        """
+        Initialize ADB Manager.
+        
+        Args:
+            adb_path: Path to ADB executable. If None, uses system PATH.
+            timeout: Default timeout for ADB operations in seconds.
+        """
+        self.adb_path = adb_path or self._find_adb()
+        self.timeout = timeout
+        self.current_device: Optional[str] = None
+        self._connected_devices: Dict[str, DeviceInfo] = {}
+        
+        # Verify ADB is available
+        if not self._verify_adb():
+            raise RuntimeError("ADB not found. Please install Android SDK Platform Tools.")
+    
+    def _find_adb(self) -> str:
+        """Find ADB executable in system PATH or common locations."""
+        # Check system PATH first
+        try:
+            result = subprocess.run(
+                ["which", "adb"],
+                capture_output=True,
+                text=True,
+                timeout=5
+            )
+            if result.returncode == 0:
+                return result.stdout.strip()
+        except Exception:
+            pass
+        
+        # Check common locations
+        common_paths = [
+            "/usr/bin/adb",
+            "/usr/local/bin/adb",
+            os.path.expanduser("~/Android/Sdk/platform-tools/adb"),
+            os.path.expanduser("~/.android/platform-tools/adb"),
+            "/opt/android-sdk/platform-tools/adb",
+        ]
+        
+        for path in common_paths:
+            if os.path.isfile(path) and os.access(path, os.X_OK):
+                return path
+        
+        return "adb"  # Fallback to PATH
+    
+    def _verify_adb(self) -> bool:
+        """Verify ADB is installed and accessible."""
+        try:
+            result = subprocess.run(
+                [self.adb_path, "version"],
+                capture_output=True,
+                text=True,
+                timeout=10
+            )
+            return result.returncode == 0
+        except Exception as e:
+            logger.error(f"ADB verification failed: {e}")
+            return False
+    
+    def _run_adb(self, args: List[str], device: Optional[str] = None,
+                 timeout: Optional[int] = None) -> Tuple[int, str, str]:
+        """
+        Execute an ADB command.
+        
+        Args:
+            args: List of command arguments
+            device: Target device serial (uses current_device if None)
+            timeout: Command timeout in seconds
+            
+        Returns:
+            Tuple of (return_code, stdout, stderr)
+        """
+        cmd = [self.adb_path]
+        
+        # Add device selector if specified
+        target_device = device or self.current_device
+        if target_device:
+            cmd.extend(["-s", target_device])
+        
+        cmd.extend(args)
+        
+        try:
+            result = subprocess.run(
+                cmd,
+                capture_output=True,
+                text=True,
+                timeout=timeout or self.timeout
+            )
+            return result.returncode, result.stdout, result.stderr
+        except subprocess.TimeoutExpired:
+            logger.error(f"ADB command timed out: {' '.join(cmd)}")
+            return -1, "", "Command timed out"
+        except Exception as e:
+            logger.error(f"ADB command failed: {e}")
+            return -1, "", str(e)
+    
+    def start_server(self) -> bool:
+        """Start the ADB server."""
+        code, _, _ = self._run_adb(["start-server"])
+        return code == 0
+    
+    def stop_server(self) -> bool:
+        """Stop the ADB server."""
+        code, _, _ = self._run_adb(["kill-server"])
+        return code == 0
+    
+    def restart_server(self) -> bool:
+        """Restart the ADB server."""
+        self.stop_server()
+        time.sleep(1)
+        return self.start_server()
+    
+    def list_devices(self) -> List[DeviceInfo]:
+        """
+        List all connected Android devices.
+        
+        Returns:
+            List of DeviceInfo objects for connected devices
+        """
+        code, stdout, _ = self._run_adb(["devices", "-l"])
+        
+        if code != 0:
+            return []
+        
+        devices = []
+        lines = stdout.strip().split('\n')[1:]  # Skip header
+        
+        for line in lines:
+            if not line.strip():
+                continue
+            
+            parts = line.split()
+            if len(parts) < 2:
+                continue
+            
+            serial = parts[0]
+            state_str = parts[1]
+            
+            # Parse state
+            try:
+                state = DeviceState(state_str)
+            except ValueError:
+                state = DeviceState.UNKNOWN
+            
+            # Parse additional info from the line
+            model = ""
+            product = ""
+            for part in parts[2:]:
+                if part.startswith("model:"):
+                    model = part.split(":")[1]
+                elif part.startswith("product:"):
+                    product = part.split(":")[1]
+            
+            device_info = DeviceInfo(
+                serial=serial,
+                state=state,
+                model=model,
+                product=product
+            )
+            
+            # Get detailed info if device is connected
+            if state == DeviceState.CONNECTED:
+                device_info = self._get_device_details(serial, device_info)
+            
+            devices.append(device_info)
+            self._connected_devices[serial] = device_info
+        
+        return devices
+    
+    def _get_device_details(self, serial: str, info: DeviceInfo) -> DeviceInfo:
+        """Get detailed information about a connected device."""
+        # Get manufacturer
+        code, stdout, _ = self._run_adb(
+            ["shell", "getprop", "ro.product.manufacturer"],
+            device=serial
+        )
+        if code == 0:
+            info.manufacturer = stdout.strip()
+        
+        # Get Android version
+        code, stdout, _ = self._run_adb(
+            ["shell", "getprop", "ro.build.version.release"],
+            device=serial
+        )
+        if code == 0:
+            info.android_version = stdout.strip()
+        
+        # Get SDK version
+        code, stdout, _ = self._run_adb(
+            ["shell", "getprop", "ro.build.version.sdk"],
+            device=serial
+        )
+        if code == 0:
+            info.sdk_version = stdout.strip()
+        
+        # Check if rooted
+        info.is_rooted = self._check_root(serial)
+        
+        # Get storage info
+        info.storage_total, info.storage_free = self._get_storage_info(serial)
+        
+        return info
+    
+    def _check_root(self, serial: str) -> bool:
+        """Check if device has root access."""
+        # Try su command
+        code, stdout, _ = self._run_adb(
+            ["shell", "su", "-c", "id"],
+            device=serial,
+            timeout=5
+        )
+        if code == 0 and "uid=0" in stdout:
+            return True
+        
+        # Check for common root indicators
+        code, stdout, _ = self._run_adb(
+            ["shell", "ls", "/system/xbin/su"],
+            device=serial,
+            timeout=5
+        )
+        if code == 0 and "No such file" not in stdout:
+            return True
+        
+        return False
+    
+    def _get_storage_info(self, serial: str) -> Tuple[int, int]:
+        """Get storage information (total, free) in bytes."""
+        code, stdout, _ = self._run_adb(
+            ["shell", "df", "/sdcard"],
+            device=serial
+        )
+        
+        if code != 0:
+            return 0, 0
+        
+        lines = stdout.strip().split('\n')
+        if len(lines) < 2:
+            return 0, 0
+        
+        # Parse df output
+        parts = lines[1].split()
+        if len(parts) >= 4:
+            try:
+                # Values are typically in KB
+                total = int(parts[1]) * 1024
+                free = int(parts[3]) * 1024
+                return total, free
+            except ValueError:
+                pass
+        
+        return 0, 0
+    
+    def connect_device(self, serial: str) -> bool:
+        """
+        Set the current device for operations.
+        
+        Args:
+            serial: Device serial number
+            
+        Returns:
+            True if device is connected and ready
+        """
+        devices = self.list_devices()
+        for device in devices:
+            if device.serial == serial and device.state == DeviceState.CONNECTED:
+                self.current_device = serial
+                logger.info(f"Connected to device: {serial}")
+                return True
+        
+        logger.warning(f"Device not found or not ready: {serial}")
+        return False
+    
+    def connect_wireless(self, ip: str, port: int = 5555) -> bool:
+        """
+        Connect to a device over WiFi.
+        
+        Args:
+            ip: Device IP address
+            port: ADB port (default 5555)
+            
+        Returns:
+            True if connection successful
+        """
+        address = f"{ip}:{port}"
+        code, stdout, _ = self._run_adb(["connect", address])
+        
+        if code == 0 and "connected" in stdout.lower():
+            self.current_device = address
+            logger.info(f"Connected wirelessly to: {address}")
+            return True
+        
+        return False
+    
+    def disconnect_wireless(self, ip: Optional[str] = None) -> bool:
+        """Disconnect from wireless device."""
+        if ip:
+            code, _, _ = self._run_adb(["disconnect", ip])
+        else:
+            code, _, _ = self._run_adb(["disconnect"])
+        return code == 0
+    
+    def shell(self, command: str, as_root: bool = False) -> Tuple[bool, str]:
+        """
+        Execute a shell command on the device.
+        
+        Args:
+            command: Shell command to execute
+            as_root: Execute with root privileges
+            
+        Returns:
+            Tuple of (success, output)
+        """
+        if as_root:
+            command = f"su -c '{command}'"
+        
+        code, stdout, stderr = self._run_adb(["shell", command])
+        
+        if code == 0:
+            return True, stdout
+        else:
+            return False, stderr or stdout
+    
+    def pull_file(self, remote_path: str, local_path: str,
+                  preserve_timestamp: bool = True) -> bool:
+        """
+        Pull a file from the device.
+        
+        Args:
+            remote_path: Path on device
+            local_path: Local destination path
+            preserve_timestamp: Preserve file timestamps
+            
+        Returns:
+            True if successful
+        """
+        args = ["pull"]
+        if preserve_timestamp:
+            args.append("-a")
+        args.extend([remote_path, local_path])
+        
+        code, _, _ = self._run_adb(args)
+        return code == 0
+    
+    def push_file(self, local_path: str, remote_path: str) -> bool:
+        """
+        Push a file to the device.
+        
+        Args:
+            local_path: Local file path
+            remote_path: Destination path on device
+            
+        Returns:
+            True if successful
+        """
+        code, _, _ = self._run_adb(["push", local_path, remote_path])
+        return code == 0
+    
+    def list_directory(self, path: str, as_root: bool = False) -> List[Dict[str, Any]]:
+        """
+        List contents of a directory on the device.
+        
+        Args:
+            path: Directory path on device
+            as_root: Use root privileges
+            
+        Returns:
+            List of file/directory information dictionaries
+        """
+        command = f"ls -la {path}"
+        success, output = self.shell(command, as_root=as_root)
+        
+        if not success:
+            return []
+        
+        entries = []
+        for line in output.strip().split('\n'):
+            if not line or line.startswith('total'):
+                continue
+            
+            # Parse ls -la output
+            parts = line.split()
+            if len(parts) < 8:
+                continue
+            
+            entry = {
+                'permissions': parts[0],
+                'links': parts[1],
+                'owner': parts[2],
+                'group': parts[3],
+                'size': int(parts[4]) if parts[4].isdigit() else 0,
+                'date': f"{parts[5]} {parts[6]}",
+                'name': ' '.join(parts[7:]),
+                'is_directory': parts[0].startswith('d'),
+                'is_link': parts[0].startswith('l'),
+            }
+            entries.append(entry)
+        
+        return entries
+    
+    def file_exists(self, path: str, as_root: bool = False) -> bool:
+        """Check if a file exists on the device."""
+        success, output = self.shell(f"[ -e {path} ] && echo 'exists'", as_root=as_root)
+        return success and 'exists' in output
+    
+    def get_file_size(self, path: str, as_root: bool = False) -> int:
+        """Get the size of a file on the device."""
+        success, output = self.shell(f"stat -c %s {path}", as_root=as_root)
+        if success:
+            try:
+                return int(output.strip())
+            except ValueError:
+                pass
+        return 0
+    
+    def create_backup(self, output_path: str, include_apk: bool = True,
+                      include_shared: bool = True) -> bool:
+        """
+        Create a full device backup.
+        
+        Args:
+            output_path: Path for backup file
+            include_apk: Include APK files
+            include_shared: Include shared storage
+            
+        Returns:
+            True if backup started successfully
+        """
+        args = ["backup", "-f", output_path]
+        if include_apk:
+            args.append("-apk")
+        if include_shared:
+            args.append("-shared")
+        args.append("-all")
+        
+        code, _, _ = self._run_adb(args, timeout=3600)  # 1 hour timeout
+        return code == 0
+    
+    def get_device_info(self) -> Optional[DeviceInfo]:
+        """Get information about the currently connected device."""
+        if self.current_device and self.current_device in self._connected_devices:
+            return self._connected_devices[self.current_device]
+        return None
+    
+    def reboot(self, mode: str = "normal") -> bool:
+        """
+        Reboot the device.
+        
+        Args:
+            mode: Reboot mode ('normal', 'recovery', 'bootloader')
+            
+        Returns:
+            True if reboot command sent successfully
+        """
+        if mode == "normal":
+            code, _, _ = self._run_adb(["reboot"])
+        elif mode == "recovery":
+            code, _, _ = self._run_adb(["reboot", "recovery"])
+        elif mode == "bootloader":
+            code, _, _ = self._run_adb(["reboot", "bootloader"])
+        else:
+            return False
+        
+        return code == 0
+    
+    def wait_for_device(self, timeout: int = 60) -> bool:
+        """
+        Wait for a device to be connected.
+        
+        Args:
+            timeout: Maximum wait time in seconds
+            
+        Returns:
+            True if device connected within timeout
+        """
+        code, _, _ = self._run_adb(["wait-for-device"], timeout=timeout)
+        return code == 0