Move python-adb from system/core/adb.
This really ought to be maintained as a separate project. Also adds a
setup.py for distribution.
Bug: http://b/22881740
Change-Id: I8e11e4154fac21d3196a533613822b236b2b73e6
diff --git a/python-packages/adb/device.py b/python-packages/adb/device.py
new file mode 100644
index 0000000..516e880
--- /dev/null
+++ b/python-packages/adb/device.py
@@ -0,0 +1,339 @@
+#
+# Copyright (C) 2015 The Android Open Source Project
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+import logging
+import os
+import re
+import subprocess
+import tempfile
+
+
+class FindDeviceError(RuntimeError):
+ pass
+
+
+class DeviceNotFoundError(FindDeviceError):
+ def __init__(self, serial):
+ self.serial = serial
+ super(DeviceNotFoundError, self).__init__(
+ 'No device with serial {}'.format(serial))
+
+
+class NoUniqueDeviceError(FindDeviceError):
+ def __init__(self):
+ super(NoUniqueDeviceError, self).__init__('No unique device')
+
+
+class ShellError(RuntimeError):
+ def __init__(self, cmd, stdout, stderr, exit_code):
+ super(ShellError, self).__init__(
+ '`{0}` exited with code {1}'.format(cmd, exit_code))
+ self.cmd = cmd
+ self.stdout = stdout
+ self.stderr = stderr
+ self.exit_code = exit_code
+
+
+def get_devices():
+ with open(os.devnull, 'wb') as devnull:
+ subprocess.check_call(['adb', 'start-server'], stdout=devnull,
+ stderr=devnull)
+ out = subprocess.check_output(['adb', 'devices']).splitlines()
+
+ # The first line of `adb devices` just says "List of attached devices", so
+ # skip that.
+ devices = []
+ for line in out[1:]:
+ if not line.strip():
+ continue
+ if 'offline' in line:
+ continue
+
+ serial, _ = re.split(r'\s+', line, maxsplit=1)
+ devices.append(serial)
+ return devices
+
+
+def _get_unique_device(product=None):
+ devices = get_devices()
+ if len(devices) != 1:
+ raise NoUniqueDeviceError()
+ return AndroidDevice(devices[0], product)
+
+
+def _get_device_by_serial(serial, product=None):
+ for device in get_devices():
+ if device == serial:
+ return AndroidDevice(serial, product)
+ raise DeviceNotFoundError(serial)
+
+
+def get_device(serial=None, product=None):
+ """Get a uniquely identified AndroidDevice if one is available.
+
+ Raises:
+ DeviceNotFoundError:
+ The serial specified by `serial` or $ANDROID_SERIAL is not
+ connected.
+
+ NoUniqueDeviceError:
+ Neither `serial` nor $ANDROID_SERIAL was set, and the number of
+ devices connected to the system is not 1. Having 0 connected
+ devices will also result in this error.
+
+ Returns:
+ An AndroidDevice associated with the first non-None identifier in the
+ following order of preference:
+
+ 1) The `serial` argument.
+ 2) The environment variable $ANDROID_SERIAL.
+ 3) The single device connnected to the system.
+ """
+ if serial is not None:
+ return _get_device_by_serial(serial, product)
+
+ android_serial = os.getenv('ANDROID_SERIAL')
+ if android_serial is not None:
+ return _get_device_by_serial(android_serial, product)
+
+ return _get_unique_device(product)
+
+# Call this instead of subprocess.check_output() to work-around issue in Python
+# 2's subprocess class on Windows where it doesn't support Unicode. This
+# writes the command line to a UTF-8 batch file that is properly interpreted
+# by cmd.exe.
+def _subprocess_check_output(*popenargs, **kwargs):
+ # Only do this slow work-around if Unicode is in the cmd line.
+ if (os.name == 'nt' and
+ any(isinstance(arg, unicode) for arg in popenargs[0])):
+ # cmd.exe requires a suffix to know that it is running a batch file
+ tf = tempfile.NamedTemporaryFile('wb', suffix='.cmd', delete=False)
+ # @ in batch suppresses echo of the current line.
+ # Change the codepage to 65001, the UTF-8 codepage.
+ tf.write('@chcp 65001 > nul\r\n')
+ tf.write('@')
+ # Properly quote all the arguments and encode in UTF-8.
+ tf.write(subprocess.list2cmdline(popenargs[0]).encode('utf-8'))
+ tf.close()
+
+ try:
+ result = subprocess.check_output(['cmd.exe', '/c', tf.name],
+ **kwargs)
+ except subprocess.CalledProcessError as e:
+ # Show real command line instead of the cmd.exe command line.
+ raise subprocess.CalledProcessError(e.returncode, popenargs[0],
+ output=e.output)
+ finally:
+ os.remove(tf.name)
+ return result
+ else:
+ return subprocess.check_output(*popenargs, **kwargs)
+
+class AndroidDevice(object):
+ # Delimiter string to indicate the start of the exit code.
+ _RETURN_CODE_DELIMITER = 'x'
+
+ # Follow any shell command with this string to get the exit
+ # status of a program since this isn't propagated by adb.
+ #
+ # The delimiter is needed because `printf 1; echo $?` would print
+ # "10", and we wouldn't be able to distinguish the exit code.
+ _RETURN_CODE_PROBE_STRING = 'echo "{0}$?"'.format(_RETURN_CODE_DELIMITER)
+
+ # Maximum search distance from the output end to find the delimiter.
+ # adb on Windows returns \r\n even if adbd returns \n.
+ _RETURN_CODE_SEARCH_LENGTH = len('{0}255\r\n'.format(_RETURN_CODE_DELIMITER))
+
+ # Shell protocol feature string.
+ SHELL_PROTOCOL_FEATURE = 'shell_2'
+
+ def __init__(self, serial, product=None):
+ self.serial = serial
+ self.product = product
+ self.adb_cmd = ['adb']
+ if self.serial is not None:
+ self.adb_cmd.extend(['-s', serial])
+ if self.product is not None:
+ self.adb_cmd.extend(['-p', product])
+ self._linesep = None
+ self._features = None
+
+ @property
+ def linesep(self):
+ if self._linesep is None:
+ self._linesep = subprocess.check_output(self.adb_cmd +
+ ['shell', 'echo'])
+ return self._linesep
+
+ @property
+ def features(self):
+ if self._features is None:
+ try:
+ self._features = self._simple_call(['features']).splitlines()
+ except subprocess.CalledProcessError:
+ self._features = []
+ return self._features
+
+ def _make_shell_cmd(self, user_cmd):
+ command = self.adb_cmd + ['shell'] + user_cmd
+ if self.SHELL_PROTOCOL_FEATURE not in self.features:
+ command.append('; ' + self._RETURN_CODE_PROBE_STRING)
+ return command
+
+ def _parse_shell_output(self, out):
+ """Finds the exit code string from shell output.
+
+ Args:
+ out: Shell output string.
+
+ Returns:
+ An (exit_code, output_string) tuple. The output string is
+ cleaned of any additional stuff we appended to find the
+ exit code.
+
+ Raises:
+ RuntimeError: Could not find the exit code in |out|.
+ """
+ search_text = out
+ if len(search_text) > self._RETURN_CODE_SEARCH_LENGTH:
+ # We don't want to search over massive amounts of data when we know
+ # the part we want is right at the end.
+ search_text = search_text[-self._RETURN_CODE_SEARCH_LENGTH:]
+ partition = search_text.rpartition(self._RETURN_CODE_DELIMITER)
+ if partition[1] == '':
+ raise RuntimeError('Could not find exit status in shell output.')
+ result = int(partition[2])
+ # partition[0] won't contain the full text if search_text was truncated,
+ # pull from the original string instead.
+ out = out[:-len(partition[1]) - len(partition[2])]
+ return result, out
+
+ def _simple_call(self, cmd):
+ logging.info(' '.join(self.adb_cmd + cmd))
+ return _subprocess_check_output(
+ self.adb_cmd + cmd, stderr=subprocess.STDOUT)
+
+ def shell(self, cmd):
+ """Calls `adb shell`
+
+ Args:
+ cmd: string shell command to execute.
+
+ Returns:
+ A (stdout, stderr) tuple. Stderr may be combined into stdout
+ if the device doesn't support separate streams.
+
+ Raises:
+ ShellError: the exit code was non-zero.
+ """
+ exit_code, stdout, stderr = self.shell_nocheck(cmd)
+ if exit_code != 0:
+ raise ShellError(cmd, stdout, stderr, exit_code)
+ return stdout, stderr
+
+ def shell_nocheck(self, cmd):
+ """Calls `adb shell`
+
+ Args:
+ cmd: string shell command to execute.
+
+ Returns:
+ An (exit_code, stdout, stderr) tuple. Stderr may be combined
+ into stdout if the device doesn't support separate streams.
+ """
+ cmd = self._make_shell_cmd(cmd)
+ logging.info(' '.join(cmd))
+ p = subprocess.Popen(
+ cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+ stdout, stderr = p.communicate()
+ if self.SHELL_PROTOCOL_FEATURE in self.features:
+ exit_code = p.returncode
+ else:
+ exit_code, stdout = self._parse_shell_output(stdout)
+ return exit_code, stdout, stderr
+
+ def install(self, filename, replace=False):
+ cmd = ['install']
+ if replace:
+ cmd.append('-r')
+ cmd.append(filename)
+ return self._simple_call(cmd)
+
+ def push(self, local, remote):
+ return self._simple_call(['push', local, remote])
+
+ def pull(self, remote, local):
+ return self._simple_call(['pull', remote, local])
+
+ def sync(self, directory=None):
+ cmd = ['sync']
+ if directory is not None:
+ cmd.append(directory)
+ return self._simple_call(cmd)
+
+ def forward(self, local, remote):
+ return self._simple_call(['forward', local, remote])
+
+ def tcpip(self, port):
+ return self._simple_call(['tcpip', port])
+
+ def usb(self):
+ return self._simple_call(['usb'])
+
+ def reboot(self):
+ return self._simple_call(['reboot'])
+
+ def root(self):
+ return self._simple_call(['root'])
+
+ def unroot(self):
+ return self._simple_call(['unroot'])
+
+ def forward_remove(self, local):
+ return self._simple_call(['forward', '--remove', local])
+
+ def forward_remove_all(self):
+ return self._simple_call(['forward', '--remove-all'])
+
+ def connect(self, host):
+ return self._simple_call(['connect', host])
+
+ def disconnect(self, host):
+ return self._simple_call(['disconnect', host])
+
+ def reverse(self, remote, local):
+ return self._simple_call(['reverse', remote, local])
+
+ def reverse_remove_all(self):
+ return self._simple_call(['reverse', '--remove-all'])
+
+ def reverse_remove(self, remote):
+ return self._simple_call(['reverse', '--remove', remote])
+
+ def wait(self):
+ return self._simple_call(['wait-for-device'])
+
+ def get_prop(self, prop_name):
+ output = self.shell(['getprop', prop_name])[0].splitlines()
+ if len(output) != 1:
+ raise RuntimeError('Too many lines in getprop output:\n' +
+ '\n'.join(output))
+ value = output[0]
+ if not value.strip():
+ return None
+ return value
+
+ def set_prop(self, prop_name, value):
+ self.shell(['setprop', prop_name, value])