"""录音引擎：设备枚举、能力探测、低延迟采集、落盘线程、自动分段。 线程模型的取舍（直接决定"有没有爆音/掉采样"）： * **音频回调线程只做一件事**——把数据从 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)