登录
使用pymobiledevice3进行iphone应用性能采集
2026年1月17日 16:05
执行步骤
安装依赖（未安装则执行）
pip install pymobiledevice3
启动Tunnel（系统版本>=17需要）
sudo pymobiledevice3 remote tunneld
获取性能数据（打开一个新的终端）
根据PID获取性能数据
# 获取PID
pymobiledevice3 developer dvt process-id-for-bundle-id com.xxx.xxx
# 获取指定PID的性能数据
pymobiledevice3 developer dvt sysmon process single -a pid=xxxx
获取全部进程的性能数据
pymobiledevice3 developer dvt sysmon process monitor --rsd xxxx:xxxx:xxxx::x 58524 0.01
简单方案
上面的多个步骤都是通过pymobiledevice3一个工具来实现的，因此是否可以一步就完成性能的采集？当然可以，通过深扒（并复制+改造）pymobiledevice3的源码，将所有操作封装到了一个脚本中~~~
"""
—————— 使用说明 ——————
usage: python3 ios-monitor.py [-h] [-g GAP] [-b BUNDLE_ID] [-r REPORT_HOST] [--detail] [--csv] [--debug]
ios设备性能收集
options:
-h, --help show this help message and exit
-g, --gap GAP 性能数据获取时间间隔，默认1s
-b, --bundle_id BUNDLE_ID
包名, 默认: com.mi.car.mobile
-r, --report_host REPORT_HOST
性能采集数据上报地址
--detail 输出详细信息
--csv 结果写入到CSV文件
--debug 打印debug日志
—————— 示例 ——————
# 进行性能采集，ctrl+c 终止后写入csv文件
sudo python3 ios-monitor.py --csv
# 进行性能采集，数据上报到指定服务
sudo python3 ios-monitor.py -r 127.0.0.1:9311
"""
import argparse
import asyncio
import csv
import json
import logging
import os
import signal
import sys
import time
from functools import partial
import multiprocessing
from multiprocessing import Process
from pymobiledevice3.remote.common import TunnelProtocol
from pymobiledevice3.remote.module_imports import verify_tunnel_imports
from pymobiledevice3.remote.remote_service_discovery import RemoteServiceDiscoveryService
from pymobiledevice3.services.dvt.dvt_secure_socket_proxy import DvtSecureSocketProxyService
from pymobiledevice3.services.dvt.instruments.device_info import DeviceInfo
from pymobiledevice3.services.dvt.instruments.process_control import ProcessControl
from pymobiledevice3.services.dvt.instruments.sysmontap import Sysmontap
from pymobiledevice3.tunneld.server import TunneldRunner
import requests
TUNNELD_DEFAULT_ADDRESS = ('127.0.0.1', 49151)
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
report_error_num = 0
csv_data_list = []
csv_fieldnames=['cpu', 'mem']
def run_tunneld():
""" Start Tunneld service for remote tunneling
sudo pymobiledevice3 remote tunneld
"""
if not verify_tunnel_imports():
logger.warning("verify_tunnel_imports false")
return
host = TUNNELD_DEFAULT_ADDRESS[0]
port = TUNNELD_DEFAULT_ADDRESS[1]
protocol = TunnelProtocol(TunnelProtocol.DEFAULT.value)
tunneld_runner = partial(TunneldRunner.create, host, port, protocol=protocol, usb_monitor=True,
wifi_monitor=True, usbmux_monitor=True, mobdev2_monitor=True)
tunneld_runner()
return
def process_id_for_bundle_id(lockdown, app_bundle_identifier: str = "com.mi.car.mobile"):
""" Get PID of a bundle identifier (only returns a valid value if its running). """
with DvtSecureSocketProxyService(lockdown=lockdown) as dvt:
return ProcessControl(dvt).process_identifier_for_bundle_identifier(app_bundle_identifier)
def sysmon_process_single(lockdown, pid, detail, report_host='', write_csv=False, tunnel_process=None):
""" show a single snapshot of currently running processes. """
count = 0
result = []
with DvtSecureSocketProxyService(lockdown=lockdown) as dvt:
device_info = DeviceInfo(dvt)
with Sysmontap(dvt) as sysmon:
for process_snapshot in sysmon.iter_processes():
count += 1
if count < 2:
# first sample doesn't contain an initialized value for cpuUsage
continue
for process in process_snapshot:
# print(process)
if str(process["pid"]) != str(pid):
continue
# adding "artificially" the execName field
process['execName'] = device_info.execname_for_pid(process['pid'])
result.append(process)
# exit after single snapshot
break
if len(result) == 0:
logger.info("[]")
return
cpu_usage = "%.2f" % result[0]['cpuUsage']
mem = "%.2f" % (result[0]['memResidentSize'] / 1024 / 1024)
if write_csv:
csv_data = {'cpu': cpu_usage, 'mem': mem}
csv_data_list.append(csv_data)
if report_host:
report(host=report_host, info=result[0], pid=str(pid), tunnel_process=tunnel_process)
return
if detail:
logger.info(json.dumps(result, indent=4, ensure_ascii=False))
else:
logger.info("[CPU]{} % [内存]{} MB".format(cpu_usage, mem))
def report(host: str, info: dict, pid: str, tunnel_process=None):
global report_error_num
url = 'http://%s/monitor/collect' % host
mem = info['memResidentSize'] / 1024 / 1024
data = {
'device': 'ios-%s' % pid,
'list': [0, info['cpuUsage'], 0, mem],
'app_cpu_rate': info['cpuUsage'],
'app_mem': mem,
'timestamp': int(time.time() * 1000)
}
report_err = False
try:
resp = requests.post(url=url, json=data, timeout=(0.5, 0.5))
except Exception:
report_err = True
if report_err is False and resp.status_code != 200:
report_err = True
if report_err:
report_error_num += 1
logger.warning("上报失败 %d" % report_error_num)
if report_error_num > 5:
logger.info("接收端已关闭, 监控退出")
if tunnel_process:
tunnel_process.terminate()
sys.exit(0)
else:
cpu_usage = "%.2f" % info['cpuUsage']
logger.info("report [CPU]{} % [内存]{} MB".format(cpu_usage, mem))
report_error_num = 0
def get_tunnel_addr(attemp_times=30):
url = 'http://127.0.0.1:%d/' % TUNNELD_DEFAULT_ADDRESS[1]
try_times = 0
while try_times < attemp_times:
try:
logger.info('--- 获取设备连接信息')
resp = requests.get(url=url, timeout=(1, 1)).json()
for v in resp.values():
if not v:
continue
return v[0]['tunnel-address'], v[0]['tunnel-port']
except Exception:
pass
try_times += 1
time.sleep(1)
continue
logger.warning('--- 未找到ios设备')
return None, None
def run_in_one(bundle_id: str, gap: int, detail: bool, report_host: str = '', write_csv: bool = False):
logger.info('--- 连接设备')
p = Process(target=run_tunneld, args=())
p.start()
def sys_exit(status: int = 0):
p.terminate()
# 写入csv
if len(csv_data_list) > 0:
logger.info('--- 写入CSV')
filename = 'ios-monitor-result-%d.csv' % int(time.time())
filepath = os.path.join(os.getcwd(), filename)
with open(filepath, 'w', encoding='UTF8', newline='') as f:
writer = csv.DictWriter(f, fieldnames=csv_fieldnames)
writer.writeheader()
writer.writerows(csv_data_list)
logger.info(' --- 退出 ---')
sys.exit(status)
def signal_handler(*args, **kwargs):
sys_exit(0)
time.sleep(3)
signal.signal(signal.SIGINT, signal_handler)
addr, port = get_tunnel_addr(attemp_times=30)
if not addr:
sys_exit(1)
logger.info("--- connect device: %s %d" % (addr, port))
logger.debug('start run')
rsd = RemoteServiceDiscoveryService(address=(addr, port))
logger.debug('start rsd connect')
asyncio.run(rsd.connect(), debug=True)
time.sleep(1)
logger.debug('get pid')
pid = process_id_for_bundle_id(lockdown=rsd, app_bundle_identifier=bundle_id)
logger.info('获取应用(%s)PID: %d' % (bundle_id, pid))
while True:
if not pid:
pid = process_id_for_bundle_id(lockdown=rsd, app_bundle_identifier=bundle_id)
logger.info('获取应用(%s)PID: %d' % (bundle_id, pid))
time.sleep(0.3)
continue
try:
sysmon_process_single(lockdown=rsd, pid=pid, detail=detail, report_host=report_host, write_csv=write_csv, tunnel_process=p)
time.sleep(gap/1000)
except Exception as err:
logger.error('获取性能指标失败: {}'.format(err))
addr, port = get_tunnel_addr(attemp_times=30)
if not addr:
sys_exit(1)
pid = process_id_for_bundle_id(lockdown=rsd, app_bundle_identifier=bundle_id)
logger.info('获取应用(%s)PID: %d' % (bundle_id, pid))
if __name__ == '__main__':
multiprocessing.freeze_support()
parser = argparse.ArgumentParser(description='ios设备性能收集', add_help=True)
parser.add_argument('-g', '--gap', type=int, required=False, default=1000,
help='性能数据获取时间间隔(ms)，默认1000ms')
parser.add_argument('-b', '--bundle_id', required=False, default='com.mi.car.mobile',
help='包名, 默认: com.mi.car.mobile')
parser.add_argument('-r', '--report_host', required=False, default='',
help='性能采集数据上报地址')
parser.add_argument('--detail', default=False, action='store_true',
help='输出详细信息')
parser.add_argument('--csv', default=False, action='store_true',
help='结果写入到CSV文件')
parser.add_argument('--debug', default=False, action='store_true',
help='打印debug日志')
args = parser.parse_args()
if args.debug:
logger.setLevel(logging.DEBUG)
if not args.bundle_id:
logger.error('bundle_id invalid')
sys.exit(1)
gap_ms = args.gap
# 最低200ms间隔
if gap_ms < 200:
gap_ms = 200
rpt_host = args.report_host
# 上报到本机客户端
if rpt_host in {'local', 'localhost', '*', '-', '9311'}:
rpt_host = '127.0.0.1:9311'
run_in_one(bundle_id=args.bundle_id, gap=gap_ms, detail=args.detail, report_host=rpt_host, write_csv=args.csv)
