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)

def logcat_thread(): time.sleep(5) while True: logcat_process = subprocess.Popen('adb logcat -v time -t 54000', stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell=True) logcat_output, logcat_error = logcat_process.communicate() if not logcat_error: with open('logcat.txt', 'w', encoding='utf-8') as f: f.write(logcat_output.decode('utf-8')) time.sleep(3600)

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