Aubio/AlsaAudio配置及特定声音检测Python实现技术问询

先明确你的核心需求：

• 检测用户发出的两个独立声音，每个声音持续时长约2秒
• 两个声音的间隔不能超过3秒
• 第二个声音的音调必须高于第一个
• 满足所有条件时，发送Wake-On-Lan（WOL）数据包

接下来我会帮你补全现有代码，实现完整的功能：

完整代码实现 #

import alsaaudio
import numpy as np
import aubio
import time
import threading
import socket

class Audio_watcher:
    # 常量定义
    SAMPLERATE = 44100
    WIN_S = 2048
    HOP_S = WIN_S // 4  # 合理设置帧移，平衡检测精度与性能
    PITCH_THRESHOLD = 0.7  # 音调检测置信度阈值，过滤无效噪音
    SOUND_DURATION_REQUIRED = 2  # 单个声音需要持续的秒数
    MAX_INTERVAL_BETWEEN_SOUNDS = 3  # 两个声音的最大间隔秒数

    def __init__(self, wol_mac_address):
        self.wol_mac = wol_mac_address
        # 初始化aubio音调检测器
        self.pitch_detector = aubio.pitch("default", self.WIN_S, self.HOP_S, self.SAMPLERATE)
        self.pitch_detector.set_unit("Hz")
        self.pitch_detector.set_silence(-40)  # 设置静音阈值，忽略低音量背景音

        # 检测状态变量
        self.first_sound_detected = False
        self.first_sound_pitch = 0.0
        self.first_sound_end_time = 0.0
        self.current_sound_duration = 0.0
        self.is_listening = True

        # 启动音频监听线程
        self.listen_thread = threading.Thread(target=self._listen_audio)
        self.listen_thread.start()

    def _listen_audio(self):
        # 初始化ALSA音频输入
        inp = alsaaudio.PCM(alsaaudio.PCM_CAPTURE, alsaaudio.PCM_NORMAL)
        inp.setchannels(1)  # 使用单声道减少计算量
        inp.setrate(self.SAMPLERATE)
        inp.setformat(alsaaudio.PCM_FORMAT_FLOAT_LE)
        inp.setperiodsize(self.HOP_S)

        while self.is_listening:
            length, data = inp.read()
            if length == 0:
                continue

            # 将音频字节数据转为numpy数组
            samples = np.frombuffer(data, dtype=np.float32)
            pitch = self.pitch_detector(samples)[0]
            confidence = self.pitch_detector.get_confidence()

            # 仅处理置信度足够的有效音调
            if confidence > self.PITCH_THRESHOLD and pitch > 0:
                self.current_sound_duration += self.HOP_S / self.SAMPLERATE
                # 检查当前声音是否达到要求的时长
                if self.current_sound_duration >= self.SOUND_DURATION_REQUIRED:
                    self._handle_valid_sound(pitch)
            else:
                # 未检测到有效音调，重置当前声音时长计数
                self.current_sound_duration = 0.0

    def _handle_valid_sound(self, current_pitch):
        if not self.first_sound_detected:
            # 记录第一个有效声音的信息
            self.first_sound_detected = True
            self.first_sound_pitch = current_pitch
            self.first_sound_end_time = time.time()
            print(f"第一个有效声音检测完成，音调: {current_pitch:.2f} Hz")
            # 重置计数，准备检测第二个声音
            self.current_sound_duration = 0.0
        else:
            # 检查两个声音的间隔是否符合要求
            time_since_first = time.time() - self.first_sound_end_time
            if time_since_first > self.MAX_INTERVAL_BETWEEN_SOUNDS:
                print(f"两个声音间隔超过{self.MAX_INTERVAL_BETWEEN_SOUNDS}秒，重置检测状态")
                self.first_sound_detected = False
                self.current_sound_duration = 0.0
                return

            # 检查第二个声音的音调是否更高
            if current_pitch > self.first_sound_pitch:
                print(f"第二个有效声音检测完成，音调: {current_pitch:.2f} Hz，符合所有条件！")
                # 发送WOL数据包
                self._send_wol_packet()
                # 重置状态，等待下一轮检测
                self.first_sound_detected = False
                self.current_sound_duration = 0.0
            else:
                print(f"第二个声音音调({current_pitch:.2f} Hz)低于第一个({self.first_sound_pitch:.2f} Hz)，重置检测状态")
                self.first_sound_detected = False
                self.current_sound_duration = 0.0

    def _send_wol_packet(self):
        # 构造WOL数据包：6个0xFF + 16次重复的MAC地址
        mac_bytes = bytes.fromhex(self.wol_mac.replace(":", ""))
        wol_packet = b"\xff" * 6 + mac_bytes * 16

        # 通过UDP广播发送数据包
        with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as s:
            s.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, 1)
            try:
                s.sendto(wol_packet, ("255.255.255.255", 9))  # 9是WOL默认端口
                print(f"已向MAC地址 {self.wol_mac} 发送WOL唤醒数据包")
            except Exception as e:
                print(f"发送WOL数据包失败: {str(e)}")

    def stop_listening(self):
        self.is_listening = False
        self.listen_thread.join()

# 使用示例
if __name__ == "__main__":
    # 替换为你要唤醒的设备的MAC地址
    TARGET_MAC = "AA:BB:CC:DD:EE:FF"
    watcher = Audio_watcher(TARGET_MAC)
    try:
        print("音频监听已启动，等待检测符合条件的声音...")
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        print("\n正在停止监听...")
        watcher.stop_listening()

关键功能说明 #

• 音频检测逻辑：通过aubio的音调检测器结合置信度过滤噪音，确保只处理有效声音；实时统计声音持续时长，保证每个声音达到2秒的要求
• 状态管理：用类变量跟踪检测状态，处理两个声音的时间间隔限制，不符合条件时自动重置检测流程
• WOL实现：严格按照WOL协议构造数据包，通过UDP广播发送到目标设备的MAC地址

使用注意事项 #

1. 确保目标设备已开启WOL功能（需在BIOS/系统设置中启用）
2. 运行代码可能需要管理员权限（发送UDP广播通常需要）
3. 可根据实际环境调整PITCH_THRESHOLD、SOUND_DURATION_REQUIRED等常量，优化检测精度
4. 确认ALSA音频输入设备配置正确，能正常捕获麦克风音频

内容的提问来源于stack exchange，提问作者EStt