普通视图

发现新文章,点击刷新页面。
昨天 — 2026年1月17日首页

使用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)

❌
❌