1083 lines
41 KiB
Python
1083 lines
41 KiB
Python
"""录音引擎：设备枚举、能力探测、低延迟采集、落盘线程、自动分段。
|
||
|
||
线程模型的取舍（直接决定"有没有爆音/掉采样"）：
|
||
|
||
* **音频回调线程只做一件事**——把数据从 PortAudio 的缓冲区拷进队列。
|
||
任何文件 IO、FFT、滤波都不允许出现在回调里，否则必然丢采样；
|
||
* **写入线程**负责软件增益、线性相位低切、电平/响度计量、写 WAV、
|
||
自动分段与磁盘速度监控；
|
||
* **界面线程**只读快照，通过 ``threading.Lock`` 保护，30 Hz 刷新。
|
||
|
||
其它"高质量"保障：录音前检查可用空间、监控写盘队列积压、统计 PortAudio
|
||
xrun（溢出）次数、设备掉线提示、每秒回写文件头（崩溃后文件仍可播放）。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import os
|
||
import queue
|
||
import shutil
|
||
import threading
|
||
import time
|
||
from dataclasses import dataclass, field, replace
|
||
from datetime import datetime
|
||
from enum import Enum
|
||
from typing import Any
|
||
|
||
import numpy as np
|
||
|
||
from . import dsp
|
||
from .wavfile import WavFormat, WavWriter, _fmt_for_bitdepth, _RiffOverflow
|
||
|
||
try: # 允许在没有音频后端的机器上 import（自检/离线处理仍可用）
|
||
import sounddevice as sd
|
||
SD_IMPORT_ERROR: str | None = None
|
||
except Exception as exc: # pragma: no cover
|
||
sd = None # type: ignore[assignment]
|
||
SD_IMPORT_ERROR = str(exc)
|
||
|
||
|
||
# --------------------------------------------------------------- 工具函数
|
||
def format_bytes(n: float) -> str:
|
||
n = float(n)
|
||
for unit in ("B", "KB", "MB", "GB", "TB"):
|
||
if abs(n) < 1024.0 or unit == "TB":
|
||
return f"{int(n)} B" if unit == "B" else f"{n:.1f} {unit}"
|
||
n /= 1024.0
|
||
return f"{n:.1f} TB"
|
||
|
||
|
||
def format_duration(seconds: float) -> str:
|
||
seconds = max(0.0, float(seconds))
|
||
h = int(seconds // 3600)
|
||
m = int((seconds % 3600) // 60)
|
||
s = seconds % 60
|
||
if h:
|
||
return f"{h:d}:{m:02d}:{s:04.1f}"
|
||
return f"{m:02d}:{s:04.1f}"
|
||
|
||
|
||
def free_space_bytes(path: str) -> int:
|
||
try:
|
||
probe = os.path.abspath(path)
|
||
while probe and not os.path.exists(probe):
|
||
parent = os.path.dirname(probe)
|
||
if parent == probe:
|
||
break
|
||
probe = parent
|
||
return int(shutil.disk_usage(probe).free)
|
||
except Exception:
|
||
return -1
|
||
|
||
|
||
# ------------------------------------------------------------- 设备枚举
|
||
@dataclass
|
||
class DeviceInfo:
|
||
index: int
|
||
name: str
|
||
hostapi: str
|
||
max_input_channels: int
|
||
default_samplerate: float
|
||
default_low_input_latency: float
|
||
default_high_input_latency: float
|
||
is_default: bool = False
|
||
|
||
@property
|
||
def label(self) -> str:
|
||
return f"{self.name} [{self.hostapi}]"
|
||
|
||
@property
|
||
def quality_rank(self) -> int:
|
||
"""宿主 API 的保真度/延迟排序，越小越优先。"""
|
||
order = ("WASAPI", "WDM-KS", "ASIO", "DirectSound", "MME")
|
||
for i, key in enumerate(order):
|
||
if key.lower() in self.hostapi.lower():
|
||
return i
|
||
return len(order)
|
||
|
||
|
||
def list_input_devices() -> list[DeviceInfo]:
|
||
"""枚举所有可用的输入设备并按音质/延迟优劣排序。"""
|
||
if sd is None:
|
||
return []
|
||
devices = sd.query_devices()
|
||
apis = sd.query_hostapis()
|
||
try:
|
||
default_in = sd.default.device[0]
|
||
except Exception:
|
||
default_in = -1
|
||
out: list[DeviceInfo] = []
|
||
for idx, dev in enumerate(devices):
|
||
if int(dev.get("max_input_channels", 0)) <= 0:
|
||
continue
|
||
api = str(apis[int(dev["hostapi"])]["name"])
|
||
out.append(DeviceInfo(
|
||
index=idx,
|
||
name=str(dev["name"]).strip(),
|
||
hostapi=api,
|
||
max_input_channels=int(dev["max_input_channels"]),
|
||
default_samplerate=float(dev.get("default_samplerate") or 48000),
|
||
default_low_input_latency=float(dev.get("default_low_input_latency") or 0.0),
|
||
default_high_input_latency=float(dev.get("default_high_input_latency") or 0.0),
|
||
is_default=(idx == default_in),
|
||
))
|
||
out.sort(key=lambda d: (d.quality_rank, not d.is_default, d.name))
|
||
return out
|
||
|
||
|
||
def default_device_index() -> int | None:
|
||
devs = list_input_devices()
|
||
if not devs:
|
||
return None
|
||
for d in devs:
|
||
if d.is_default and d.quality_rank == 0:
|
||
return d.index
|
||
return devs[0].index
|
||
|
||
|
||
def probe_capabilities(device: int, *, channels: int | None = None,
|
||
candidate_rates: tuple[int, ...] = (44100, 48000, 88200, 96000, 192000),
|
||
try_exclusive: bool = True) -> dict:
|
||
"""探测设备支持的采样率/位深组合（界面上的"检测设备支持"）。"""
|
||
result: dict[str, Any] = {"device": device, "channels": 0,
|
||
"hostapi": "", "rates": {}, "errors": []}
|
||
if sd is None:
|
||
result["errors"].append("缺少 sounddevice")
|
||
return result
|
||
info = sd.query_devices(device)
|
||
max_ch = int(info["max_input_channels"])
|
||
ch = int(channels or min(2, max_ch))
|
||
api = str(sd.query_hostapis(int(info["hostapi"]))["name"])
|
||
is_wasapi = "wasapi" in api.lower()
|
||
result["channels"] = ch
|
||
result["hostapi"] = api
|
||
for rate in candidate_rates:
|
||
entry: dict[str, Any] = {"supported": False, "dtypes": [],
|
||
"exclusive": False, "error": ""}
|
||
modes = (True, False) if (is_wasapi and try_exclusive) else (False,)
|
||
for excl in modes:
|
||
for dtype in ("float32", "int16", "int32"):
|
||
try:
|
||
extra = sd.WasapiSettings(exclusive=excl) if is_wasapi else None
|
||
sd.check_input_settings(device=device, channels=ch,
|
||
samplerate=rate, dtype=dtype,
|
||
extra_settings=extra)
|
||
entry["supported"] = True
|
||
if dtype not in entry["dtypes"]:
|
||
entry["dtypes"].append(dtype)
|
||
entry["exclusive"] = entry["exclusive"] or bool(excl)
|
||
except Exception as exc:
|
||
entry["error"] = str(exc)
|
||
result["rates"][rate] = entry
|
||
return result
|
||
|
||
|
||
# --------------------------------------------------------------- 录音配置
|
||
class RecorderState(str, Enum):
|
||
IDLE = "idle"
|
||
RECORDING = "recording"
|
||
PAUSED = "paused"
|
||
STOPPING = "stopping"
|
||
ERROR = "error"
|
||
|
||
|
||
@dataclass
|
||
class RecordConfig:
|
||
device: int | None = None
|
||
samplerate: int = 48000
|
||
channels: int = 2
|
||
bit_depth: str = "24"
|
||
gain_db: float = 0.0
|
||
exclusive: bool = True
|
||
blocksize: int = 0 # 0 = 交给 PortAudio 自动选择
|
||
latency: str = "high" # 'low' | 'high'：录音优先稳定
|
||
lowcut_hz: float = 0.0 # 0 = 关闭
|
||
dither: bool = True
|
||
rf64: bool = False
|
||
output_dir: str = ""
|
||
name_template: str = "{datetime}_{device}"
|
||
split_seconds: float = 0.0 # 0 = 不按时间分段
|
||
split_megabytes: float = 0.0 # 0 = 不按体积分段
|
||
split_on_silence: bool = False
|
||
silence_threshold_dbfs: float = -50.0
|
||
silence_gap_seconds: float = 2.0
|
||
auto_stop_silence_seconds: float = 0.0 # 0 = 不自动停止
|
||
|
||
def wav_format(self) -> WavFormat:
|
||
encoding, bits = _fmt_for_bitdepth(self.bit_depth)
|
||
return WavFormat(int(self.samplerate), int(self.channels), encoding, bits)
|
||
|
||
def sanitized(self) -> "RecordConfig":
|
||
cfg = replace(self)
|
||
cfg.samplerate = int(np.clip(cfg.samplerate, 8000, 384000))
|
||
cfg.channels = int(np.clip(cfg.channels, 1, 32))
|
||
cfg.gain_db = float(np.clip(cfg.gain_db, -60.0, 60.0))
|
||
cfg.blocksize = int(np.clip(cfg.blocksize, 0, 65536))
|
||
cfg.lowcut_hz = float(np.clip(cfg.lowcut_hz, 0.0, 500.0))
|
||
cfg.split_seconds = max(0.0, float(cfg.split_seconds))
|
||
cfg.split_megabytes = max(0.0, float(cfg.split_megabytes))
|
||
cfg.silence_gap_seconds = float(np.clip(cfg.silence_gap_seconds, 0.2, 3600.0))
|
||
cfg.silence_threshold_dbfs = float(
|
||
np.clip(cfg.silence_threshold_dbfs, -120.0, -10.0))
|
||
cfg.auto_stop_silence_seconds = max(0.0, float(cfg.auto_stop_silence_seconds))
|
||
return cfg
|
||
|
||
def estimate_bytes_per_hour(self) -> float:
|
||
return self.wav_format().bytes_per_second * 3600.0
|
||
|
||
|
||
@dataclass
|
||
class Marker:
|
||
label: str
|
||
file: str
|
||
seconds: float
|
||
timestamp: str
|
||
|
||
|
||
@dataclass
|
||
class LiveStats:
|
||
state: str = RecorderState.IDLE.value
|
||
elapsed: float = 0.0
|
||
frames: int = 0
|
||
bytes_written: int = 0
|
||
current_file: str = ""
|
||
files: list[str] = field(default_factory=list)
|
||
queue_backlog: int = 0
|
||
queue_backlog_ms: float = 0.0
|
||
overflow_blocks: int = 0
|
||
xruns: int = 0
|
||
disk_write_mbps: float = 0.0
|
||
free_space: int = -1
|
||
meter: dsp.MeterSnapshot = field(default_factory=dsp.MeterSnapshot)
|
||
clips_total: int = 0
|
||
error: str = ""
|
||
markers: int = 0
|
||
paused_seconds: float = 0.0
|
||
peak_dbfs: float = dsp.SILENCE_DBFS_FLOOR
|
||
|
||
|
||
@dataclass
|
||
class TakeResult:
|
||
files: list[str] = field(default_factory=list)
|
||
duration: float = 0.0
|
||
frames: int = 0
|
||
bytes_written: int = 0
|
||
peak_dbfs: float = float("-inf")
|
||
clipped_samples: int = 0
|
||
xruns: int = 0
|
||
overflow_blocks: int = 0
|
||
markers: list[Marker] = field(default_factory=list)
|
||
config: RecordConfig | None = None
|
||
started_at: str = ""
|
||
ended_at: str = ""
|
||
device_label: str = ""
|
||
format_label: str = ""
|
||
analysis: dict | None = None
|
||
notes: list[str] = field(default_factory=list)
|
||
|
||
@property
|
||
def primary_file(self) -> str:
|
||
return self.files[0] if self.files else ""
|
||
|
||
|
||
# --------------------------------------------- 线性相位低切（时间轴对齐）
|
||
class AlignedLowCut:
|
||
"""把线性相位 FIR 的输出对齐回原始时间轴。
|
||
|
||
线性相位 FIR 有 ``latency`` 个采样的群延迟。这里在开头丢弃
|
||
``latency`` 个输出样本、停止时再补出尾部 ``latency`` 个样本，
|
||
从而保证 **写出的样本数 == 采集的样本数**，既不偏移也不丢头掉尾。
|
||
"""
|
||
|
||
def __init__(self, cutoff_hz: float, samplerate: int, channels: int):
|
||
self.coeffs = dsp.design_highpass_fir(cutoff_hz, samplerate)
|
||
self.filter = dsp.FIRFilter(self.coeffs, channels)
|
||
self.latency = self.filter.latency_samples
|
||
self.channels = channels
|
||
self._dropped = 0
|
||
|
||
@property
|
||
def taps(self) -> int:
|
||
return int(self.coeffs.size)
|
||
|
||
def process(self, block: np.ndarray) -> np.ndarray:
|
||
y = self.filter.process(block)
|
||
if self._dropped < self.latency:
|
||
drop = min(self.latency - self._dropped, y.shape[0])
|
||
self._dropped += drop
|
||
y = y[drop:]
|
||
return y
|
||
|
||
def flush(self) -> np.ndarray:
|
||
need = 2 * self.latency
|
||
if need <= 0:
|
||
return np.zeros((0, self.channels))
|
||
y = self.filter.process(np.zeros((need, self.channels)))
|
||
return y[:self.latency]
|
||
|
||
|
||
# ------------------------------------------------------------------ 录音器
|
||
class Recorder:
|
||
"""一次录音会话。``live()`` 可在任意时刻从其它线程安全调用。"""
|
||
|
||
def __init__(self, config: RecordConfig):
|
||
self.config = config.sanitized()
|
||
self.state = RecorderState.IDLE
|
||
self._lock = threading.RLock()
|
||
self._q: queue.Queue[Any] = queue.Queue(maxsize=512)
|
||
self._thread: threading.Thread | None = None
|
||
self._stream: Any = None
|
||
self._stop_flag = threading.Event()
|
||
self._pause_flag = threading.Event()
|
||
|
||
self._writer: WavWriter | None = None
|
||
self._writer_format: WavFormat | None = None
|
||
self._lowcut: AlignedLowCut | None = None
|
||
self._meter: dsp.LevelMeter | None = None
|
||
self._out_dir = self.config.output_dir or os.path.join(os.getcwd(), "recordings")
|
||
|
||
self._frames = 0
|
||
self._frames_current = 0
|
||
self._bytes = 0
|
||
self._peak = 0.0
|
||
self._start_monotonic = 0.0
|
||
self._started_at = ""
|
||
self._files: list[str] = []
|
||
self._markers: list[Marker] = []
|
||
self._seq = 0
|
||
self._notes: list[str] = []
|
||
self._overflow_blocks = 0
|
||
self._xruns = 0
|
||
self._paused_samples = 0
|
||
self._write_times: list[tuple[float, int]] = []
|
||
self._split_requested = False
|
||
self._result: TakeResult | None = None
|
||
|
||
self._silence_run = 0
|
||
|
||
# 界面波形数据
|
||
self.scope_seconds = 1.0
|
||
self._scope_len = 0
|
||
self._scope: np.ndarray | None = None
|
||
self._scope_pos = 0
|
||
self._env_bucket = 256
|
||
self._env: list[np.ndarray] = []
|
||
self._env_pending: np.ndarray | None = None
|
||
self._env_arr: np.ndarray | None = None
|
||
self._env_dirty = False
|
||
|
||
self._live = LiveStats()
|
||
self._free_space = -1
|
||
|
||
# ------------------------------------------------------------- 生命周期
|
||
def start(self) -> None:
|
||
if sd is None:
|
||
raise RuntimeError(f"音频后端不可用：{SD_IMPORT_ERROR or 'sounddevice 未安装'}")
|
||
if self.state in (RecorderState.RECORDING, RecorderState.STOPPING):
|
||
raise RuntimeError("录音已在进行中")
|
||
|
||
cfg = self.config
|
||
if cfg.device is None:
|
||
cfg.device = default_device_index()
|
||
if cfg.device is None:
|
||
raise RuntimeError("没有找到任何可用的录音输入设备")
|
||
|
||
dev = sd.query_devices(cfg.device)
|
||
max_ch = int(dev["max_input_channels"])
|
||
if max_ch <= 0:
|
||
raise RuntimeError("所选设备没有输入通道")
|
||
if cfg.channels > max_ch:
|
||
self._note(f"设备最多支持 {max_ch} 个输入通道，已自动调整为 {max_ch}")
|
||
cfg.channels = max_ch
|
||
|
||
api_name = str(sd.query_hostapis(int(dev["hostapi"]))["name"])
|
||
is_wasapi = "wasapi" in api_name.lower()
|
||
extra = None
|
||
if is_wasapi:
|
||
try:
|
||
extra = sd.WasapiSettings(exclusive=bool(cfg.exclusive))
|
||
except Exception:
|
||
extra = None
|
||
elif cfg.exclusive:
|
||
self._note(f"{api_name} 不支持独占模式，已按共享模式录音")
|
||
|
||
try:
|
||
sd.check_input_settings(device=cfg.device, channels=cfg.channels,
|
||
samplerate=cfg.samplerate, dtype="float32",
|
||
extra_settings=extra)
|
||
except Exception as exc:
|
||
raise RuntimeError(self._explain_open_error(exc, dev, is_wasapi)) from exc
|
||
|
||
os.makedirs(self._out_dir, exist_ok=True)
|
||
self._free_space = free_space_bytes(self._out_dir)
|
||
if 0 <= self._free_space < 512 * 1024 * 1024:
|
||
self._note(f"磁盘可用空间不足 512 MB（剩余 {format_bytes(self._free_space)}），"
|
||
"长时间录音可能中断")
|
||
|
||
self._reset_state()
|
||
self._meter = dsp.LevelMeter(cfg.samplerate, cfg.channels)
|
||
self._scope_len = max(1024, int(self.scope_seconds * cfg.samplerate))
|
||
self._scope = np.zeros((self._scope_len, cfg.channels), dtype=np.float32)
|
||
self._lowcut = AlignedLowCut(cfg.lowcut_hz, cfg.samplerate, cfg.channels) \
|
||
if cfg.lowcut_hz > 0 else None
|
||
if self._lowcut is not None:
|
||
self._note(f"已启用 {cfg.lowcut_hz:.0f} Hz 线性相位低切"
|
||
f"（{self._lowcut.taps} 抽头，群延迟 "
|
||
f"{self._lowcut.latency / cfg.samplerate * 1000:.1f} ms，"
|
||
"输出已对齐，不会丢头掉尾）")
|
||
|
||
self._open_writer()
|
||
self._started_at = datetime.now().isoformat(timespec="seconds")
|
||
self._start_monotonic = time.monotonic()
|
||
self._stop_flag.clear()
|
||
self._pause_flag.clear()
|
||
self._payload_type = np.float32
|
||
|
||
self._thread = threading.Thread(target=self._writer_loop,
|
||
name="recorder-writer", daemon=True)
|
||
self._thread.start()
|
||
|
||
try:
|
||
self._stream = sd.InputStream(
|
||
device=cfg.device,
|
||
channels=cfg.channels,
|
||
samplerate=cfg.samplerate,
|
||
dtype="float32",
|
||
blocksize=int(cfg.blocksize) or 0,
|
||
latency="low" if cfg.latency == "low" else "high",
|
||
callback=self._audio_callback,
|
||
extra_settings=extra,
|
||
never_drop_input=False,
|
||
)
|
||
self._stream.start()
|
||
except Exception as exc:
|
||
self._stop_flag.set()
|
||
self._q.put(None)
|
||
if self._thread is not None:
|
||
self._thread.join(timeout=5.0)
|
||
self._thread = None
|
||
self._close_writer()
|
||
self.state = RecorderState.ERROR
|
||
raise RuntimeError(self._explain_open_error(exc, dev, is_wasapi)) from exc
|
||
|
||
real_rate = int(getattr(self._stream, "samplerate", cfg.samplerate) or cfg.samplerate)
|
||
if real_rate != cfg.samplerate:
|
||
self._note(f"设备实际采样率为 {real_rate} Hz（请求 {cfg.samplerate} Hz）")
|
||
self.state = RecorderState.RECORDING
|
||
self._live.state = self.state.value
|
||
|
||
def _explain_open_error(self, exc: Exception, dev: Any, is_wasapi: bool) -> str:
|
||
msg = f"无法打开录音设备「{dev['name']}」：{exc}"
|
||
hints: list[str] = []
|
||
low = str(exc).lower()
|
||
if "device unavailable" in low or "busy" in low or "-9985" in low:
|
||
hints.append("设备可能正被其它程序占用（浏览器 / 会议软件 / 直播工具），请先关闭它们")
|
||
if "invalid sample rate" in low or "-9997" in low:
|
||
hints.append(f"该设备不支持 {self.config.samplerate} Hz：可改用 44100 或 48000 Hz，"
|
||
"或关闭独占模式让系统做重采样")
|
||
if is_wasapi and self.config.exclusive:
|
||
hints.append("也可以关闭「WASAPI 独占模式」再试（共享模式兼容性更好）")
|
||
if "invalid number of channels" in low:
|
||
hints.append(f"通道数超出设备能力（最多 {dev['max_input_channels']} 个）")
|
||
if "unanticipated host error" in low or "-9999" in low:
|
||
hints.append("检查 Windows 隐私设置里是否允许桌面应用访问麦克风")
|
||
if hints:
|
||
msg += "\n\n建议：\n- " + "\n- ".join(hints)
|
||
return msg
|
||
|
||
def stop(self) -> TakeResult:
|
||
with self._lock:
|
||
if self.state not in (RecorderState.RECORDING, RecorderState.PAUSED,
|
||
RecorderState.ERROR):
|
||
return self._result or TakeResult(config=self.config)
|
||
self.state = RecorderState.STOPPING
|
||
self._live.state = self.state.value
|
||
|
||
if self._stream is not None:
|
||
try:
|
||
self._stream.stop()
|
||
self._stream.close()
|
||
except Exception as exc:
|
||
self._note(f"关闭音频流时出错：{exc}")
|
||
self._stream = None
|
||
|
||
# 先置停止标志、再投毒丸：队列是 FIFO，之前排队的音频一定会先被写完
|
||
self._stop_flag.set()
|
||
try:
|
||
self._q.put_nowait(None)
|
||
except queue.Full:
|
||
pass
|
||
if self._thread is not None:
|
||
self._thread.join(timeout=30.0)
|
||
self._thread = None
|
||
|
||
result = self._result or TakeResult(config=self.config)
|
||
result.notes = list(self._notes)
|
||
if result.files:
|
||
try:
|
||
result.analysis = self._analyze(result.files)
|
||
except Exception as exc:
|
||
self._note(f"录音体检失败：{exc}")
|
||
result.notes = list(self._notes)
|
||
self._result = result
|
||
with self._lock:
|
||
self.state = RecorderState.IDLE
|
||
self._live.state = self.state.value
|
||
return result
|
||
|
||
def _analyze(self, files: list[str]) -> dict:
|
||
segments = []
|
||
for f in files:
|
||
try:
|
||
if os.path.getsize(f) <= 44:
|
||
continue
|
||
segments.append(dsp.analyze_file(f))
|
||
except Exception:
|
||
continue
|
||
return _merge_analysis(segments) or {}
|
||
|
||
def pause(self) -> None:
|
||
if self.state == RecorderState.RECORDING:
|
||
self._pause_flag.set()
|
||
self.state = RecorderState.PAUSED
|
||
self._live.state = self.state.value
|
||
|
||
def resume(self) -> None:
|
||
if self.state == RecorderState.PAUSED:
|
||
self._pause_flag.clear()
|
||
self.state = RecorderState.RECORDING
|
||
self._live.state = self.state.value
|
||
|
||
def toggle_pause(self) -> None:
|
||
if self.state == RecorderState.RECORDING:
|
||
self.pause()
|
||
elif self.state == RecorderState.PAUSED:
|
||
self.resume()
|
||
|
||
def split_now(self) -> None:
|
||
with self._lock:
|
||
self._split_requested = True
|
||
|
||
def add_marker(self, label: str = "") -> Marker | None:
|
||
with self._lock:
|
||
if self.state not in (RecorderState.RECORDING, RecorderState.PAUSED):
|
||
return None
|
||
m = Marker(
|
||
label=label or f"标记 {len(self._markers) + 1}",
|
||
file=os.path.basename(self._writer.path) if self._writer else "",
|
||
seconds=self._frames_current / max(1, self.config.samplerate),
|
||
timestamp=datetime.now().isoformat(timespec="milliseconds"),
|
||
)
|
||
self._markers.append(m)
|
||
self._live.markers = len(self._markers)
|
||
return m
|
||
|
||
def shutdown(self) -> None:
|
||
"""程序退出兜底：确保流关闭、文件收尾。"""
|
||
try:
|
||
if self.state in (RecorderState.RECORDING, RecorderState.PAUSED):
|
||
self.stop()
|
||
except Exception:
|
||
pass
|
||
|
||
# ------------------------------------------------------------- 音频回调
|
||
def _audio_callback(self, indata, frames, time_info, status) -> None: # noqa: ANN001
|
||
"""PortAudio 回调：只做拷贝 + 入队，绝不做任何重活。"""
|
||
if status is not None and getattr(status, "input_overflow", False):
|
||
with self._lock:
|
||
self._xruns += 1
|
||
if self._pause_flag.is_set():
|
||
with self._lock:
|
||
self._paused_samples += frames
|
||
return
|
||
try:
|
||
self._q.put_nowait(np.array(indata, dtype=np.float32, copy=True))
|
||
except queue.Full:
|
||
with self._lock:
|
||
self._overflow_blocks += 1
|
||
|
||
# ------------------------------------------------------------- 写入线程
|
||
def _writer_loop(self) -> None:
|
||
q = self._q
|
||
last_checkpoint = time.monotonic()
|
||
while True:
|
||
try:
|
||
block = q.get(timeout=0.25)
|
||
except queue.Empty:
|
||
if self._stop_flag.is_set():
|
||
break
|
||
self._refresh_backlog()
|
||
continue
|
||
if block is None:
|
||
break
|
||
try:
|
||
self._handle_block(block)
|
||
except _RiffOverflow:
|
||
self._note("已达 RIFF 4 GiB 上限，自动分段继续录音")
|
||
self._roll_over()
|
||
except Exception as exc:
|
||
self._note(f"写入文件失败：{exc}")
|
||
self._fail(f"写入文件失败：{exc}")
|
||
break
|
||
|
||
now = time.monotonic()
|
||
if now - last_checkpoint > 1.0:
|
||
last_checkpoint = now
|
||
if self._writer is not None:
|
||
try:
|
||
self._writer.checkpoint(now)
|
||
except Exception:
|
||
pass
|
||
self._refresh_backlog()
|
||
|
||
self._finalize()
|
||
|
||
def _handle_block(self, block: np.ndarray) -> None:
|
||
cfg = self.config
|
||
if self._pause_flag.is_set():
|
||
return
|
||
|
||
x = block
|
||
if cfg.gain_db != 0.0:
|
||
x = x * dsp.db_to_lin(cfg.gain_db)
|
||
if self._lowcut is not None:
|
||
x = self._lowcut.process(x)
|
||
if x.shape[0] == 0:
|
||
return
|
||
|
||
blk_peak = float(np.max(np.abs(x)))
|
||
if blk_peak > self._peak:
|
||
self._peak = blk_peak
|
||
|
||
if self._meter is not None:
|
||
self._meter.process(x)
|
||
|
||
self._push_scope(x)
|
||
self._push_envelope(x)
|
||
self._track_silence(x)
|
||
|
||
if self._writer is None:
|
||
return
|
||
n = self._writer.write(x)
|
||
self._frames += n
|
||
self._frames_current += n
|
||
written = n * (self._writer_format.frame_bytes if self._writer_format else 0)
|
||
self._bytes += written
|
||
self._write_times.append((time.monotonic(), written))
|
||
if len(self._write_times) > 400:
|
||
del self._write_times[:200]
|
||
|
||
with self._lock:
|
||
self._live.frames = self._frames
|
||
self._live.bytes_written = self._bytes
|
||
self._live.elapsed = self._frames / max(1, cfg.samplerate)
|
||
self._live.peak_dbfs = dsp.lin_to_db(self._peak)
|
||
if self._writer is not None:
|
||
self._live.current_file = self._writer.path
|
||
self._live.clips_total = self._writer.clipped_samples
|
||
if self._meter is not None:
|
||
self._live.meter = self._meter.snapshot()
|
||
|
||
if self._should_split():
|
||
self._roll_over()
|
||
|
||
# --------------------------------------------------- 静音 / 分段 / 轮转
|
||
def _engine_stop(self, reason: str) -> None:
|
||
"""由引擎内部主动结束录音（例如静音自动停止 / 写入失败）。
|
||
|
||
必须同时关闭 PortAudio 流：只置停止标志的话，回调会继续往队列里塞数据，
|
||
而写入线程已经退出，队列很快被塞满并开始统计溢出。
|
||
"""
|
||
with self._lock:
|
||
if self._stop_flag.is_set():
|
||
return
|
||
self._notes.append(reason)
|
||
self._stop_flag.set()
|
||
stream = self._stream
|
||
if stream is not None:
|
||
try:
|
||
stream.stop()
|
||
except Exception as exc:
|
||
self._note(f"自动停止时关闭音频流出错：{exc}")
|
||
try:
|
||
self._q.put_nowait(None)
|
||
except queue.Full:
|
||
pass
|
||
|
||
def _track_silence(self, x: np.ndarray) -> None:
|
||
cfg = self.config
|
||
if not (cfg.split_on_silence or cfg.auto_stop_silence_seconds > 0):
|
||
return
|
||
thr = dsp.db_to_lin(cfg.silence_threshold_dbfs)
|
||
win = max(1, int(0.01 * cfg.samplerate))
|
||
n = (x.shape[0] // win) * win
|
||
if n == 0:
|
||
return
|
||
env = np.max(np.abs(x[:n]), axis=1).reshape(-1, win).max(axis=1)
|
||
run = self._silence_run
|
||
limit = int(cfg.silence_gap_seconds * cfg.samplerate)
|
||
stop_limit = int(cfg.auto_stop_silence_seconds * cfg.samplerate)
|
||
triggered_split = False
|
||
triggered_stop = False
|
||
for v in env:
|
||
if v < thr:
|
||
run += win
|
||
if cfg.split_on_silence and not triggered_split and run >= limit:
|
||
triggered_split = True
|
||
if cfg.auto_stop_silence_seconds > 0 and not triggered_stop \
|
||
and run >= stop_limit:
|
||
triggered_stop = True
|
||
else:
|
||
run = 0
|
||
self._silence_run = run
|
||
if triggered_split:
|
||
self._split_requested = True
|
||
if triggered_stop:
|
||
self._engine_stop(
|
||
f"静音持续 {cfg.auto_stop_silence_seconds:.1f} 秒，已自动停止录音")
|
||
|
||
def _should_split(self) -> bool:
|
||
cfg = self.config
|
||
if self._split_requested:
|
||
self._split_requested = False
|
||
return True
|
||
if cfg.split_seconds > 0 and \
|
||
self._frames_current >= cfg.split_seconds * cfg.samplerate:
|
||
return True
|
||
if cfg.split_megabytes > 0 and self._writer is not None and \
|
||
self._writer.bytes_written >= cfg.split_megabytes * 1024 * 1024:
|
||
return True
|
||
return False
|
||
|
||
def _roll_over(self) -> None:
|
||
self._close_writer()
|
||
self._frames_current = 0
|
||
self._silence_run = 0
|
||
self._open_writer()
|
||
|
||
def _open_writer(self) -> None:
|
||
path = self._next_path()
|
||
fmt = self.config.wav_format()
|
||
self._writer = WavWriter(path, fmt, dither=self.config.dither,
|
||
rf64=self.config.rf64)
|
||
self._writer_format = fmt
|
||
self._files.append(path)
|
||
with self._lock:
|
||
self._live.files = list(self._files)
|
||
self._live.current_file = path
|
||
self._note(f"写入 {os.path.basename(path)}（{fmt.describe()}）")
|
||
|
||
def _close_writer(self) -> None:
|
||
if self._writer is None:
|
||
return
|
||
try:
|
||
stats = self._writer.close()
|
||
with self._lock:
|
||
self._live.current_file = stats["path"]
|
||
self._live.clips_total = stats["clipped_samples"]
|
||
except Exception as exc:
|
||
self._note(f"收尾文件时出错：{exc}")
|
||
self._writer = None
|
||
|
||
def _next_path(self) -> str:
|
||
cfg = self.config
|
||
self._seq += 1
|
||
dev_name = "input"
|
||
if cfg.device is not None and sd is not None:
|
||
try:
|
||
dev_name = str(sd.query_devices(cfg.device)["name"]).strip()
|
||
except Exception:
|
||
pass
|
||
safe_dev = "".join(c for c in dev_name if c not in '<>:"/\\|?*').strip() or "input"
|
||
now = datetime.now()
|
||
mapping = {
|
||
"date": now.strftime("%Y%m%d"),
|
||
"time": now.strftime("%H%M%S"),
|
||
"datetime": now.strftime("%Y%m%d_%H%M%S"),
|
||
"device": safe_dev,
|
||
"sr": str(cfg.samplerate),
|
||
"bits": "f32" if "float" in str(cfg.bit_depth).lower() else str(cfg.bit_depth),
|
||
"ch": f"{cfg.channels}ch",
|
||
"seq": f"{self._seq:03d}",
|
||
}
|
||
name = cfg.name_template or "{datetime}_{device}"
|
||
for k, v in mapping.items():
|
||
name = name.replace("{" + k + "}", v)
|
||
name = "".join(c for c in name if c not in '<>:"/\\|?*').strip() or "recording"
|
||
path = os.path.join(self._out_dir, f"{name}.wav")
|
||
k = 2
|
||
while os.path.exists(path):
|
||
path = os.path.join(self._out_dir, f"{name}_{k}.wav")
|
||
k += 1
|
||
return path
|
||
|
||
# ------------------------------------------------------------ 波形数据
|
||
def _push_scope(self, x: np.ndarray) -> None:
|
||
if self._scope is None:
|
||
return
|
||
n = x.shape[0]
|
||
with self._lock:
|
||
if n >= self._scope_len:
|
||
self._scope[:] = x[-self._scope_len:]
|
||
self._scope_pos = 0
|
||
return
|
||
end = self._scope_pos + n
|
||
if end <= self._scope_len:
|
||
self._scope[self._scope_pos:end] = x
|
||
else:
|
||
first = self._scope_len - self._scope_pos
|
||
self._scope[self._scope_pos:] = x[:first]
|
||
self._scope[:n - first] = x[first:]
|
||
self._scope_pos = end % self._scope_len
|
||
|
||
def scope_data(self) -> np.ndarray:
|
||
"""最近约 1 秒的波形（按时间顺序）。"""
|
||
with self._lock:
|
||
if self._scope is None:
|
||
return np.zeros((0, self.config.channels), dtype=np.float32)
|
||
if self._scope_pos == 0:
|
||
return self._scope.copy()
|
||
return np.concatenate((self._scope[self._scope_pos:],
|
||
self._scope[:self._scope_pos]), axis=0)
|
||
|
||
def _push_envelope(self, x: np.ndarray) -> None:
|
||
data = x
|
||
if self._env_pending is not None and self._env_pending.shape[0]:
|
||
data = np.concatenate((self._env_pending, data), axis=0)
|
||
b = self._env_bucket
|
||
usable = (data.shape[0] // b) * b
|
||
if usable:
|
||
chunk = data[:usable].reshape(-1, b, data.shape[1])
|
||
env = np.stack((chunk.min(axis=1), chunk.max(axis=1)), axis=1)
|
||
with self._lock:
|
||
self._env.append(env.astype(np.float32))
|
||
self._env_dirty = True
|
||
self._env_pending = data[usable:] if usable < data.shape[0] else None
|
||
|
||
def envelope(self) -> tuple[np.ndarray, np.ndarray, int]:
|
||
"""整段录音的峰值包络 ``(mins, maxs, bucket_frames)``。
|
||
|
||
结果做了增量缓存——界面每秒调用几十次也不会重复拼接大数组。
|
||
"""
|
||
with self._lock:
|
||
if not self._env:
|
||
ch = self.config.channels
|
||
return (np.zeros((0, ch), np.float32),
|
||
np.zeros((0, ch), np.float32), self._env_bucket)
|
||
if self._env_dirty or self._env_arr is None \
|
||
or self._env_arr.shape[0] != len(self._env):
|
||
self._env_arr = np.concatenate(self._env, axis=0)
|
||
self._env_dirty = False
|
||
arr = self._env_arr
|
||
return arr[:, 0, :], arr[:, 1, :], self._env_bucket
|
||
|
||
# ------------------------------------------------------------ 只读访问
|
||
def notes(self) -> list[str]:
|
||
"""本次录音的运行日志（含自动调整、低切参数等提示）。"""
|
||
with self._lock:
|
||
return list(self._notes)
|
||
|
||
def current_markers(self) -> list[Marker]:
|
||
with self._lock:
|
||
return list(self._markers)
|
||
|
||
# ---------------------------------------------------------------- 快照
|
||
def live(self) -> LiveStats:
|
||
with self._lock:
|
||
snap = replace(self._live)
|
||
snap.files = list(self._live.files)
|
||
snap.meter = self._meter.snapshot() if self._meter is not None \
|
||
else self._live.meter
|
||
snap.free_space = self._free_space
|
||
snap.paused_seconds = self._paused_samples / max(1, self.config.samplerate)
|
||
snap.overflow_blocks = self._overflow_blocks
|
||
snap.xruns = self._xruns
|
||
return snap
|
||
|
||
def _refresh_backlog(self) -> None:
|
||
backlog = self._q.qsize()
|
||
now = time.monotonic()
|
||
recent = [(t, b) for t, b in self._write_times if now - t <= 2.0]
|
||
mbps = sum(b for _, b in recent) / 2.0 / (1024 * 1024)
|
||
block_frames = self.config.blocksize or 1024
|
||
with self._lock:
|
||
self._live.queue_backlog = backlog
|
||
self._live.queue_backlog_ms = backlog * block_frames / \
|
||
max(1, self.config.samplerate) * 1000.0
|
||
self._live.disk_write_mbps = mbps
|
||
self._live.free_space = free_space_bytes(self._out_dir)
|
||
self._live.overflow_blocks = self._overflow_blocks
|
||
self._live.xruns = self._xruns
|
||
|
||
# -------------------------------------------------------------- 收尾
|
||
def _finalize(self) -> None:
|
||
if self._lowcut is not None and self._writer is not None:
|
||
try:
|
||
tail = self._lowcut.flush()
|
||
if tail.shape[0]:
|
||
if self.config.gain_db != 0.0:
|
||
tail = tail * dsp.db_to_lin(self.config.gain_db)
|
||
n = self._writer.write(np.asarray(tail, dtype=np.float32))
|
||
self._frames += n
|
||
self._frames_current += n
|
||
self._bytes += n * (self._writer_format.frame_bytes
|
||
if self._writer_format else 0)
|
||
except Exception as exc:
|
||
self._note(f"低切尾部处理失败：{exc}")
|
||
|
||
self._close_writer()
|
||
|
||
fmt = self._writer_format or self.config.wav_format()
|
||
self._result = TakeResult(
|
||
files=list(self._files),
|
||
duration=self._frames / max(1, self.config.samplerate),
|
||
frames=self._frames,
|
||
bytes_written=self._bytes,
|
||
peak_dbfs=dsp.lin_to_db(self._peak),
|
||
clipped_samples=self._live.clips_total,
|
||
xruns=self._xruns,
|
||
overflow_blocks=self._overflow_blocks,
|
||
markers=list(self._markers),
|
||
config=self.config,
|
||
started_at=self._started_at,
|
||
ended_at=datetime.now().isoformat(timespec="seconds"),
|
||
device_label=self._device_label(),
|
||
format_label=fmt.describe(),
|
||
notes=list(self._notes),
|
||
)
|
||
with self._lock:
|
||
self._live.state = RecorderState.IDLE.value
|
||
self._live.frames = self._frames
|
||
self._live.elapsed = self._result.duration
|
||
self._live.bytes_written = self._bytes
|
||
self._live.peak_dbfs = self._result.peak_dbfs
|
||
|
||
def _device_label(self) -> str:
|
||
if self.config.device is None or sd is None:
|
||
return ""
|
||
try:
|
||
d = sd.query_devices(self.config.device)
|
||
api = sd.query_hostapis(int(d["hostapi"]))["name"]
|
||
return f"{d['name']} [{api}]"
|
||
except Exception:
|
||
return ""
|
||
|
||
def _reset_state(self) -> None:
|
||
self._frames = 0
|
||
self._frames_current = 0
|
||
self._bytes = 0
|
||
self._peak = 0.0
|
||
self._files = []
|
||
self._markers = []
|
||
self._seq = 0
|
||
self._notes = []
|
||
self._overflow_blocks = 0
|
||
self._xruns = 0
|
||
self._paused_samples = 0
|
||
self._write_times = []
|
||
self._env = []
|
||
self._env_pending = None
|
||
self._env_arr = None
|
||
self._env_dirty = False
|
||
self._silence_run = 0
|
||
self._split_requested = False
|
||
self._result = None
|
||
self._live = LiveStats(state=self.state.value)
|
||
|
||
def _note(self, msg: str) -> None:
|
||
with self._lock:
|
||
self._notes.append(msg)
|
||
|
||
def _fail(self, msg: str) -> None:
|
||
with self._lock:
|
||
self._notes.append(msg)
|
||
self._live.error = msg
|
||
self.state = RecorderState.ERROR
|
||
self._live.state = self.state.value
|
||
|
||
|
||
def _merge_analysis(segments: list[dict]) -> dict | None:
|
||
"""把多个分段文件的体检结果合并成一份总结。"""
|
||
if not segments:
|
||
return None
|
||
if len(segments) == 1:
|
||
return segments[0]
|
||
merged = dict(segments[0])
|
||
merged["segments"] = segments
|
||
merged["segment_count"] = len(segments)
|
||
merged["duration"] = round(sum(s.get("duration", 0.0) for s in segments), 3)
|
||
|
||
def _max_of(key: str) -> float | None:
|
||
vals = []
|
||
for s in segments:
|
||
v = s.get(key)
|
||
if isinstance(v, list):
|
||
vals.extend([x for x in v if x is not None])
|
||
elif v is not None:
|
||
vals.append(v)
|
||
return max(vals) if vals else None
|
||
|
||
peak = _max_of("peak_dbfs")
|
||
tp = _max_of("true_peak_dbtp")
|
||
if peak is not None:
|
||
merged["peak_dbfs"] = [peak]
|
||
if tp is not None:
|
||
merged["true_peak_dbtp"] = [tp]
|
||
merged["clipped_total"] = sum(s.get("clipped_total", 0) for s in segments)
|
||
|
||
num = 0.0
|
||
den = 0.0
|
||
for s in segments:
|
||
lufs = s.get("integrated_lufs")
|
||
if lufs is not None:
|
||
num += dsp.db_to_lin(lufs) ** 2 * s.get("duration", 0.0)
|
||
den += s.get("duration", 0.0)
|
||
merged["integrated_lufs"] = round(
|
||
float(10.0 * np.log10(num / den)), 3) if den > 0 and num > 0 else None
|
||
return merged
|
||
|
||
|
||
# ------------------------------------------------------------ 便捷函数
|
||
def quick_record(seconds: float, path: str, *, device: int | None = None,
|
||
samplerate: int = 48000, channels: int = 2,
|
||
bit_depth: str = "24", **kwargs) -> TakeResult:
|
||
"""定时录音（命令行 / 自动化的便捷入口）。``path`` 不带扩展名。"""
|
||
out_dir = os.path.dirname(os.path.abspath(path)) or os.getcwd()
|
||
base = os.path.splitext(os.path.basename(path))[0]
|
||
cfg = RecordConfig(device=device, samplerate=samplerate, channels=channels,
|
||
bit_depth=bit_depth, output_dir=out_dir,
|
||
name_template=base, **kwargs)
|
||
rec = Recorder(cfg)
|
||
rec.start()
|
||
try:
|
||
time.sleep(max(0.1, float(seconds)))
|
||
finally:
|
||
result = rec.stop()
|
||
return result
|
||
|
||
|
||
def device_summary() -> str:
|
||
"""人类可读的设备清单（命令行 --list-devices）。"""
|
||
devs = list_input_devices()
|
||
if not devs:
|
||
extra = f"（{SD_IMPORT_ERROR}）" if SD_IMPORT_ERROR else ""
|
||
return f"未发现可用输入设备。{extra}"
|
||
lines = [f"共 {len(devs)} 个录音输入设备（★ = 系统默认）：", ""]
|
||
current_api = None
|
||
for d in devs:
|
||
if d.hostapi != current_api:
|
||
current_api = d.hostapi
|
||
lines.append(f"── {current_api} ──")
|
||
mark = "★" if d.is_default else " "
|
||
lines.append(f" {mark} [{d.index:3d}] {d.name}"
|
||
f" ({d.max_input_channels} ch, 默认 {d.default_samplerate:.0f} Hz)")
|
||
lines += [
|
||
"",
|
||
"选择建议：",
|
||
" · 追求最高清晰度：WASAPI + 独占模式 + 24-bit/48 kHz（绕开系统混音器，不重采样）",
|
||
" · 录制电脑内部声音：选 WASAPI 下的“立体声混音 / Stereo Mix”，",
|
||
" 或 WDM-KS 下的“主声音捕获驱动程序”",
|
||
" · 兼容性优先：MME / DirectSound（但会经过系统混音器，可能被重采样）",
|
||
]
|
||
return "\n".join(lines)
|