# Copyright 2016 Google Inc. # # Use of this source code is governed by a BSD-style license that can be # found in the LICENSE file. from __future__ import print_function import re import time import subprocess import sys class Adb: def __init__(self, device_serial=None, adb_binary=None, echo=False): self.__invocation = [adb_binary] if device_serial: self.__invocation.extend(['-s', device_serial]) self.__echo = echo self.__is_root = None self.__has_established_connection = False def shell(self, cmd): if self.__echo: self.__echo_shell_cmd(cmd) self.__invoke('shell', cmd) def check(self, cmd): if self.__echo: self.__echo_shell_cmd(cmd) self.__establish_connection() result = subprocess.check_output(self.__invocation + ['shell', cmd]) if self.__echo: print(result, file=sys.stderr) return result def root(self): if not self.is_root(): self.__invoke('root') self.__has_established_connection = False self.__is_root = None return self.is_root() def is_root(self): if self.__is_root is None: self.__is_root = ('root' == self.check('whoami').strip()) return self.__is_root def remount(self): self.__invoke('remount') def reboot(self): self.__is_root = None self.shell('reboot') self.__has_established_connection = False def __echo_shell_cmd(self, cmd): escaped = [re.sub(r'([^a-zA-Z0-9])', r'\\\1', x) for x in cmd.strip().splitlines()] self.__invoke('shell', 'echo', '$(whoami)@$(getprop ro.serialno)$', " '\n>' ".join(escaped)) def __invoke(self, *args): self.__establish_connection() subprocess.call(self.__invocation + list(args), stdout=sys.stderr) def __establish_connection(self): if self.__has_established_connection: return self.__has_established_connection = True self.__invoke('wait-for-device') while True: time.sleep(1) if '1' == self.check('getprop sys.boot_completed').strip(): break