Source code for adb_command

import subprocess
from functools import singledispatchmethod
import adb_tool_py as adb_tool
import adb_tool_py.command as command


[docs] class AdbCommand: """ A class to execute ADB commands on a specified Android device. """ device: adb_tool.AdbDevice = None def __init__(self, adb: str = "adb", device: adb_tool.AdbDevice = None): """ Initializes the AdbCommand class. :param adb: Path to the ADB executable, defaults to "adb". :param device: An instance of AdbDevice to specify the target device, defaults to None. """ self.adb = adb self.set_device(device)
[docs] def set_device(self, device: adb_tool.AdbDevice) -> 'AdbCommand': """ Sets the target device for ADB commands. :param device: An instance of AdbDevice. :return: The current instance of AdbCommand. """ self.device = device return self
def _base_cmd(self) -> list: """ Constructs the base ADB command with the target device. :return: A list representing the base ADB command. """ if self.device is None or self.device.serial is None: return [self.adb] else: return [self.adb, '-s', self.device.serial]
[docs] @singledispatchmethod def query(self, cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE) -> subprocess.CompletedProcess: """ Executes an ADB command and waits for it to complete. :param cmd: The command to execute (str or list). :param stdout: Standard output pipe, defaults to subprocess.PIPE. :param stderr: Standard error pipe, defaults to subprocess.PIPE. :return: The completed process. """ raise NotImplementedError("Unsupported type")
@query.register def _(self, cmd: str, stdout=subprocess.PIPE, stderr=subprocess.PIPE) -> subprocess.CompletedProcess: """ Executes an ADB command given as a string. :param cmd: The command to execute (str). :param stdout: Standard output pipe, defaults to subprocess.PIPE. :param stderr: Standard error pipe, defaults to subprocess.PIPE. :return: The completed process. """ return self.query(cmd.split(' '), stdout, stderr) @query.register def _(self, cmd: list, stdout=subprocess.PIPE, stderr=subprocess.PIPE) -> subprocess.CompletedProcess: """ Executes an ADB command given as a list of arguments. :param cmd: The command to execute (list). :param stdout: Standard output pipe, defaults to subprocess.PIPE. :param stderr: Standard error pipe, defaults to subprocess.PIPE. :return: The completed process. """ ret = command.command(*self._base_cmd(), *cmd, stdout=stdout, stderr=stderr) if ret.returncode != 0: raise RuntimeError(f"Error: {ret.stderr.decode()}") return ret
[docs] @singledispatchmethod def query_async(self, cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE) -> subprocess.Popen: """ Executes an ADB command asynchronously. :param cmd: The command to execute (str or list). :param stdout: Standard output pipe, defaults to subprocess.PIPE. :param stderr: Standard error pipe, defaults to subprocess.PIPE. :return: The process object. """ raise NotImplementedError("Unsupported type")
@query_async.register def _(self, cmd: str, stdout=subprocess.PIPE, stderr=subprocess.PIPE) -> subprocess.Popen: """ Executes an ADB command asynchronously given as a string. :param cmd: The command to execute (str). :param stdout: Standard output pipe, defaults to subprocess.PIPE. :param stderr: Standard error pipe, defaults to subprocess.PIPE. :return: The process object. """ return self.query_async(cmd.split(' '), stdout, stderr) @query_async.register def _(self, cmd: list, stdout=subprocess.PIPE, stderr=subprocess.PIPE) -> subprocess.Popen: """ Executes an ADB command asynchronously given as a list of arguments. :param cmd: The command to execute (list). :param stdout: Standard output pipe, defaults to subprocess.PIPE. :param stderr: Standard error pipe, defaults to subprocess.PIPE. :return: The process object. """ return command.command_async(*self._base_cmd(), *cmd, stdout=stdout, stderr=stderr)