Package com :: Package dtmilano :: Package android :: Package adb :: Module adbclient |
[hide private]
[frames] | no frames]

# Source Code for Module com.dtmilano.android.adb.adbclient

      1  '''
      2  Copyright (C) 2012-2013  Diego Torres Milano
      3  Created on Dec 1, 2012
      4
      5  Licensed under the Apache License, Version 2.0 (the "License");
      6  you may not use this file except in compliance with the License.
      7  You may obtain a copy of the License at
      8
      9         http://www.apache.org/licenses/LICENSE-2.0
     10
     11  Unless required by applicable law or agreed to in writing, software
     12  distributed under the License is distributed on an "AS IS" BASIS,
     13  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
     14  See the License for the specific language governing permissions and
     15  limitations under the License.
     16
     17  @author: Diego Torres Milano
     18  '''
     19
     20  __version__ = '5.7.0'
     21
     22  import sys
     23  import warnings
     24  if sys.executable:
     25      if 'monkeyrunner' in sys.executable:
     26          warnings.warn(
     27      '''
     28
     29      You should use a 'python' interpreter, not 'monkeyrunner' for this module
     30
     31      ''', RuntimeWarning)
     32  import socket
     33  import time
     34  import re
     35  import signal
     36  import os
     37  import types
     38  import platform
     39
     40  DEBUG = False
     41
     42  HOSTNAME = 'localhost'
     43  try:
     44      PORT = int(os.environ['ANDROID_ADB_SERVER_PORT'])
     45  except KeyError:
     46      PORT = 5037
     47
     48  OKAY = 'OKAY'
     49  FAIL = 'FAIL'
     50
     51  UP = 0
     52  DOWN = 1
     53  DOWN_AND_UP = 2
     54
     55  TIMEOUT = 15
     56
     57  # some device properties
     58  VERSION_SDK_PROPERTY = 'ro.build.version.sdk'
     59  VERSION_RELEASE_PROPERTY = 'ro.build.version.release'
     60
     61
     62 -class Device:
     63      @staticmethod
     64 -    def factory(_str):
     65          if DEBUG:
     66              print >> sys.stderr, "Device.factory(", _str, ")"
     67          values = _str.split(None, 2)
     68          if DEBUG:
     69              print >> sys.stderr, "values=", values
     70          return Device(*values)
     71
     72 -    def __init__(self, serialno, status, qualifiers=None):
     73          self.serialno = serialno
     74          self.status = status
     75          self.qualifiers = qualifiers
     76
     77 -    def __str__(self):
     78          return "<<<" + self.serialno + ", " + self.status + ", %s>>>" % self.qualifiers
     79
     80
     81 -class AdbClient:
     82
     83 -    def __init__(self, serialno=None, hostname=HOSTNAME, port=PORT, settransport=True, reconnect=True):
     84          self.serialno = serialno
     85          self.hostname = hostname
     86          self.port = port
     87
     88          self.reconnect = reconnect
     89          self.__connect()
     90
     91          self.checkVersion()
     92
     93          self.build = {}
     94          ''' Build properties '''
     95
     96          self.isTransportSet = False
     97          if settransport and serialno != None:
     98              self.__setTransport()
     99              self.build[VERSION_SDK_PROPERTY] = self.__getProp(VERSION_SDK_PROPERTY)
    100
    101
    102      @staticmethod
    103 -    def setAlarm(timeout):
    104          osName = platform.system()
    105          if osName.startswith('Windows'):  # alarm is not implemented in Windows
    106              return
    107          if DEBUG:
    108              print >> sys.stderr, "setAlarm(%d)" % timeout
    109          signal.alarm(timeout)
    110
    111 -    def setSerialno(self, serialno):
    112          if self.isTransportSet:
    113              raise ValueError("Transport is already set, serialno cannot be set once this is done.")
    114          self.serialno = serialno
    115          self.__setTransport()
    116          self.build[VERSION_SDK_PROPERTY] = self.__getProp(VERSION_SDK_PROPERTY)
    117
    118 -    def setReconnect(self, val):
    119          self.reconnect = val
    120
    121 -    def __connect(self):
    122          if DEBUG:
    123              print >> sys.stderr, "__connect()"
    124          self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    125          self.socket.settimeout(TIMEOUT)
    126          try:
    127              self.socket.connect((self.hostname, self.port))
    128          except socket.error, ex:
    129              raise RuntimeError("ERROR: Connecting to %s:%d: %s.\nIs adb running on your computer?" % (self.socket, self.port, ex))
    130
    131 -    def close(self):
    132          if DEBUG:
    133              print >> sys.stderr, "Closing socket...", self.socket
    134          if self.socket:
    135              self.socket.close()
    136
    137 -    def __del__(self):
    138          try:
    139              self.close()
    140          except:
    141              pass
    142
    143 -    def __send(self, msg, checkok=True, reconnect=False):
    144          if DEBUG:
    145              print >> sys.stderr, "__send(%s, checkok=%s, reconnect=%s)" % (msg, checkok, reconnect)
    146          if not re.search('^host:', msg):
    147              if not self.isTransportSet:
    148                  self.__setTransport()
    149          else:
    150              self.checkConnected()
    151          b = bytearray(msg, 'utf-8')
    152          self.socket.send('%04X%s' % (len(b), b))
    153          if checkok:
    154              self.__checkOk()
    155          if reconnect:
    156              if DEBUG:
    157                  print >> sys.stderr, "    __send: reconnecting"
    158              self.__connect()
    159              self.__setTransport()
    160
    161 -    def __receive(self, nob=None):
    162          if DEBUG:
    163              print >> sys.stderr, "__receive()"
    164          self.checkConnected()
    165          if nob is None:
    166              nob = int(self.socket.recv(4), 16)
    167          if DEBUG:
    168              print >> sys.stderr, "    __receive: receiving", nob, "bytes"
    169          recv = bytearray()
    170          nr = 0
    171          while nr < nob:
    172              chunk = self.socket.recv(min((nob - nr), 4096))
    173              recv.extend(chunk)
    174              nr += len(chunk)
    175          if DEBUG:
    176              print >> sys.stderr, "    __receive: returning len=", len(recv)
    177          return str(recv)
    178
    179 -    def __checkOk(self):
    180          if DEBUG:
    181              print >> sys.stderr, "__checkOk()"
    182          self.checkConnected()
    183          self.setAlarm(TIMEOUT)
    184          recv = self.socket.recv(4)
    185          if DEBUG:
    186              print >> sys.stderr, "    __checkOk: recv=", repr(recv)
    187          try:
    188              if recv != OKAY:
    189                  error = self.socket.recv(1024)
    190                  raise RuntimeError("ERROR: %s %s" % (repr(recv), error))
    191          finally:
    192              self.setAlarm(0)
    193          if DEBUG:
    194              print >> sys.stderr, "    __checkOk: returning True"
    195          return True
    196
    197 -    def checkConnected(self):
    198          if DEBUG:
    199              print >> sys.stderr, "checkConnected()"
    200          if not self.socket:
    201              raise RuntimeError("ERROR: Not connected")
    202          if DEBUG:
    203              print >> sys.stderr, "    checkConnected: returning True"
    204          return True
    205
    206 -    def checkVersion(self, reconnect=True):
    207          if DEBUG:
    208              print >> sys.stderr, "checkVersion(reconnect=%s)" % reconnect
    209          self.__send('host:version', reconnect=False)
    210          version = self.socket.recv(8)
    211          VERSION = '0004001f'
    212          if version != VERSION:
    213              raise RuntimeError("ERROR: Incorrect ADB server version %s (expecting %s)" % (version, VERSION))
    214          if reconnect:
    215              self.__connect()
    216
    217 -    def __setTransport(self):
    218          if DEBUG:
    219              print >> sys.stderr, "__setTransport()"
    220          if not self.serialno:
    221              raise ValueError("serialno not set, empty or None")
    222          self.checkConnected()
    223          serialnoRE = re.compile(self.serialno)
    224          found = False
    225          for device in self.getDevices():
    226              if serialnoRE.match(device.serialno):
    227                  found = True
    228                  break
    229          if not found:
    230              raise RuntimeError("ERROR: couldn't find device that matches '%s'" % self.serialno)
    231          self.serialno = device.serialno
    232          msg = 'host:transport:%s' % self.serialno
    233          if DEBUG:
    234              print >> sys.stderr, "    __setTransport: msg=", msg
    235          self.__send(msg, reconnect=False)
    236          self.isTransportSet = True
    237
    238 -    def getDevices(self):
    239          if DEBUG:
    240              print >> sys.stderr, "getDevices()"
    241          self.__send('host:devices-l', checkok=False)
    242          try:
    243              self.__checkOk()
    244          except RuntimeError, ex:
    245              print >> sys.stderr, "**ERROR:", ex
    246              return None
    247          devices = []
    248          for line in self.__receive().splitlines():
    249              devices.append(Device.factory(line))
    250          self.__connect()
    251          return devices
    252
    253 -    def shell(self, cmd=None):
    254          if DEBUG:
    255              print >> sys.stderr, "shell(cmd=%s)" % cmd
    256          if cmd:
    257              self.__send('shell:%s' % cmd, checkok=True, reconnect=False)
    258              out = ''
    259              while True:
    260                  _str = None
    261                  try:
    262                      _str = self.socket.recv(4096)
    263                  except Exception, ex:
    264                      print >> sys.stderr, "ERROR:", ex
    265                  if not _str:
    266                      break
    267                  out += _str
    268              if self.reconnect:
    269                  if DEBUG:
    270                      print >> sys.stderr, "Reconnecting..."
    271                  self.close()
    272                  self.__connect()
    273                  self.__setTransport()
    274              return out
    275          else:
    276              self.__send('shell:')
    277              # sin = self.socket.makefile("rw")
    278              # sout = self.socket.makefile("r")
    279              # return (sin, sin)
    280              sout = adbClient.socket.makefile("r")
    281              return sout
    282
    283 -    def getRestrictedScreen(self):
    284          ''' Gets C{mRestrictedScreen} values from dumpsys. This is a method to obtain display dimensions '''
    285
    286          rsRE = re.compile('\s*mRestrictedScreen=\((?P<x>\d+),(?P<y>\d+)\) (?P<w>\d+)x(?P<h>\d+)')
    287          for line in self.shell('dumpsys window').splitlines():
    288              m = rsRE.match(line)
    289              if m:
    290                  return m.groups()
    291          raise RuntimeError("Couldn't find mRestrictedScreen in dumpsys")
    292
    293 -    def __getProp(self, key, strip=True):
    294          prop = self.shell('getprop %s' % key)
    295          if strip:
    296              prop = prop.rstrip('\r\n')
    297          return prop
    298
    299 -    def __getDisplayWidth(self, key, strip=True):
    300          (x, y, w, h) = self.getRestrictedScreen()
    301          return int(w)
    302
    303 -    def __getDisplayHeight(self, key, strip=True):
    304          (x, y, w, h) = self.getRestrictedScreen()
    305          return int(h)
    306
    307 -    def getSystemProperty(self, key, strip=True):
    308          return self.getProperty(key, strip)
    309
    310 -    def getProperty(self, key, strip=True):
    311          ''' Gets the property value for key '''
    312
    313          import collections
    314          MAP_KEYS = collections.OrderedDict([
    315                            (re.compile('display.width'), self.__getDisplayWidth),
    316                            (re.compile('display.height'), self.__getDisplayHeight),
    317                            (re.compile('.*'), self.__getProp),
    318                            ])
    319          '''Maps properties key values (as regexps) to instance methods to obtain its values.'''
    320
    321          for kre in MAP_KEYS.keys():
    322              if kre.match(key):
    323                  return MAP_KEYS[kre](key=key, strip=strip)
    324          raise ValueError("key='%s' does not match any map entry")
    325
    326 -    def getSdkVersion(self):
    327          '''
    328          Gets the SDK version.
    329          '''
    330
    331          return self.build[VERSION_SDK_PROPERTY]
    332
    333 -    def press(self, name, eventType=DOWN_AND_UP):
    334          cmd = 'input keyevent %s' % name
    335          if DEBUG:
    336              print >> sys.stderr, "press(%s)" % cmd
    337          self.shell(cmd)
    338
    339 -    def longPress(self, name):
    340          # WORKAROUND:
    341          # Using 'input keyevent --longpress POWER' does not work correctly in
    342          # KitKat (API 19), it sends a short instead of a long press.
    343          # This uses the events instead, but it may vary from device to device.
    344          # The events sent are device dependent and may not work on other devices.
    345          # If this does not work on your device please do:
    346          #     $ adb shell getevent -l
    347          # and post the output to https://github.com/dtmilano/AndroidViewClient/issues
    348          # specifying the device and API level.
    349          if name == 'POWER' or name == 'KEY_POWER':
    350              self.shell('sendevent /dev/input/event0 1 116 1')
    351              self.shell('sendevent /dev/input/event0 0 0 0')
    352              time.sleep(0.5)
    353              self.shell('sendevent /dev/input/event0 1 116 0')
    354              self.shell('sendevent /dev/input/event0 0 0 0')
    355              return
    356
    357          version = self.getSdkVersion()
    358          if version >= 19:
    359              cmd = 'input keyevent --longpress %s' % name
    360              if DEBUG:
    361                  print >> sys.stderr, "longPress(%s)" % cmd
    362              self.shell(cmd)
    363          else:
    364              raise RuntimeError("longpress: not supported for API < 19 (version=%d)" % version)
    365
    366 -    def startActivity(self, component=None, flags=None, uri=None):
    367          cmd = 'am start'
    368          if component:
    369              cmd += ' -n %s' % component
    370          if flags:
    371              cmd += ' -f %s' % flags
    372          if uri:
    373              cmd += ' %s' % uri
    374          if DEBUG:
    375              print >> sys.stderr, "Starting activity: %s" % cmd
    376          out = self.shell(cmd)
    377          if re.search(r"(Error type)|(Error: )|(Cannot find 'App')", out, re.IGNORECASE | re.MULTILINE):
    378              raise RuntimeError(out)
    379
    380 -    def takeSnapshot(self, reconnect=False):
    381          '''
    382          Takes a snapshot of the device and return it as a PIL Image.
    383          '''
    384
    385          try:
    386              from PIL import Image
    387          except:
    388              raise Exception("You have to install PIL to use takeSnapshot()")
    389          self.__send('framebuffer:', checkok=True, reconnect=False)
    390          import struct
    391          # case 1: // version
    392          #           return 12; // bpp, size, width, height, 4*(length, offset)
    393          received = self.__receive(1 * 4 + 12 * 4)
    394          (version, bpp, size, width, height, roffset, rlen, boffset, blen, goffset, glen, aoffset, alen) = struct.unpack('<' + 'L' * 13, received)
    395          if DEBUG:
    396              print >> sys.stderr, "    takeSnapshot:", (version, bpp, size, width, height, roffset, rlen, boffset, blen, goffset, glen, aoffset, alen)
    397          offsets = {roffset:'R', goffset:'G', boffset:'B'}
    398          if bpp == 32:
    399              if alen != 0:
    400                  offsets[aoffset] = 'A'
    401              else:
    402                  warnings.warn('''framebuffer is specified as 32bpp but alpha length is 0''')
    403          argMode = ''.join([offsets[o] for o in sorted(offsets)])
    404          if DEBUG:
    405              print >> sys.stderr, "    takeSnapshot:", (version, bpp, size, width, height, roffset, rlen, boffset, blen, goffset, blen, aoffset, alen, argMode)
    406          if argMode == 'BGRA':
    407              argMode = 'RGBA'
    408          if bpp == 16:
    409              mode = 'RGB'
    410              argMode += ';16'
    411          else:
    412              mode = argMode
    413          self.__send('\0', checkok=False, reconnect=False)
    414          if DEBUG:
    415              print >> sys.stderr, "    takeSnapshot: reading %d bytes" % (size)
    416          received = self.__receive(size)
    417          if reconnect:
    418              self.__connect()
    419              self.__setTransport()
    420          if DEBUG:
    421              print >> sys.stderr, "    takeSnapshot: Image.frombuffer(%s, %s, %s, %s, %s, %s, %s)" % (mode, (width, height), 'data', 'raw', argMode, 0, 1)
    422          return Image.frombuffer(mode, (width, height), received, 'raw', argMode, 0, 1)
    423
    424 -    def touch(self, x, y, eventType=DOWN_AND_UP):
    425          self.shell('input tap %d %d' % (x, y))
    426
    427 -    def drag(self, (x0, y0), (x1, y1), duration, steps=1):
    428          '''
    429          Sends drag event (actually it's using C{input swipe} command.
    430
    431          @param (x0, y0): starting point
    432          @param (x1, y1): ending point
    433          @param duration: duration of the event in ms
    434          @param steps: number of steps (currently ignored by @{input swipe}
    435          '''
    436
    437          version = self.getSdkVersion()
    438          if version <= 15:
    439              raise RuntimeError('drag: API <= 15 not supported (version=%d)' % version)
    440          elif version <= 17:
    441              self.shell('input swipe %d %d %d %d' % (x0, y0, x1, y1))
    442          else:
    443              self.shell('input swipe %d %d %d %d %d' % (x0, y0, x1, y1, duration))
    444
    445 -    def type(self, text):
    446          self.shell(u'input text "%s"' % text)
    447
    448 -    def wake(self):
    449          if not self.isScreenOn():
    450              self.shell('input keyevent POWER')
    451
    452 -    def isLocked(self):
    453          '''
    454          Checks if the device screen is locked.
    455
    456          @return True if the device screen is locked
    457          '''
    458
    459          lockScreenRE = re.compile('mShowingLockscreen=(true|false)')
    460          m = lockScreenRE.search(self.shell('dumpsys window policy'))
    461          if m:
    462              return (m.group(1) == 'true')
    463          raise RuntimeError("Couldn't determine screen lock state")
    464
    465 -    def isScreenOn(self):
    466          '''
    467          Checks if the screen is ON.
    468
    469          @return True if the device screen is ON
    470          '''
    471
    472          screenOnRE = re.compile('mScreenOnFully=(true|false)')
    473          m = screenOnRE.search(self.shell('dumpsys window policy'))
    474          if m:
    475              return (m.group(1) == 'true')
    476          raise RuntimeError("Couldn't determine screen ON state")
    477
    478 -    def unlock(self):
    479          '''
    480          Unlocks the screen of the device.
    481          '''
    482
    483          self.shell('input keyevent MENU')
    484          self.shell('input keyevent BACK')
    485
    486      @staticmethod
    487 -    def percentSame(image1, image2):
    488          '''
    489          Returns the percent of pixels that are equal
    490
    491          @author: catshoes
    492          '''
    493
    494          # If the images differ in size, return 0% same.
    495          size_x1, size_y1 = image1.size
    496          size_x2, size_y2 = image2.size
    497          if (size_x1 != size_x2 or
    498              size_y1 != size_y2):
    499              return 0
    500
    501          # Images are the same size
    502          # Return the percent of pixels that are equal.
    503          numPixelsSame = 0
    504          numPixelsTotal = size_x1 * size_y1
    505          image1Pixels = image1.load()
    506          image2Pixels = image2.load()
    507
    508          # Loop over all pixels, comparing pixel in image1 to image2
    509          for x in range(size_x1):
    510              for y in range(size_y1):
    511                  if (image1Pixels[x, y] == image2Pixels[x, y]):
    512                      numPixelsSame += 1
    513
    514          return numPixelsSame / float(numPixelsTotal)
    515
    516      @staticmethod
    517 -    def sameAs(image1, image2, percent=1.0):
    518          '''
    519          Compares 2 images
    520
    521          @author: catshoes
    522          '''
    523
    524          return (AdbClient.percentSame(image1, image2) >= percent)
    525
    526
    527  if __name__ == '__main__':
    528      adbClient = AdbClient(os.environ['ANDROID_SERIAL'])
    529      INTERACTIVE = False
    530      if INTERACTIVE:
    531          sout = adbClient.shell()
    532          prompt = re.compile(".+@android:(.*) [$#] \r\r\n")
    533          while True:
    534              try:
    535                  cmd = raw_input('adb $ ')
    536              except EOFError:
    537                  break
    538              if cmd == 'exit':
    539                  break
    540              adbClient.socket.__send(cmd + "\r\n")
    541              sout.readline(4096)  # eat first line, which is the command
    542              while True:
    543                  line = sout.readline(4096)
    544                  if prompt.match(line):
    545                      break
    546                  print line,
    547                  if not line:
    548                      break
    549
    550          print "\nBye"
    551      else:
    552          print 'date:', adbClient.shell('date')
    553