1# Copyright 2019 - The Android Open Source Project 2# 3# Licensed under the Apache License, Version 2.0 (the "License"); 4# you may not use this file except in compliance with the License. 5# You may obtain a copy of the License at 6# 7# http://www.apache.org/licenses/LICENSE-2.0 8# 9# Unless required by applicable law or agreed to in writing, software 10# distributed under the License is distributed on an "AS IS" BASIS, 11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 12# See the License for the specific language governing permissions and 13# limitations under the License. 14"""A tool that help to run adb to check device status.""" 15 16import re 17import subprocess 18 19from acloud import errors 20from acloud.internal import constants 21from acloud.internal.lib import utils 22 23_ADB_CONNECT = "connect" 24_ADB_DEVICE = "devices" 25_ADB_DISCONNECT = "disconnect" 26_ADB_STATUS_DEVICE = "device" 27_ADB_STATUS_DEVICE_ARGS = "-l" 28_RE_ADB_DEVICE_INFO = (r"%s\s*(?P<adb_status>[\S]+)? ?" 29 r"(usb:(?P<usb>[\S]+))? ?" 30 r"(product:(?P<product>[\S]+))? ?" 31 r"(model:(?P<model>[\S]+))? ?" 32 r"(device:(?P<device>[\S]+))? ?" 33 r"(transport_id:(?P<transport_id>[\S]+))? ?") 34_RE_SERIAL = r"(?P<ip>.+):(?P<port>.+)" 35_DEVICE_ATTRIBUTES = ["adb_status", "usb", "product", "model", "device", "transport_id"] 36_MAX_RETRIES_ON_WAIT_ADB_GONE = 5 37#KEY_CODE 82 = KEY_MENU 38_UNLOCK_SCREEN_KEYEVENT = ("%(adb_bin)s -s %(device_serial)s " 39 "shell input keyevent 82") 40_WAIT_ADB_RETRY_BACKOFF_FACTOR = 1.5 41_WAIT_ADB_SLEEP_MULTIPLIER = 2 42 43 44class AdbTools: 45 """Adb tools. 46 47 Attributes: 48 _adb_command: String, combine adb commands then execute it. 49 _adb_port: Integer, Specified adb port to establish connection. 50 _device_address: String, the device's host and port for adb to connect 51 to. For example, adb connect 127.0.0.1:5555. 52 _device_serial: String, adb device's serial number. The value can be 53 different from _device_address. For example, 54 adb -s emulator-5554 shell. 55 _device_information: Dict, will be added to adb information include usb, 56 product model, device and transport_id 57 """ 58 _adb_command = None 59 60 def __init__(self, adb_port=None, device_serial=""): 61 """Initialize. 62 63 Args: 64 adb_port: String of adb port number. 65 device_serial: String, adb device's serial number. 66 """ 67 self._adb_port = adb_port 68 self._device_address = "" 69 self._device_serial = "" 70 self._SetDeviceSerial(device_serial) 71 self._device_information = {} 72 self._CheckAdb() 73 self._GetAdbInformation() 74 75 def _SetDeviceSerial(self, device_serial): 76 """Set device serial and address. 77 78 Args: 79 device_serial: String, the device's serial number. If this 80 argument is empty, the serial number is set to the 81 network address. 82 """ 83 if device_serial: 84 match = re.match(_RE_SERIAL, device_serial) 85 if match: 86 self._adb_port = match.group("port") 87 self._device_address = device_serial 88 self._device_serial = device_serial 89 return 90 self._device_address = ("127.0.0.1:%s" % self._adb_port if 91 self._adb_port else "") 92 self._device_serial = (device_serial if device_serial else 93 self._device_address) 94 95 @classmethod 96 def _CheckAdb(cls): 97 """Find adb bin path. 98 99 Raises: 100 errors.NoExecuteCmd: Can't find the execute adb bin. 101 """ 102 if cls._adb_command: 103 return 104 cls._adb_command = utils.FindExecutable(constants.ADB_BIN) 105 if not cls._adb_command: 106 raise errors.NoExecuteCmd("Can't find the adb command.") 107 108 def GetAdbConnectionStatus(self): 109 """Get Adb connect status. 110 111 Check if self._adb_port is null (ssh tunnel is broken). 112 113 Returns: 114 String, the result of adb connection. 115 """ 116 if not self._adb_port: 117 return None 118 119 return self._device_information["adb_status"] 120 121 def _GetAdbInformation(self): 122 """Get Adb connect information. 123 124 1. Check adb devices command to get the connection information. 125 126 2. Gather information include usb, product model, device and transport_id 127 when the attached field is device. 128 129 e.g. 130 Case 1: 131 List of devices attached 132 127.0.0.1:48451 device product:aosp_cf model:Cuttlefish device:vsoc_x86 transport_id:147 133 _device_information = {"adb_status":"device", 134 "usb":None, 135 "product":"aosp_cf", 136 "model":"Cuttlefish", 137 "device":"vsoc_x86", 138 "transport_id":"147"} 139 140 Case 2: 141 List of devices attached 142 127.0.0.1:48451 offline 143 _device_information = {"adb_status":"offline", 144 "usb":None, 145 "product":None, 146 "model":None, 147 "device":None, 148 "transport_id":None} 149 150 Case 3: 151 List of devices attached 152 _device_information = {"adb_status":None, 153 "usb":None, 154 "product":None, 155 "model":None, 156 "device":None, 157 "transport_id":None} 158 """ 159 adb_cmd = [self._adb_command, _ADB_DEVICE, _ADB_STATUS_DEVICE_ARGS] 160 device_info = utils.CheckOutput(adb_cmd) 161 self._device_information = { 162 attribute: None for attribute in _DEVICE_ATTRIBUTES} 163 164 for device in device_info.splitlines(): 165 match = re.match(_RE_ADB_DEVICE_INFO % self._device_serial, device) 166 if match: 167 self._device_information = { 168 attribute: match.group(attribute) if match.group(attribute) 169 else None for attribute in _DEVICE_ATTRIBUTES} 170 171 @classmethod 172 def GetDeviceSerials(cls): 173 """Get the serial numbers of connected devices.""" 174 cls._CheckAdb() 175 adb_cmd = [cls._adb_command, _ADB_DEVICE] 176 device_info = utils.CheckOutput(adb_cmd) 177 serials = [] 178 # Skip the first line which is "List of devices attached". Each of the 179 # following lines consists of the serial number, a tab character, and 180 # the state. The last line is empty. 181 for line in device_info.splitlines()[1:]: 182 serial_state = line.split() 183 if len(serial_state) > 1: 184 serials.append(serial_state[0]) 185 return serials 186 187 def IsAdbConnectionAlive(self): 188 """Check devices connect alive. 189 190 Returns: 191 Boolean, True if adb status is device. False otherwise. 192 """ 193 return self.GetAdbConnectionStatus() == _ADB_STATUS_DEVICE 194 195 def IsAdbConnected(self): 196 """Check devices connected or not. 197 198 If adb connected and the status is device or offline, return True. 199 If there is no any connection, return False. 200 201 Returns: 202 Boolean, True if adb status not none. False otherwise. 203 """ 204 return self.GetAdbConnectionStatus() is not None 205 206 def _DisconnectAndRaiseError(self): 207 """Disconnect adb. 208 209 Disconnect from the device's network address if it shows up in adb 210 devices. For example, adb disconnect 127.0.0.1:5555. 211 212 Raises: 213 errors.WaitForAdbDieError: adb is alive after disconnect adb. 214 """ 215 try: 216 if self.IsAdbConnected(): 217 adb_disconnect_args = [self._adb_command, 218 _ADB_DISCONNECT, 219 self._device_address] 220 subprocess.check_call(adb_disconnect_args) 221 # check adb device status 222 self._GetAdbInformation() 223 if self.IsAdbConnected(): 224 raise errors.AdbDisconnectFailed( 225 "adb disconnect failed, device is still connected and " 226 "has status: [%s]" % self.GetAdbConnectionStatus()) 227 228 except subprocess.CalledProcessError: 229 utils.PrintColorString("Failed to adb disconnect %s" % 230 self._device_address, 231 utils.TextColors.FAIL) 232 233 def DisconnectAdb(self, retry=False): 234 """Retry to disconnect adb. 235 236 When retry=True, this method will retry to disconnect adb until adb 237 device is completely gone. 238 239 Args: 240 retry: Boolean, True to retry disconnect on error. 241 """ 242 retry_count = _MAX_RETRIES_ON_WAIT_ADB_GONE if retry else 0 243 # Wait for adb device is reset and gone. 244 utils.RetryExceptionType(exception_types=errors.AdbDisconnectFailed, 245 max_retries=retry_count, 246 functor=self._DisconnectAndRaiseError, 247 sleep_multiplier=_WAIT_ADB_SLEEP_MULTIPLIER, 248 retry_backoff_factor= 249 _WAIT_ADB_RETRY_BACKOFF_FACTOR) 250 251 def ConnectAdb(self): 252 """Connect adb. 253 254 Connect adb to the device's network address if the connection is not 255 alive. For example, adb connect 127.0.0.1:5555. 256 """ 257 try: 258 if not self.IsAdbConnectionAlive(): 259 adb_connect_args = [self._adb_command, 260 _ADB_CONNECT, 261 self._device_address] 262 subprocess.check_call(adb_connect_args) 263 except subprocess.CalledProcessError: 264 utils.PrintColorString("Failed to adb connect %s" % 265 self._device_address, 266 utils.TextColors.FAIL) 267 268 def AutoUnlockScreen(self): 269 """Auto unlock screen. 270 271 Auto unlock screen after invoke vnc client. 272 """ 273 try: 274 adb_unlock_args = _UNLOCK_SCREEN_KEYEVENT % { 275 "adb_bin": self._adb_command, 276 "device_serial": self._device_serial} 277 subprocess.check_call(adb_unlock_args.split()) 278 except subprocess.CalledProcessError: 279 utils.PrintColorString("Failed to unlock screen." 280 "(adb_port: %s)" % self._adb_port, 281 utils.TextColors.WARNING) 282 283 def EmuCommand(self, *args): 284 """Send an emulator command to the device. 285 286 Args: 287 args: List of strings, the emulator command. 288 289 Returns: 290 Integer, the return code of the adb command. 291 The return code is 0 if adb successfully sends the command to 292 emulator. It is irrelevant to the result of the command. 293 """ 294 adb_cmd = [self._adb_command, "-s", self._device_serial, "emu"] 295 adb_cmd.extend(args) 296 proc = subprocess.Popen(adb_cmd, stdin=subprocess.PIPE, 297 stdout=subprocess.PIPE, 298 stderr=subprocess.PIPE) 299 proc.communicate() 300 return proc.returncode 301 302 @property 303 def device_information(self): 304 """Return the device information.""" 305 return self._device_information 306