Python 音频播放和 Logcat 抓取脚本

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 抓取
        subprocess.Popen('adb logcat -v time -d > ' + data_dir + 'logcat.log', shell=True)


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()

    # 循环 15 小时，抓取 logcat 日志
    for index in range(1, 15 * 60 * 6):  # 15 小时，每分钟 6 次，共 15 * 60 * 6 次

        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()

            # 清除 logcat 缓存
            subprocess.Popen('adb logcat -v time -c', shell=True)
        except Exception as e:
            continue

解释:

在 record_play_thread 函数中添加 subprocess.Popen('adb logcat -v time -d > ' + data_dir + 'logcat.log', shell=True) 命令，用来启动 logcat 抓取，并将结果保存到 data_dir 目录下的 logcat.log 文件中。
在 __main__ 函数中，将循环次数改为 15 * 60 * 6，即 15 小时，每分钟抓取 6 次。
在循环中添加 subprocess.Popen('adb logcat -v time -c', shell=True) 命令，用来在每次抓取前清除 logcat 缓存，避免文件过大。

注意：

确保您的 Android 设备已连接到电脑并已开启 USB 调试模式。
确保 data_dir 目录存在，否则脚本会报错。
subprocess.Popen 命令需要根据您的实际情况修改参数，例如 adb 命令的路径，以及 logcat 日志的保存路径。

该代码会每分钟抓取一次 Logcat 日志，并保存到 data_dir 目录下，最终会保存 15 小时的 Logcat 日志。