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
        logcat_thread = Thread(target=capture_logcat)
        logcat_thread.start()


def capture_logcat():
    '抓取 logcat 并保存到本地文件'
    logcat_file = open('logcat.txt', 'w')
    process = subprocess.Popen('adb logcat -d', stdout=logcat_file, stderr=subprocess.PIPE)
    process.wait()
    logcat_file.close()


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

解释：

创建了一个新的函数 capture_logcat()，该函数使用 subprocess 模块调用 adb logcat 命令，并将其输出写入到名为 logcat.txt 的文件中。
在 record_play_thread() 函数中，添加了一个新的线程 logcat_thread 来运行 capture_logcat() 函数。
由于您需要抓取 15 小时的 logcat，您需要调整循环和线程的运行时间，确保 logcat_thread 能够持续运行 15 小时。

这段代码只是示例，您需要根据自己的需求进行调整，例如，您可能需要更改保存 logcat 文件的路径和文件名，或者更改 logcat 抓取的过滤条件等。

此外，您需要确保您的机器上安装了 adb 工具，并且您的 Android 设备已连接到您的计算机。

注意：

由于 logcat 会不断输出信息，如果长时间抓取，可能会导致文件过大。您需要定期清空 logcat.txt 文件或者使用其他方法来管理 logcat 文件大小。

希望这能帮助到您。