#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ 高级语音检测器 结合能量+ZCR双重检测的自适应语音检测算法 针对16000Hz采样率优化 """ import numpy as np import time from collections import deque from typing import Dict, Any, Optional import pyaudio class EnhancedVoiceDetector: """增强版语音检测器""" def __init__(self, sample_rate=16000, chunk_size=1024): self.sample_rate = sample_rate self.chunk_size = chunk_size # 历史数据窗口 self.energy_window = deque(maxlen=100) self.zcr_window = deque(maxlen=100) # 统计信息 self.energy_stats = { 'mean': 0, 'std': 0, 'min': float('inf'), 'max': 0, 'median': 0, 'q75': 0, 'q25': 0 } self.zcr_stats = { 'mean': 0, 'std': 0, 'min': float('inf'), 'max': 0, 'median': 0, 'q75': 0, 'q25': 0 } # 检测参数 self.calibration_mode = True self.calibration_samples = 0 self.required_calibration = 100 # 需要100个样本来校准 # 自适应参数 - 调整为更敏感 self.energy_multiplier = 1.0 # 能量阈值倍数（降低） self.zcr_std_multiplier = 1.0 # ZCR标准差倍数（降低） self.min_energy_threshold = 80 # 最小能量阈值（降低） self.consecutive_voice_threshold = 2 # 连续语音检测阈值（降低） self.consecutive_silence_threshold = 15 # 连续静音检测阈值（增加） # 状态跟踪 self.consecutive_voice_count = 0 self.consecutive_silence_count = 0 self.last_voice_time = 0 # 调试信息 self.debug_mode = True self.voice_count = 0 self.total_samples = 0 self._last_voice_state = False def calculate_energy(self, audio_data: bytes) -> float: """计算音频能量（RMS）""" if len(audio_data) == 0: return 0 audio_array = np.frombuffer(audio_data, dtype=np.int16) # RMS能量计算 rms = np.sqrt(np.mean(audio_array.astype(float) ** 2)) return rms def calculate_zcr(self, audio_data: bytes) -> float: """计算零交叉率""" if len(audio_data) == 0: return 0 audio_array = np.frombuffer(audio_data, dtype=np.int16) zero_crossings = np.sum(np.diff(np.sign(audio_array)) != 0) zcr = zero_crossings / len(audio_array) * self.sample_rate return zcr def update_statistics(self, energy: float, zcr: float): """更新统计信息""" self.energy_window.append(energy) self.zcr_window.append(zcr) if len(self.energy_window) >= 20: # 计算详细统计信息 energy_array = np.array(self.energy_window) zcr_array = np.array(self.zcr_window) # 基础统计 self.energy_stats['mean'] = np.mean(energy_array) self.energy_stats['std'] = np.std(energy_array) self.energy_stats['min'] = np.min(energy_array) self.energy_stats['max'] = np.max(energy_array) self.energy_stats['median'] = np.median(energy_array) self.energy_stats['q25'] = np.percentile(energy_array, 25) self.energy_stats['q75'] = np.percentile(energy_array, 75) self.zcr_stats['mean'] = np.mean(zcr_array) self.zcr_stats['std'] = np.std(zcr_array) self.zcr_stats['min'] = np.min(zcr_array) self.zcr_stats['max'] = np.max(zcr_array) self.zcr_stats['median'] = np.median(zcr_array) self.zcr_stats['q25'] = np.percentile(zcr_array, 25) self.zcr_stats['q75'] = np.percentile(zcr_array, 75) def get_adaptive_thresholds(self) -> Dict[str, float]: """获取自适应阈值""" if len(self.energy_window) < 30: # 使用更敏感的固定阈值 return { 'energy_threshold': 120, 'zcr_min': 2000, 'zcr_max': 13000 } # 计算动态能量阈值 - 使用更合理的算法 # 基于中位数和标准差，但使用更保守的倍数 base_energy_threshold = (self.energy_stats['median'] + self.energy_multiplier * self.energy_stats['std']) # 使用四分位数来避免异常值影响 q75 = self.energy_stats['q75'] q25 = self.energy_stats['q25'] iqr = q75 - q25 # 四分位距 # 基于IQR的鲁棒阈值 - 更敏感 iqr_threshold = q75 + 0.5 * iqr # 结合两种方法的阈值 - 使用更低的阈值 energy_threshold = max(self.min_energy_threshold, min(base_energy_threshold * 0.7, iqr_threshold)) # 计算动态ZCR阈值 zcr_center = self.zcr_stats['median'] zcr_spread = self.zcr_std_multiplier * self.zcr_stats['std'] # 确保ZCR范围在合理区间内 - 更宽松 zcr_min = max(1500, min(3000, zcr_center - zcr_spread)) zcr_max = min(14000, max(6000, zcr_center + zcr_spread * 2.0)) # 确保最小范围 if zcr_max - zcr_min < 2000: zcr_max = zcr_min + 2000 return { 'energy_threshold': energy_threshold, 'zcr_min': zcr_min, 'zcr_max': zcr_max } def is_voice_basic(self, energy: float, zcr: float) -> bool: """基础语音检测（单帧）""" thresholds = self.get_adaptive_thresholds() # 能量检测 energy_ok = energy > thresholds['energy_threshold'] # ZCR检测 zcr_ok = thresholds['zcr_min'] < zcr < thresholds['zcr_max'] # 双重条件 return energy_ok and zcr_ok def is_voice_advanced(self, audio_data: bytes) -> Dict[str, Any]: """高级语音检测（带状态跟踪）""" # 计算特征 energy = self.calculate_energy(audio_data) zcr = self.calculate_zcr(audio_data) # 更新统计 self.update_statistics(energy, zcr) # 总样本计数 self.total_samples += 1 # 校准模式 if self.calibration_mode: self.calibration_samples += 1 if self.calibration_samples >= self.required_calibration: self.calibration_mode = False if self.debug_mode: print(f"\n🎯 校准完成!") print(f" 能量统计: {self.energy_stats['median']:.0f}±{self.energy_stats['std']:.0f}") print(f" ZCR统计: {self.zcr_stats['median']:.0f}±{self.zcr_stats['std']:.0f}") return { 'is_voice': False, 'energy': energy, 'zcr': zcr, 'calibrating': True, 'calibration_progress': self.calibration_samples / self.required_calibration, 'confidence': 0.0 } # 基础检测 is_voice_frame = self.is_voice_basic(energy, zcr) # 状态机处理 if is_voice_frame: self.consecutive_voice_count += 1 self.consecutive_silence_count = 0 self.last_voice_time = time.time() else: self.consecutive_silence_count += 1 if self.consecutive_silence_count >= self.consecutive_silence_threshold: self.consecutive_voice_count = 0 # 最终决策（需要连续检测到语音） final_voice_detected = self.consecutive_voice_count >= self.consecutive_voice_threshold if final_voice_detected and not hasattr(self, '_last_voice_state') or not self._last_voice_state: self.voice_count += 1 # 更新最后状态 self._last_voice_state = final_voice_detected # 计算置信度 thresholds = self.get_adaptive_thresholds() energy_confidence = min(1.0, energy / thresholds['energy_threshold']) zcr_confidence = 1.0 if thresholds['zcr_min'] < zcr < thresholds['zcr_max'] else 0.0 confidence = (energy_confidence + zcr_confidence) / 2 return { 'is_voice': final_voice_detected, 'energy': energy, 'zcr': zcr, 'confidence': confidence, 'energy_threshold': thresholds['energy_threshold'], 'zcr_min': thresholds['zcr_min'], 'zcr_max': thresholds['zcr_max'], 'consecutive_voice_count': self.consecutive_voice_count, 'consecutive_silence_count': self.consecutive_silence_count, 'calibrating': False, 'voice_detection_rate': self.voice_count / self.total_samples if self.total_samples > 0 else 0 } def get_debug_info(self) -> str: """获取调试信息""" if self.calibration_mode: return f"校准中: {self.calibration_samples}/{self.required_calibration}" thresholds = self.get_adaptive_thresholds() return (f"能量阈值: {thresholds['energy_threshold']:.0f} | " f"ZCR范围: {thresholds['zcr_min']:.0f}-{thresholds['zcr_max']:.0f} | " f"检测率: {self.voice_count}/{self.total_samples} ({self.voice_count/self.total_samples*100:.1f}%)") def reset(self): """重置检测器状态""" self.energy_window.clear() self.zcr_window.clear() self.calibration_mode = True self.calibration_samples = 0 self.consecutive_voice_count = 0 self.consecutive_silence_count = 0 self.voice_count = 0 self.total_samples = 0 class VoiceDetectorTester: """语音检测器测试器""" def __init__(self): self.detector = EnhancedVoiceDetector() def run_test(self, duration=30): """运行测试""" print("🎙️ 增强版语音检测器测试") print("=" * 50) print("📊 检测算法: 能量+ZCR双重检测") print("📈 采样率: 16000Hz") print("🔄 自适应阈值: 启用") print("⏱️ 测试时长: 30秒") print("💡 请说话测试检测效果...") print("🛑 按 Ctrl+C 提前结束") print("=" * 50) try: # 初始化音频 audio = pyaudio.PyAudio() stream = audio.open( format=pyaudio.paInt16, channels=1, rate=16000, input=True, frames_per_buffer=1024 ) start_time = time.time() voice_segments = [] current_segment = None while time.time() - start_time < duration: # 读取音频数据 data = stream.read(1024, exception_on_overflow=False) # 检测语音 result = self.detector.is_voice_advanced(data) # 处理语音段 if result['is_voice']: if current_segment is None: current_segment = { 'start_time': time.time(), 'start_sample': self.detector.total_samples } else: if current_segment is not None: current_segment['end_time'] = time.time() current_segment['end_sample'] = self.detector.total_samples current_segment['duration'] = current_segment['end_time'] - current_segment['start_time'] voice_segments.append(current_segment) current_segment = None # 显示状态 if result['calibrating']: progress = result['calibration_progress'] * 100 status = f"\r🔧 校准中: {progress:.0f}% | 能量: {result['energy']:.0f} | ZCR: {result['zcr']:.0f}" else: status_icon = "🎤" if result['is_voice'] else "🔇" status_color = "\033[92m" if result['is_voice'] else "\033[90m" reset_color = "\033[0m" status = (f"{status_color}{status_icon} " f"能量: {result['energy']:.0f}/{result['energy_threshold']:.0f} | " f"ZCR: {result['zcr']:.0f} ({result['zcr_min']:.0f}-{result['zcr_max']:.0f}) | " f"置信度: {result['confidence']:.2f} | " f"连续: {result['consecutive_voice_count']}/{result['consecutive_silence_count']}{reset_color}") print(f"\r{status}", end='', flush=True) time.sleep(0.01) # 结束当前段 if current_segment is not None: current_segment['end_time'] = time.time() current_segment['duration'] = current_segment['end_time'] - current_segment['start_time'] voice_segments.append(current_segment) # 显示统计结果 print(f"\n\n📊 测试结果统计:") print(f" 总检测时长: {duration}秒") print(f" 检测到语音段: {len(voice_segments)}") print(f" 总语音时长: {sum(s['duration'] for s in voice_segments):.1f}秒") print(f" 语音占比: {sum(s['duration'] for s in voice_segments)/duration*100:.1f}%") print(f" 平均置信度: {np.mean([r['confidence'] for r in [self.detector.is_voice_advanced(b'test') for _ in range(10)]]):.2f}") if voice_segments: print(f" 平均语音段时长: {np.mean([s['duration'] for s in voice_segments]):.1f}秒") print(f" 最长语音段: {max(s['duration'] for s in voice_segments):.1f}秒") print(f"\n🎯 检测器状态:") print(f" {self.detector.get_debug_info()}") except KeyboardInterrupt: print(f"\n\n🛑 测试被用户中断") except Exception as e: print(f"\n\n❌ 测试出错: {e}") finally: try: if 'stream' in locals(): stream.stop_stream() stream.close() if 'audio' in locals(): audio.terminate() except: pass def main(): """主函数""" tester = VoiceDetectorTester() tester.run_test() if __name__ == "__main__": main()