import os, time, pprint
DEBUG=True
class ADB:
def __init__(self, deviceId):
self.__deviceId=deviceId
self.__adb_header='adb -s ' + self.__deviceId + ' '
self.__default_wait=1
def shell(self, cmd):
return self.__execute('shell ' + cmd)
def wait_device_online(self, max_delay=0):
while(max_delay>=0):
result = self.__execute('get-state')
if (('error:' in result) and ('not found' in result)):
max_delay=max_delay-1
else:
return True
time.sleep(1)
return False
def wait_device_ready(self, max_delay=0):
delaytimse=0;
while (True):
window_policy = self.dumpsys_window_policy('isStatusBarKeyguard')
if ('isStatusBarKeyguard=true' in window_policy):
return True
elif (len(window_policy) == 0):
window_policy = self.dumpsys_window_policy('mShowingLockscreen')
if ('mShowingLockscreen=true' in window_policy):
return True
time.sleep(1)
delaytimse = delaytimse + 1
if (delaytimse > max_delay):
return False
def reboot(self):
result=self.__execute('shell reboot')
time.sleep(5)
return result
def dumpsys_window_policy(self, grepCondition=None):
if(grepCondition==None):
return self.__execute('shell dumpsys window policy')
else:
grepCondition=self.__replace_space(grepCondition)
return self.__execute('shell dumpsys window policy|grep ' + grepCondition)
def dumpsys_power(self, grepCondition=None):
if (grepCondition == None):
return self.__execute('shell dumpsys power')
else:
grepCondition = self.__replace_space(grepCondition)
return self.__execute('shell dumpsys power|grep ' + grepCondition)
def rm(self, file):
time.sleep(self.__default_wait)
return self.__execute('shell rm -rf ' + file)
def ls(self, folder):
time.sleep(self.__default_wait)
return self.__execute('shell ls ' + folder)
def cat(self, file, grepCondition=None):
time.sleep(self.__default_wait)
if(grepCondition==None):
return self.__execute('shell cat ' + file)
else:
return self.__execute('shell cat ' + file + ' |grep ' + grepCondition)
def memory(self, grepCondition=None):
if(grepCondition==None):
return self.__execute('shell top -d 1 -n 1')
else:
return self.__execute('shell top -d 1 -n 1 |grep ' + grepCondition)
def ps(self, grepCondition=None):
if(grepCondition==None):
return self.__execute('shell ps')
else:
grepCondition = self.__replace_space(grepCondition)
return self.__execute('shell ps |grep ' + grepCondition)
def logcatClear(self):
time.sleep(self.__default_wait)
return self.__execute('shell logcat -c')
def logcatCollect(self, lines):
return self.__execute('shell logcat -v time -t ' + str(lines))
def productName(self):
time.sleep(self.__default_wait)
return self.__execute('shell getprop ro.product.model')
def os(self):
time.sleep(self.__default_wait)
return self.__execute('shell getprop ro.build.version.release')
def touch(self, file):
time.sleep(self.__default_wait)
return self.__execute('shell touch ' + file)
def install(self, packagename, file):
max_retry=3
packgesearch=self.__execute('shell pm list packages |grep ' + packagename)
while(packagename in packgesearch):
self.uninstall(packagename)
packgesearch=self.__execute('shell pm list packages |grep ' + packagename)
max_retry=max_retry-1
if(max_retry==0):
return False
max_retry=3
while(not(packagename in packgesearch)):
self.__execute('install -r ' + file)
packgesearch = self.__execute('shell pm list packages |grep ' + packagename)
max_retry = max_retry - 1
if (max_retry == 0):
return False
return True
def uninstall(self, packagename):
max_retry = 3
self.__execute('uninstall ' + packagename)
packgesearch = self.__execute('shell pm list packages |grep ' + packagename)
while (packagename in packgesearch):
self.uninstall(packagename)
packgesearch = self.__execute('shell pm list packages |grep ' + packagename)
max_retry = max_retry - 1
if (max_retry == 0):
return False
return True
def am(self, packageName, activity):
max_retry=5
while(True):
self.__execute('shell am start -n ' + packageName + '/' + activity)
time.sleep(2)
result=self.ps(packageName)
if(packageName in result):
return True
max_retry=max_retry-1
if(max_retry==0):
return False
def input(self, keyevent, delay=0):
if (delay > 0):
time.sleep(delay)
return self.__execute('shell input keyevent ' + str(keyevent))
def swipe(self, start_x, start_y, end_x, end_y, delay=0):
if(delay>0):
time.sleep(delay)
return self.__execute('shell input touchscreen swipe ' + str(start_x) + ' ' + str(start_y) + ' ' + str(end_x) + ' ' + str(end_y))
def screenOn(self):
#greater than android 19
screen_status = self.dumpsys_power('Display Power')
if ('Display Power' in screen_status):
if (not ('state=ON' in screen_status)):
self.input(26)
#android 19
else:
screen_status = self.dumpsys_power('mScreenOn')
if (not('mScreenOn=true' in screen_status)):
self.input(26)
def ipaddress(self):
max_retry=3
while(True):
ip=self.__execute('shell shell ifconfig wlan0|grep ask')
if('wlan0: ip ' in ip):
ip=ip.replace('wlan0: ip ','')
if(' inet addr:' in ip):
ip=ip.replace(' inet addr:','')
ip_tmp=ip.split(' ')
if(len(ip_tmp)>0):
return ip_tmp[0]
max_retry=max_retry-1
if(max_retry==0):
return ''
def __execute(self, cmd):
cmd = self.__adb_header + cmd
t_f = os.popen(cmd)
result=t_f.read()
self.__print_output(cmd, result)
return result
def __print_output(self, cmd, output):
if(DEBUG):
pprint.pprint('###DeviceID:' + self.__deviceId + ' ###')
pprint.pprint('###CMD:' + cmd + ' ###')
pprint.pprint(output)
def __replace_space(self, data):
data = str(data).strip()
data = data.replace(' ', '\ ')
return data