• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
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
16from distutils.spawn import find_executable
17
18import re
19import subprocess
20
21from acloud import errors
22from acloud.internal import constants
23from acloud.internal.lib import utils
24
25_ADB_CONNECT = "connect"
26_ADB_DEVICE = "devices"
27_ADB_DISCONNECT = "disconnect"
28_ADB_STATUS_DEVICE = "device"
29
30
31class AdbTools(object):
32    """Adb tools.
33
34    Args:
35        adb_port: String of adb port number.
36    """
37    def __init__(self, adb_port=None):
38        self._adb_command = ""
39        self._adb_port = adb_port
40        self._device_serial = ""
41        self._SetDeviceSerial()
42        self._CheckAdb()
43
44    def _SetDeviceSerial(self):
45        """Set device serial."""
46        if self._adb_port:
47            self._device_serial = "127.0.0.1:%s" % self._adb_port
48
49    def _CheckAdb(self):
50        """Find adb bin path.
51
52        Raises:
53            errors.NoExecuteCmd: Can't find the execute adb bin.
54        """
55        self._adb_command = find_executable(constants.ADB_BIN)
56        if not self._adb_command:
57            raise errors.NoExecuteCmd("Can't find the adb command.")
58
59    def GetAdbConnectionStatus(self):
60        """Get Adb connect status.
61
62        We determine adb connection in below manner:
63        1. Check if self._adb_port is null (ssh tunnel is broken).
64        2. Check adb devices command to get the connection status of the
65        adb devices. When the attached field is device, then device is returned,
66        if it is offline, then offline is returned. If no device is found,
67        the None is returned.
68
69        e.g.
70            Case 1: return device
71            List of devices attached
72            127.0.0.1:48451 device
73
74            Case 2: return offline
75            List of devices attached
76            127.0.0.1:48451 offline
77
78            Case 3: return None
79            List of devices attached
80
81        Returns:
82            String, the result of adb connection.
83        """
84        if not self._adb_port:
85            return None
86
87        adb_cmd = [self._adb_command, _ADB_DEVICE]
88        device_info = subprocess.check_output(adb_cmd)
89        for device in device_info.splitlines():
90            match = re.match(r"%s\s(?P<adb_status>.+)" % self._device_serial, device)
91            if match:
92                return match.group("adb_status")
93        return None
94
95    def IsAdbConnectionAlive(self):
96        """Check devices connect alive.
97
98        Returns:
99            Boolean, True if adb status is device. False otherwise.
100        """
101        return self.GetAdbConnectionStatus() == _ADB_STATUS_DEVICE
102
103    def IsAdbConnected(self):
104        """Check devices connected or not.
105
106        If adb connected and the status is device or offline, return True.
107        If there is no any connection, return False.
108
109        Returns:
110            Boolean, True if adb status not none. False otherwise.
111        """
112        return self.GetAdbConnectionStatus() is not None
113
114    def DisconnectAdb(self):
115        """Disconnect adb.
116
117        Only disconnect if the devices shows up in adb devices.
118        """
119        try:
120            if self.IsAdbConnected():
121                adb_disconnect_args = [self._adb_command,
122                                       _ADB_DISCONNECT,
123                                       self._device_serial]
124                subprocess.check_call(adb_disconnect_args)
125        except subprocess.CalledProcessError:
126            utils.PrintColorString("Failed to adb disconnect %s" %
127                                   self._device_serial,
128                                   utils.TextColors.FAIL)
129
130    def ConnectAdb(self):
131        """Connect adb.
132
133        Only connect if adb connection is not alive.
134        """
135        try:
136            if not self.IsAdbConnectionAlive():
137                adb_connect_args = [self._adb_command,
138                                    _ADB_CONNECT,
139                                    self._device_serial]
140                subprocess.check_call(adb_connect_args)
141        except subprocess.CalledProcessError:
142            utils.PrintColorString("Failed to adb connect %s" %
143                                   self._device_serial,
144                                   utils.TextColors.FAIL)
145