Python实现Android性能数据获取及压测场景模拟

在做自动化脚本测试中，需要去获取Android设备CPU和内存数据，并对不同CPU占用场景进行模拟，用来测试系统不同CPU状态下对应用运行状态的影响。本文内容包括Python与adb shell命令的交互，Android性能数据的获取，及压测场景模拟工具的调用。

一、Python与adb shell的交互

1. Python常规adb shell命令交互调用

使用subprocess，执行adb shell命令，获取输出结果和异常信息，一次性执行，一版用来获取一些系统信息。

# coding=utf-8
import subprocess
def adbShell(cmds):
    """
    :param cmds: adb shell commands string
    :return:
        a subprocess
    """
    try:
        proc = subprocess.Popen(
            cmds,
            stdin=subprocess.PIPE,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            shell=True
        )
        stdout, stderr = proc.communicate()
        return stdout, stderr
    except Exception, e:
        raise e
if __name__ == '__main__':
    stdout, stderr = adbShell("adb shell getprop ro.build.version.sdk")

2. Python多个adb命令调用交互方式一

比如进入到“/data/local/tmp/“目录做一些操作，先进入到adb shell的执行环境，在commnuicate中传入要执行的命令。

def multiShell():
    try:
        cmds = [
            "cd /data/local/tmp/",
            # do something
            "exit" # 执行完操作退出
        ]
        proc = subprocess.Popen(
            "adb shell",
            stdin=subprocess.PIPE,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            shell=True
        )
        stdout, stderr = proc.communicate(("\n".join(cmds) + "\n").encode('utf-8'))
        return stdout, stderr
    except Exception as e:
        print e.message

3. Python多个adb命令调用交互方式二

def multiShell():
    try:
        proc = subprocess.Popen(
            "adb shell",
            stdin=subprocess.PIPE,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            shell=True
        )
        proc.stdin.write("cd /data/local/tmp/\n")
        # do something
        # proc.stdin.write("rm test.so\n")
        proc.stdin.write("exit\n") # 执行完操作退出
        stdout, stderr = proc.communicate()
        return stdout, stderr
    except Exception as e:
        print e.message

二、Python获取Android性能数据

这边给下Python获取Android系统和独立进程CPU数据的实现，package参数替换成自己要测试的包名。关于Android CPU数据的获取和计算原理见Android 性能监控之CPU监控，里面已经做了详细的说明，此处不再赘述。

# coding=utf-8
import subprocess
import time

def shell(cmds):
    """
    :param cmds: adb shell commands string
    :return:
        a subprocess
    """
    try:
        proc = subprocess.Popen(
            cmds,
            stdin=subprocess.PIPE,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            shell=True
        )
        return proc
    except Exception, e:
        raise e

def isSDKHigh(serial=None):
    """
    判断SDK是否为8.0以上的版本
    :return:
    """
    try:
        if not serial:
            proc = shell("adb shell getprop ro.build.version.sdk")
        else:
            proc = shell("adb -s {} shell getprop ro.build.version.sdk".format(serial))
        stdout, stderr = proc.communicate()
        print "isSDKHigh stdout={}".format(stdout.strip())
        if int(stdout.strip()) >= 26:
            return True
        return False
    except Exception, e:
        print e.message

def getPackagePid(package, isSDKHigh, serial=None):
    """
    获取对应包名的主进程PID
    :param package:
    :return: PID
    """
    try:
        if not serial:
            if isSDKHigh:
                proc = shell("adb shell ps -A | grep {}".format(package))
            else:
                proc = shell("adb shell ps | grep {}".format(package))
        else:
            if isSDKHigh:
                proc = shell("adb -s {} shell ps -A | grep {}".format(serial, package))
            else:
                proc = shell("adb -s {} shell ps | grep {}".format(serial, package))
        stdout, stderr = proc.communicate()
        if len(stderr.strip()) > 0:
            return None
        processArray = stdout.split("\n")
        """
        一个应用可能包含多个进程，只过滤主要进程PID
        """
        for item in processArray:
            # 过滤掉数组中的空格对象
            processDetailList = filter(lambda x: x, item.split(" "))
            if processDetailList[len(processDetailList) - 1] == package:
                return processDetailList[1]
    except Exception, e:
        print e.message
        return None

def getSystemCpu(serial=None):
    """
    获取系统CPU占用
    :param serial: 设备序列号
    :return:
    """
    try:
        if not serial:
            proc = shell("adb shell cat /proc/stat")
        else:
            proc = shell("adb -s {} shell cat /proc/stat".format(serial))
        stdout, stderr = proc.communicate()
        # print "getSystemCpu err:{}".format(stderr)
        if len(stderr.strip()) > 0:
            return None
        cpuCoreArray = stdout.split("\n")
        # 第一行为系统CPU汇总数据
        systemCpu = cpuCoreArray[0].split(" ")
        systemCpuList = filter(lambda x: x, systemCpu)
        # print systemCpuList
        return systemCpuList
    except Exception as err:
        print err.message
        return None

def getProcessCpu(pid, serial=None):
    """
    获取进程CPU信息
    :param pid: 进程PID
    :param serial: 设备序列号
    :return:
    """
    try:
        if not serial:
            proc = shell("adb shell cat /proc/{}/stat".format(pid))
        else:
            proc = shell("adb -s {} shell cat /proc/{}/stat".format(serial, pid))
        stdout, stderr = proc.communicate()
        # print "getProcessCpu err:{}".format(stderr)
        if len(stderr.strip()) > 0:
            return None
        processCpu = stdout.split(" ")
        processCpuList = filter(lambda x: x, processCpu)
        # print processCpuList
        return processCpuList
    except Exception as e:
        print e.message
        return None

def calculateProcRate(oldList, newList, totalCPUTime):
    """
    计算独立进程CPU占用率，原理见：http://101.200.49.38/android-performance-monitor-cpu/
    :param oldList: 缓存的前一个CPU采样数据
    :param newList: 当前采样的CPU数据
    :return:
    """
    procTime1 = 0
    procTime2 = 0
    for index in range(13, 16):
        procTime1 += int(oldList[index])
        procTime2 += int(newList[index])
    # print "calculateProcRate procTime1:{} procTime2:{} totalCPUTime:{}".format(procTime1, procTime2, totalCpuTime)
    return (procTime2 - procTime1) * 100 / totalCPUTime

def calculateSysRate(oldList, newList):
    """
    计算系统CPU占用率，原理见：http://101.200.49.38/android-performance-monitor-cpu/
    :param oldList: 缓存的前一个CPU采样数据
    :param newList: 当前采样的CPU数据
    :return:
    """
    cpuTime1 = 0
    cpuTime2 = 0
    for index in range(1, len(oldList)):
        cpuTime1 += int(oldList[index])
        cpuTime2 += int(newList[index])

    totalCpuTime = cpuTime2 - cpuTime1
    idleCpuTime = int(newList[4]) - int(oldList[4])
    sysRate = (totalCpuTime - idleCpuTime) * 100 / totalCpuTime
    # print "calculateSysRate cpuTime1:{} cpuTime2:{} totalCpuTime:{} idleCpuTime:{}".format(cpuTime1, cpuTime2, totalCpuTime, idleCpuTime)
    return sysRate, totalCpuTime

if __name__ == "__main__":
    try:
        package = "com.test.performance"
        cacheSystemCpu = None
        cacheProcessCpu = None
        sdkHigh = isSDKHigh(serial=None)
        pid = getPackagePid(package, sdkHigh, serial=None)
        cacheProcessCpu = getProcessCpu(pid, serial=None)
        cacheSystemCpu = getSystemCpu()
        while True:
            time.sleep(1)
            # 获取系统CPU数据
            newSystemCpu = getSystemCpu()
            if newSystemCpu:
                sysRate, totalCpuTime = calculateSysRate(cacheSystemCpu, newSystemCpu)
                print ">>>>>>sysRate:{} totalCpuTime:{}".format(sysRate, totalCpuTime)
                cacheSystemCpu = newSystemCpu
            # 获取进程CPU数据
            newProcessCpu = getProcessCpu(pid, serial=None)
            if not newProcessCpu:
                pid = getPackagePid(package, sdkHigh, serial=None)
                newProcessCpu = getProcessCpu(pid, serial=None)
            if not newProcessCpu:
                continue
            if totalCpuTime > 0:
                processRate = calculateProcRate(cacheProcessCpu, newProcessCpu, totalCpuTime)
                cacheProcessCpu = newProcessCpu
                print ">>>>>>processRate:{}".format(processRate)
    except Exception, e:
        print e.message

关于Android系统内存信息的获取通过以下命令，计算方式和原理说明见Android 性能监控之内存监控，此处不再赘述，具体Python实现可以参考以上CPU获取的实现做下改造。

adb shell cat /proc/meminfo

独立进程内存信息的获取通过以下命令，<package>替换成要监控的包名参数：

adb shell dumpsys meminfo <package>

三、Python实现Android压测场景的模拟（stress-ng）

stress-ng是stress的加强版，功能更加完善。它可以充分测试Linux、BSD等类Unix服务器以获得高负载并监控压力下CPU，内存，I/O和磁盘读写能力的状况，也可包括：Socket处理能力，进线程创建和终止、上下文切换等。采用C语言开发并在GPLv2协议下授权。CentOS 7 的EPEL源包含2个压力测试工具，一个是标准的stress，另一个是其升级版stress-ng。stress-ng是stress的加强版，完全兼容stress，并在此基础上增加了几百个参数。

首先是从Github上下载源码，编译生成可执行文件，项目地址：

https://github.com/ColinIanKing/stress-ng

不想编译的话，可以临时拿这边提供的版本测试，这个可执行文件编译时间在2021年5月份：

https://github.com/wenguan0927/stress-ng/blob/master/stress-ng

1. 模拟原理说明

1) 安卓低版本（Android 7.0以下 SDK < 26）

安卓低版本使用 top -d 1 -n 1 ，查看System和User的系统占用，不足场景要求的部分使用模拟工具增加负荷。如System+User=5%，则模拟增加35%的负荷，以达到40%的cpu占用。执行命令后输出信息形式如下：

User 3%, System 9%, IOW 0%, IRQ 0%
User 5 + Nice 0 + Sys 14 + Idle 121 + IOW 1 + IRQ 0 + SIRQ 0 = 141
  PID USER     PR  NI CPU% S  #THR     VSS     RSS PCY Name
 3115 shell    20   0   8% R     1   9120K   1892K  fg top
...

2) 安卓高版本（Android 8.0 SDK >= 26）

安卓高版本使用 top -d 1 -n 1 ，查看idle的值，此为闲置CPU百分比。如idle为700%，总cpu为800%，模拟40%占用的场景：800%*60%=480%，700%-480%=220%，220%/8核=27.5%，所以模拟工具应该增加28%的负荷（参数不支持小数，向上取整），执行命令后输出信息形式如下：

Tasks: 406 total,   7 running, 390 sleeping,   0 stopped,   2 zombie            
Mem:   1858340k total,  1817892k used,    40448k free,    97196k buffers        
Swap:   520908k total,   206340k used,   314568k free,   367388k cached         
600%cpu  37%user  69%nice  57%sys 429%idle   0%iow   6%irq   3%sirq   0%host    
  PID USER         PR  NI VIRT  RES  SHR S[%CPU] %MEM     TIME+ ARGS            
26051 u0_a88       20   0 2.1G 143M  34M S 82.8   7.8 180:04.80 com.google.andr+
...

对应的Python实现如下：

# coding=utf-8
import subprocess
def isSDKHigh(serial=None):
    """
    判断SDK是否为8.0以上的版本
    :return:
    """
    try:
        if not serial:
            proc = shell("adb shell getprop ro.build.version.sdk")
        else:
            proc = shell("adb -s {} shell getprop ro.build.version.sdk".format(serial))
        stdout, stderr = proc.communicate()
        print "isSDKHigh stdout={}".format(stdout.strip())
        if int(stdout.strip()) >= 26:
            return True
        return False
    except Exception, e:
        print e.message
        return None

def getCpuCoreNum(serial=None):
    """
    获取CPU核数
    :param serial:
    :return: cpu core number
    """
    try:
        if not serial:
            proc = shell("adb shell cat /proc/cpuinfo| grep \"processor\"| wc -l")
        else:
            proc = shell("adb -s {} shell cat /proc/cpuinfo| grep \"processor\"| wc -l".format(serial))
        stdout, stderr = proc.communicate()
        return int(stdout.strip())
    except Exception, e:
        print e.message
        return None

def getCPUMessage(stdout, target, serial=None):
    """
    计算还需要填充多少CPU
    :param stdout: CPU信息
    :param target: 测试场景需要的CPU值
    :param serial: 设备序列号
    :return: 剩余需要填充的值
    """
    messageArr = stdout.split("\n");
    if isSDKHigh(serial):
        for item in messageArr:
            if item.find("cpu") > -1 and item.find("idle") > -1:
                nums = re.findall('\d+', item)
                coreNum = getCpuCoreNum(serial)
                total = int(nums[0])
                idle = int(nums[4])
                print "total={} idle={} coreNum={}".format(total, idle, coreNum)
                return (idle - (total - total * target/100))/coreNum
    else:
        for item in messageArr:
            if item.find("User") > -1 and item.find("System") > -1:
                nums = re.findall('\d+', item)
                print "userRate={} systemRate={}".format(nums[0], nums[1])
                return target - int(nums[0]) - int(nums[1])

def getCPURate(target, serial=None):
    """
    获取系统CPU信息，并计算需要填充的CPU数据
    :param target: 测试场景需要的CPU值
    :param serial: 设备序列号
    :return: 剩余需要填充的值
    """
    try:
        if not serial:
            proc = shell("adb shell top -d 1 -n 1")
        else:
            proc = shell("adb -s {} shell top -d 1 -n 1".format(serial))
        stdout, stderr = proc.communicate()
        # print "getCPURate stdout={}".format(stdout)
        return getCPUMessage(stdout, target, serial)
    except Exception, e:
        print e.message
        return None

2. Python stress-ng 调用实现

这边要注意以下四点：

• 因为在测试的过程中stress-ng是持续在运行的，这边用proc.stdin的方式调用命令，且不加proc.communicate()，stress-ng调用之后持续在运行并不会返回任何信息，如果调用communicate()脚本会卡在这边等待信息返回。
• 在Android 7.0开始以上的版本是调用adb shell killall -9 stress-ng来杀stress-ng进程，在7.0以下要加上busybox参数。
• stress-ng只支持Android 5.0开始的版本，5.0以下的版本不支持，执行的时候会报kernel版本太老的FATAL异常。
• stress-ng CPU压力模拟只会影响到用户态，不影响内核态。

具体Python粗略实现逻辑如下，可根据自身的需求做修改调整。

# coding=utf-8
import subprocess

def shell(cmds):
    """
    :param cmds: adb shell commands string
    :return:
        a subprocess
    """
    try:
        proc = subprocess.Popen(
            cmds,
            stdin=subprocess.PIPE,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            shell=True
        )
        return proc
    except Exception as e:
        raise e

def getSDK(serial=None):
    """
    获取Android SDK版本参数
    :return: int
    """
    try:
        if not serial:
            proc = shell("adb shell getprop ro.build.version.sdk")
        else:
            proc = shell("adb -s {} shell getprop ro.build.version.sdk".format(serial))
        stdout, stderr = proc.communicate()
        print "getSDK stdout={} stderr={}".format(stdout.strip(), stderr)
        return int(stdout.strip())
    except Exception, e:
        print e.message
        return None

def initStressNg(path, serial=None):
    """
    Push stress-ng可执行文件到/data/local/tmp目录，加755权限
    :param path:
    :param serial:
    :return:
    """
    try:
        if not serial:
            proc = shell("adb push {} /data/local/tmp".format(path))
        else:
            proc = shell("adb -s {} push {} /data/local/tmp".format(serial, path))
        stdout, stderr = proc.communicate()
        print "initStressNg push stdout:{} stderr:{}".format(stdout, stderr)
        cmds = [
            "chmod 755 /data/local/tmp/stress-ng",
            "exit"
        ]
        if not serial:
            proc = shell("adb shell")
        else:
            proc = shell("adb -s {} shell".format(serial))
        stdout, stderr = proc.communicate(("\n".join(cmds) + "\n"))
        print "initStressNg chmod stdout:{} stderr:{}".format(stdout, stderr)
    except Exception, e:
        print e.message

def startStressNg(rate):
    """
    开始stress-ng模拟
    :param rate: 压测要模拟的CPU值
    :return:
    """
    try:
        proc = shell("adb shell")
        proc.stdin.write("cd /data/local/tmp/\n")
        proc.stdin.write("./stress-ng -c 0 -l {}\n".format(rate))
    except Exception, e:
        print e.message

def stopStressNg():
    """
    停止stress-ng模拟
    :return:
    """
    try:
        if not serial:
            # Android 6.0 以上的版本
            if getSDK(serial) > 23:
                proc = shell("adb shell killall -9 stress-ng")
            else:
                proc = shell("adb shell busybox killall -9 stress-ng")
        else:
            if getSDK(serial) > 23:
                proc = shell("adb -s {} shell killall -9 stress-ng".format(serial))
            else:
                proc = shell("adb -s {} busybox shell killall -9 stress-ng".format(serial))
        stdout, stderr = proc.communicate()
        print "stopStressNg stdout:{}".format(stdout)
    except Exception, e:
        print e.message

if __name__ == '__main__':
    # 替换成自己的放置路径
    stressNgPath = "/Users/chenwenguan/Documents/stress-ng"
    initStressNg(stressNgPath)
    # 模拟系统80%CPU占用
    startStressNg(80)
    # do something
    stopStressNg()

至于内存、IO或其他场景的模拟后续有实现再做补充，有需要自行参考以下文章的内容去实现：

系统压力测试工具-stress-ng

WIKI Ubuntu stress-ng

Kernel Ubuntu stress-ng

TIPS

Python实现获取adb logcat日志输出

# coding=utf-8
import subprocess

def adb_logcat():
    """
    获取adb 日志输出信息
    :param cmds:
    :return:
    """
    try:
        proc = subprocess.Popen(
            ['adb', 'logcat', '-v', 'time'],
            stdout=subprocess.PIPE
        )
        # 不能用communicate(),也会卡住
        # stdout, stderr = proc.communicate()
        it = iter(proc.stdout.readline, 'b')
        while(True):
            line = it.__iter__().next()
            print line
            if not line:
                break
    except Exception, e:
        raise e

扩展阅读：

Android 性能监控之CPU监控

Android 性能监控之内存监控

Python性能分析优化及耗时异常自动化监控

转载请注明出处：陈文管的博客 – Python实现Android性能数据获取及压测场景模拟

扫码或搜索：文呓

微信公众号 扫一扫关注