用 AI 编程工具 8 小时搭一套睡眠噪音追踪器——从凌晨惊醒到精准定位干扰源

#!/usr/bin/env python3
"""
sleep_noise_tracker.py — 最小睡眠噪音追踪器
依赖: pip install sounddevice numpy
运行: python sleep_noise_tracker.py
按 Ctrl+C 停止录音，结束后自动生成报告。
"""

import sounddevice as sd
import numpy as np
import datetime
import json
import os
import wave

# ========== 配置区（根据你的环境修改） ==========
RECORD_DURATION_HOURS = 8          # 录音时长（小时）
SAMPLE_RATE = 16000                 # 采样率，16kHz 足够捕捉环境噪音
CHANNELS = 1                        # 单声道
SLICE_SECONDS = 60                  # 每 60 秒切一个片段
NOISE_THRESHOLD_DB = 35             # 超过此分贝值视为噪音事件（需根据实际校准）
OUTPUT_DIR = "sleep_noise_data"     # 数据输出目录
# ================================================

os.makedirs(OUTPUT_DIR, exist_ok=True)

def rms_to_db(rms, ref=1.0):
    """将 RMS 值转为分贝（相对参考值）"""
    if rms <= 0:
        return -100
    return 20 * np.log10(rms / ref)

def save_wav(filename, data, sr, channels):
    """将 numpy 数组保存为 WAV 文件"""
    with wave.open(filename, "wb") as wf:
        wf.setnchannels(channels)
        wf.setsampwidth(2)  # 16-bit
        wf.setframerate(sr)
        wf.writeframes((data * 32767).astype(np.int16).tobytes())

def main():
    total_slices = int(RECORD_DURATION_HOURS * 3600 / SLICE_SECONDS)
    slice_samples = SLICE_SECONDS * SAMPLE_RATE
    events = []

    print(f"开始录音: {RECORD_DURATION_HOURS}h, 切片 {SLICE_SECONDS}s, 阈值 {NOISE_THRESHOLD_DB}dB")
    print(f"预计切片数: {total_slices}, 按 Ctrl+C 可提前结束")

    start_time = datetime.datetime.now()

    for slice_idx in range(total_slices):
        try:
            # 录一个切片
            recording = sd.rec(slice_samples, samplerate=SAMPLE_RATE,
                               channels=CHANNELS, dtype="float32")
            sd.wait()
        except KeyboardInterrupt:
            print("\n录音提前结束")
            break

        # 计算音量
        rms_val = np.sqrt(np.mean(recording ** 2))
        db_val = rms_to_db(rms_val)

        timestamp = start_time + datetime.timedelta(seconds=slice_idx * SLICE_SECONDS)
        ts_str = timestamp.strftime("%Y-%m-%d_%H-%M-%S")
        wav_path = os.path.join(OUTPUT_DIR, f"slice_{ts_str}.wav")

        # 保存音频切片
        save_wav(wav_path, recording.flatten(), SAMPLE_RATE, CHANNELS)

        # 判断是否为噪音事件
        if db_val > NOISE_THRESHOLD_DB:
            events.append({
                "timestamp": timestamp.isoformat(),
                "db": round(db_val, 1),
                "wav_file": wav_path,
            })
            print(f"  ⚡ 噪音事件 @ {ts_str} | {db_val:.1f} dB")
        else:
            print(f"  ✓ 安静 @ {ts_str} | {db_val:.1f} dB")

    # 生成报告
    report = {
        "start_time": start_time.isoformat(),
        "end_time": datetime.datetime.now().isoformat(),
        "threshold_db": NOISE_THRESHOLD_DB,
        "total_events": len(events),
        "events": events,
    }

    report_path = os.path.join(OUTPUT_DIR, "report.json")
    with open(report_path, "w") as f:
        json.dump(report, f, indent=2, ensure_ascii=False)

    print(f"\n报告已保存: {report_path}")
    print(f"共检测到 {len(events)} 个噪音事件")
    if events:
        print("噪音事件时间线:")
        for ev in events:
            print(f"  {ev['timestamp']}  {ev['db']} dB  → {ev['wav_file']}")

if __name__ == "__main__":
    main()