Python 音频播放与 Android 设备性能监控工具
import wave
import os
import time
import datetime
from threading import Thread
import subprocess
import random
from pyaudio import PyAudio, paInt16, paContinue, paComplete
class AudioPlay(PyAudio):
def __init__(self, channels=1):
super().__init__()
self.chunk = 1024 # 每个缓冲区的帧数
self.format_sample = paInt16 # 采样位数
self.channels = channels # 声道：1，单声道；2，双声道
self.fps = 44100 # 采样频率
self.input_dict = None
self.output_dict = None
self.stream = None
self.filename = '~test.wav'
self.duration = 0 # 音频时长
self.flag = False
self.kill = False
def __call__(self, filename):
'重载文件名'
self.filename = filename
def callback_output(self, in_data, frame_count, time_info, status):
'播放回调函数'
data = self.wf.readframes(frame_count)
return data, paContinue
def run(self, filename=None, name=None, sleep=0):
time.sleep(1)
'音频录制线程'
# 播放
if not filename:
raise Exception('未输入音频文件名，不能播放，请输入后再试！')
thread_2 = Thread(target=self.read_audio, args=(filename, name, sleep, ))
thread_2.start()
def read_audio(self, filename, name=None, sleep=0):
'音频播放'
output_device_index = self.get_device_index(name, False) if name else None
time.sleep(sleep)
print('play audio time, ', datetime.datetime.now())
with wave.open(filename, 'rb') as self.wf:
self.duration = self.get_duration(self.wf)
self.stream = self.open(format=self.get_format_from_width(self.wf.getsampwidth()),
channels=self.wf.getnchannels(),
rate=self.wf.getframerate(),
output=True,
output_device_index=output_device_index, # 输出设备索引
stream_callback=self.callback_output
)
self.stream.start_stream()
while self.stream.is_active():
time.sleep(0.1)
print(self.duration)
# self.terminate_run()
@staticmethod
def get_duration(wf):
'获取音频时长'
return round(wf.getnframes() / wf.getframerate(), 2)
def get_in_out_devices(self):
'获取系统输入输出设备'
self.input_dict = {}
self.output_dict = {}
for i in range(self.get_device_count()):
dev_info = self.get_device_info_by_index(i)
default_rate = int(dev_info['defaultSampleRate'])
if not dev_info['hostApi'] and default_rate == self.fps and '映射器' not in dev_info['name']:
if dev_info['maxInputChannels']:
self.input_dict[dev_info['name']] = i
elif dev_info['maxOutputChannels']:
self.output_dict[dev_info['name']] = i
def get_device_index(self, name, input_in=True):
'获取选定设备索引'
if input_in and self.input_dict:
return self.input_dict.get(name, -1)
elif not input_in and self.output_dict:
return self.output_dict.get(name, -1)
def terminate_run(self):
'结束流录制或流播放'
if self.stream:
self.stream.stop_stream()
self.stream.close()
self.terminate()
def record_play_thread():
time.sleep(5)
while True:
audio_play = AudioPlay()
audio_play.get_in_out_devices()
audio_play.run(base_dir_wakeup + '唤醒词.wav', name='test', sleep=0)
path = 'D:\python-jiaoben\performance\online\'
str_list = os.listdir(path)
print(len(str_list))
index = random.randint(0, len(str_list) - 1)
print(str_list[index])
a = str_list[index]
audio_play.run(base_dir_order + str_list[index], name='test', sleep=3)
# sleep_time = random.randint(10, 30)
# print(sleep_time, datetime.datetime.now())
time.sleep(15)
audio_play.terminate_run()
sleep_time = random.randint(10, 30)
print(sleep_time, datetime.datetime.now())
time.sleep(sleep_time)
# 抓取logcat
logcat_dir = 'D:\python-jiaoben\performance\logcat\'
get_logcat(logcat_dir)
# 等待15小时
time.sleep(15 * 60 * 60)
# 删除logcat文件
# os.remove(logcat_file)
import subprocess
import time
def get_logcat(dir_path):
timestamp = time.strftime('%Y%m%d_%H%M%S', time.localtime())
file_name = dir_path + 'logcat_{}.txt'.format(timestamp)
cmd = 'adb logcat -v time > {}'.format(file_name)
p = subprocess.Popen(cmd, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
p.wait()
return file_name
if __name__ == '__main__':
import adbUtil as adb_device
import threadMemCpu as thread
base_dir_wakeup = r'D:\python-jiaoben\performance\'
base_dir_order = r'D:\python-jiaoben\performance\online\'
data_dir = r'D:\python-jiaoben\performance\android\0421\'
if not os.path.exists(data_dir):
os.makedirs(data_dir)
thread_play = Thread(target=record_play_thread)
thread_play.start()
for index in range(1, 13):
try:
adb = adb_device.AdbDevice()
true = thread.threadmem(adb, 'cpu_log', 'mem_log', 'view', 'com.tencent.tvs.assistant.sample')
true.only_catch_memory()
print('adb time, ', datetime.datetime.now())
# p1 = subprocess.Popen('adb shell top -b -d 1 -n 60 | grep SmartSpeaker', stdout=subprocess.PIPE, stderr=subprocess.PIPE)
pp = subprocess.Popen('adb shell top -b -d 10 -n 360 | grep com.tencent.tvs.assistant.sample', stdout=subprocess.PIPE, stderr=subprocess.PIPE)
xingnneg_data = pp.stdout.read()
if xingnneg_data:
out_file = open(data_dir + str(index) + '.log', 'w', encoding='utf-8')
out_file.write(xingnneg_data.decode('utf-8'))
out_file.close()
except Exception as e:
continue
使用方法：
- 将代码保存为 Python 文件（例如：audio_monitor.py）。
- 确保已安装所需的 Python 库：
pyaudio、adb、threadMemCpu。 - 修改代码中的文件路径：
base_dir_wakeup：唤醒词音频文件路径base_dir_order：指令音频文件路径data_dir：保存性能数据的文件路径logcat_dir：保存 logcat 日志的文件路径
- 运行 Python 文件：
python audio_monitor.py
功能说明：
- 音频播放：播放唤醒词和指令音频，并记录音频播放时间。
- Android 设备性能监控：监控设备 CPU、内存使用率，并保存数据到本地。
- logcat 抓取：每 15 小时抓取一次 logcat 日志并保存到本地。
注意事项：
- 确保 Android 设备已连接到电脑并已开启 USB 调试模式。
- logcat 文件会不断增大，建议定期删除或设置自动删除机制。
- 代码中使用了
adb命令，请确保已配置好 Android 开发环境。 threadMemCpu库可以从 GitHub 上下载：https://github.com/xxx/threadMemCpu
优化说明：
- 将代码中的双引号改为单引号，方便搜索引擎识别。
- 添加标题、描述和关键词，方便搜索引擎收录。
- 优化代码结构，使其更易读和理解。
- 添加注释说明代码功能，方便他人理解。
- 提供使用方法和注意事项，方便用户使用。
- 使用 subprocess 模块调用 adb 命令，并使用线程定时抓取 logcat 日志。
- 提供代码示例和详细说明，方便用户快速上手。
- 提供相关库的下载地址，方便用户获取所需资源。
希望本工具能够帮助您进行 Android 设备性能监控和音频播放测试！
原文地址: https://www.cveoy.top/t/topic/nAXj 著作权归作者所有。请勿转载和采集!