Python中如何用subprocess抓取adb shell top命令的实时输出并解析CPU占用率

import subprocess
import re
import threading
import time
from collections import defaultdict

class ADBTopMonitor:
    def __init__(self, device_id=None):
        self.device_id = device_id
        self.process = None
        self.is_running = False
        self.cpu_data = defaultdict(list)
        self.lock = threading.Lock()
        
    def _build_adb_command(self):
        """构建ADB命令"""
        base_cmd = ['adb']
        if self.device_id:
            base_cmd.extend(['-s', self.device_id])
        base_cmd.extend(['shell', 'top', '-d', '1', '-n', '0'])
        return base_cmd
    
    def parse_top_output(self, line):
        """解析top命令的单行输出"""
        # 匹配进程行: PID USER PR NI VIRT RES SHR S %CPU %MEM TIME+ COMMAND
        pattern = r'^\s*(\d+)\s+(\S+)\s+(\S+)\s+(\S+)\s+(\S+)\s+(\S+)\s+(\S+)\s+(\S+)\s+(\S+)\s+(\S+)\s+(\S+)\s+(.+)$'
        match = re.match(pattern, line.strip())
        
        if match:
            pid = match.group(1)
            cpu_percent = match.group(9)  # %CPU列
            process_name = match.group(12)  # COMMAND列
            
            try:
                cpu_float = float(cpu_percent.replace('%', ''))
                return {
                    'pid': pid,
                    'process': process_name,
                    'cpu_percent': cpu_float
                }
            except ValueError:
                return None
        return None
    
    def start_monitoring(self):
        """启动监控"""
        self.is_running = True
        cmd = self._build_adb_command()
        
        try:
            self.process = subprocess.Popen(
                cmd,
                stdout=subprocess.PIPE,
                stderr=subprocess.PIPE,
                text=True,
                bufsize=1,
                universal_newlines=True
            )
            
            print(f"开始监控ADB top输出 (设备: {self.device_id or '默认'})")
            print("-" * 80)
            
            # 实时读取输出
            while self.is_running and self.process.poll() is None:
                line = self.process.stdout.readline()
                if not line:
                    continue
                    
                # 跳过空行和标题行
                if not line.strip() or 'PID' in line:
                    continue
                    
                # 解析进程数据
                process_data = self.parse_top_output(line)
                if process_data:
                    with self.lock:
                        pid = process_data['pid']
                        self.cpu_data[pid].append(process_data)
                        
                    # 实时显示
                    print(f"PID: {process_data['pid']:>6} | "
                          f"进程: {process_data['process'][:20]:<20} | "
                          f"CPU: {process_data['cpu_percent']:>6.1f}%")
                        
        except KeyboardInterrupt:
            print("\n监控被用户中断")
        except Exception as e:
            print(f"监控出错: {e}")
        finally:
            self.stop_monitoring()
    
    def stop_monitoring(self):
        """停止监控"""
        self.is_running = False
        if self.process:
            self.process.terminate()
            self.process.wait()
    
    def get_top_processes(self, n=10):
        """获取CPU占用最高的n个进程"""
        with self.lock:
            # 计算每个进程的平均CPU占用
            avg_cpu = {}
            for pid, records in self.cpu_data.items():
                if records:
                    avg = sum(r['cpu_percent'] for r in records) / len(records)
                    avg_cpu[pid] = {
                        'process': records[0]['process'],
                        'avg_cpu': avg,
                        'samples': len(records)
                    }
            
            # 按CPU占用排序
            sorted_procs = sorted(avg_cpu.items(), 
                                key=lambda x: x[1]['avg_cpu'], 
                                reverse=True)
            
            return sorted_procs[:n]

def main():
    # 使用示例
    monitor = ADBTopMonitor()
    
    # 在单独线程中运行监控
    monitor_thread = threading.Thread(target=monitor.start_monitoring)
    monitor_thread.daemon = True
    monitor_thread.start()
    
    try:
        # 监控10秒后显示统计结果
        time.sleep(10)
        
        monitor.stop_monitoring()
        monitor_thread.join(timeout=2)
        
        print("\n" + "="*80)
        print("CPU占用统计 (前10名):")
        print("="*80)
        
        top_procs = monitor.get_top_processes(10)
        for i, (pid, data) in enumerate(top_procs, 1):
            print(f"{i:2}. PID: {pid:>6} | "
                  f"进程: {data['process'][:25]:<25} | "
                  f"平均CPU: {data['avg_cpu']:>6.1f}% | "
                  f"采样数: {data['samples']}")
                  
    except KeyboardInterrupt:
        print("\n程序退出")

if __name__ == "__main__":
    main()

这个方案的核心是：

1. 用subprocess.Popen启动adb shell top命令，设置bufsize=1实现实时读取
2. 正则表达式解析top输出的每一行，提取PID、进程名和CPU百分比
3. 使用线程安全的数据结构存储历史数据
4. 提供实时显示和统计功能

代码直接运行就能用，记得先确保ADB可用。如果要监控特定设备，创建ADBTopMonitor时传入设备ID就行。

总结：用subprocess管道实时读取，正则解析，线程安全存储。