1472 lines
63 KiB
Python
1472 lines
63 KiB
Python
"""主窗口：设备选择、参数面板、实时计量、体检报告、后期处理与导出。"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import os
|
||
import traceback
|
||
|
||
import numpy as np
|
||
from PyQt5.QtCore import QSettings, Qt, QThread, QTimer, QUrl, pyqtSignal
|
||
from PyQt5.QtGui import QDesktopServices, QFont, QKeySequence
|
||
from PyQt5.QtWidgets import (QAction, QApplication, QCheckBox, QComboBox,
|
||
QDialog, QDialogButtonBox, QDoubleSpinBox,
|
||
QFileDialog, QFormLayout, QFrame, QGridLayout,
|
||
QGroupBox, QHBoxLayout, QLabel, QLineEdit,
|
||
QMainWindow, QMessageBox, QPlainTextEdit,
|
||
QProgressBar, QPushButton, QScrollArea,
|
||
QShortcut, QSizePolicy, QSlider, QSpinBox,
|
||
QSplitter, QStatusBar, QVBoxLayout, QWidget)
|
||
|
||
from . import dsp, engine, post
|
||
from .engine import (RecordConfig, Recorder, RecorderState, TakeResult,
|
||
format_bytes, format_duration, free_space_bytes,
|
||
list_input_devices)
|
||
from .widgets import (BORDER, HistoryPlot, LevelMeter, LoudnessBar,
|
||
MarkerRail, OK, PANEL, Scope, StatusPill, TEXT,
|
||
TEXT_DIM, WARN, RecordButton)
|
||
|
||
BIT_DEPTHS = [
|
||
("24 位 PCM（推荐）", "24", "动态范围 144 dB，远超人耳与话筒本底，兼容性极好"),
|
||
("16 位 PCM", "16", "CD 标准。动态范围 96 dB，后期余量较小，建议录 24 位再降位深"),
|
||
("32 位 PCM", "32", "32 位整数，极少使用；一般选 24 位或 32 位浮点更实用"),
|
||
("32 位浮点", "float32", "录音不会因超过 0 dBFS 而削波，适合需要大量后期的素材"),
|
||
]
|
||
SAMPLE_RATES = [44100, 48000, 88200, 96000, 176400, 192000]
|
||
LOWCUT_CHOICES = [
|
||
("关闭（完全不处理）", 0.0, "保留全部低频，链路最透明"),
|
||
("40 Hz", 40.0, "去掉空调声、桌面震动等极低频隆隆声"),
|
||
("60 Hz", 60.0, "男声录音推荐"),
|
||
("80 Hz", 80.0, "人声通用推荐（自检用的就是这一档）"),
|
||
("120 Hz", 120.0, "去除近讲喷麦造成的低频堆积"),
|
||
("180 Hz", 180.0, "播客/语音，声音最干净但会明显变薄"),
|
||
]
|
||
LUFS_TARGETS = [("不设定", None), ("播客 -16 LUFS", -16.0),
|
||
("流媒体 -14 LUFS", -14.0), ("EBU R128 -23 LUFS", -23.0)]
|
||
|
||
|
||
def _short(text: str, limit: int = 22) -> str:
|
||
return text if len(text) <= limit else text[:limit - 1] + "…"
|
||
|
||
|
||
def _clock(seconds: float) -> str:
|
||
"""大号计时器用的紧凑格式：不足 1 小时显示 MM:SS.d，超过则显示 H:MM:SS。
|
||
|
||
这样字符串最长只有 8 个字符，28pt 粗体也能在窄窗口里完整显示，
|
||
不会出现时间被截断看不清的情况。
|
||
"""
|
||
s = max(0.0, float(seconds))
|
||
h = int(s // 3600)
|
||
if h:
|
||
return f"{h}:{int((s % 3600) // 60):02d}:{int(s % 60):02d}"
|
||
return f"{int(s // 60):02d}:{s % 60:04.1f}"
|
||
|
||
|
||
# ------------------------------------------------------------------ 后台任务
|
||
class Task(QThread):
|
||
progress = pyqtSignal(str)
|
||
done = pyqtSignal(object)
|
||
failed = pyqtSignal(str)
|
||
|
||
def __init__(self, fn, *args, **kwargs):
|
||
super().__init__()
|
||
self._fn = fn
|
||
self._args = args
|
||
self._kwargs = kwargs
|
||
|
||
def run(self) -> None: # noqa: D102
|
||
fn = self._fn
|
||
try:
|
||
accepts_progress = self._accepts_progress(fn)
|
||
except (TypeError, ValueError):
|
||
accepts_progress = False
|
||
try:
|
||
if accepts_progress:
|
||
result = fn(*self._args,
|
||
progress=lambda m: self.progress.emit(m),
|
||
**self._kwargs)
|
||
else:
|
||
result = fn(*self._args, **self._kwargs)
|
||
except Exception as exc: # pragma: no cover
|
||
self.failed.emit(f"{exc}\n\n{traceback.format_exc()}")
|
||
return
|
||
self.done.emit(result)
|
||
|
||
@staticmethod
|
||
def _accepts_progress(fn) -> bool:
|
||
import inspect
|
||
params = inspect.signature(fn).parameters
|
||
if "progress" in params:
|
||
return True
|
||
return any(p.kind == p.VAR_KEYWORD for p in params.values())
|
||
|
||
|
||
# ------------------------------------------------------------------ 报告窗口
|
||
class ReportDialog(QDialog):
|
||
"""录音体检报告 + 后续操作入口。"""
|
||
|
||
def __init__(self, result: TakeResult, parent=None):
|
||
super().__init__(parent)
|
||
self.result = result
|
||
self.setWindowTitle("录音完成 · 质量体检")
|
||
self.resize(760, 620)
|
||
lay = QVBoxLayout(self)
|
||
|
||
head = QLabel(self._headline(result))
|
||
head.setStyleSheet(f"color:{TEXT}; font-size:15px; font-weight:600;")
|
||
head.setWordWrap(True)
|
||
lay.addWidget(head)
|
||
|
||
self.view = QPlainTextEdit()
|
||
self.view.setReadOnly(True)
|
||
self.view.setPlainText(post.render_report(result))
|
||
self.view.setStyleSheet(
|
||
f"background:{PANEL}; color:{TEXT}; border:1px solid {BORDER};"
|
||
"border-radius:6px; padding:8px;")
|
||
font = QFont("Consolas")
|
||
font.setStyleHint(QFont.Monospace)
|
||
font.setPointSize(10)
|
||
self.view.setFont(font)
|
||
lay.addWidget(self.view, 1)
|
||
|
||
row = QHBoxLayout()
|
||
for text, slot in (("打开文件位置", self._open_folder),
|
||
("打开波形图", self._open_preview),
|
||
("生成波形图与报告", self._write_extras),
|
||
("后期处理…", self._process),
|
||
("导出为其它格式…", self._export)):
|
||
btn = QPushButton(text)
|
||
btn.clicked.connect(slot)
|
||
row.addWidget(btn)
|
||
row.addStretch(1)
|
||
close = QPushButton("关闭")
|
||
close.clicked.connect(self.accept)
|
||
row.addWidget(close)
|
||
lay.addLayout(row)
|
||
|
||
@staticmethod
|
||
def _headline(result: TakeResult) -> str:
|
||
an = result.analysis or {}
|
||
peak = an.get("true_peak_dbtp")
|
||
peak = max([v for v in peak if v is not None], default=None) \
|
||
if isinstance(peak, list) else peak
|
||
parts = [f"已保存 {len(result.files)} 个文件",
|
||
f"总时长 {format_duration(result.duration)}",
|
||
f"数据量 {format_bytes(result.bytes_written)}"]
|
||
if peak is not None:
|
||
parts.append(f"真峰值 {peak:+.1f} dBTP")
|
||
if an.get("integrated_lufs") is not None:
|
||
parts.append(f"整体响度 {an['integrated_lufs']} LUFS")
|
||
if result.clipped_samples:
|
||
parts.append(f"⚠ 削波 {result.clipped_samples} 个样本")
|
||
if result.xruns or result.overflow_blocks:
|
||
parts.append(f"⚠ 驱动 xrun {result.xruns} / 溢出 {result.overflow_blocks}")
|
||
return " · ".join(parts)
|
||
|
||
def _open_folder(self) -> None:
|
||
if self.result.primary_file:
|
||
QDesktopServices.openUrl(
|
||
QUrl.fromLocalFile(os.path.dirname(os.path.abspath(
|
||
self.result.primary_file))))
|
||
|
||
def _open_preview(self) -> None:
|
||
png = post.make_preview_for(self.result)
|
||
if png:
|
||
QDesktopServices.openUrl(QUrl.fromLocalFile(png))
|
||
else:
|
||
QMessageBox.information(self, "提示", "无法生成波形预览图。")
|
||
|
||
def _write_extras(self) -> None:
|
||
files = post.write_metadata(self.result)
|
||
png = post.make_preview_for(self.result)
|
||
if png:
|
||
files.append(png)
|
||
QMessageBox.information(self, "已生成",
|
||
"已生成：\n" + "\n".join(os.path.basename(f)
|
||
for f in files))
|
||
|
||
def _process(self) -> None:
|
||
dlg = ProcessDialog(self.result.primary_file, self)
|
||
dlg.exec_()
|
||
|
||
def _export(self) -> None:
|
||
dlg = ExportDialog(self.result.primary_file, self)
|
||
dlg.exec_()
|
||
|
||
|
||
# ------------------------------------------------------------ 后期处理窗口
|
||
class ProcessDialog(QDialog):
|
||
"""后期处理（写出新文件，绝不动原始母版）。"""
|
||
|
||
def __init__(self, src: str, parent=None):
|
||
super().__init__(parent)
|
||
self.src = src
|
||
self.setWindowTitle(f"后期处理 · {os.path.basename(src)}")
|
||
self.resize(560, 560)
|
||
lay = QVBoxLayout(self)
|
||
|
||
tip = QLabel("所有处理都会另存为新文件（*_processed.wav），原文件保持不变。")
|
||
tip.setWordWrap(True)
|
||
tip.setStyleSheet(f"color:{TEXT_DIM};")
|
||
lay.addWidget(tip)
|
||
|
||
form = QFormLayout()
|
||
self.dc = QCheckBox("去除直流偏移")
|
||
self.dc.setChecked(True)
|
||
form.addRow(self.dc)
|
||
|
||
self.lowcut = QComboBox()
|
||
for item in LOWCUT_CHOICES:
|
||
self.lowcut.addItem(item[0])
|
||
if item[2]:
|
||
self.lowcut.setItemData(self.lowcut.count() - 1, item[2],
|
||
Qt.ToolTipRole)
|
||
self.lowcut.setCurrentIndex(2)
|
||
form.addRow("低切（高通）", self.lowcut)
|
||
|
||
self.gate = QCheckBox("启用噪声门")
|
||
self.gate_thr = QDoubleSpinBox()
|
||
self.gate_thr.setRange(-100, -20)
|
||
self.gate_thr.setValue(-60)
|
||
self.gate_thr.setSuffix(" dBFS")
|
||
form.addRow(self.gate, self.gate_thr)
|
||
|
||
self.trim = QCheckBox("裁剪首尾静音")
|
||
self.trim.setChecked(True)
|
||
self.trim_thr = QDoubleSpinBox()
|
||
self.trim_thr.setRange(-90, -20)
|
||
self.trim_thr.setValue(-50)
|
||
self.trim_thr.setSuffix(" dBFS")
|
||
form.addRow(self.trim, self.trim_thr)
|
||
|
||
self.norm = QComboBox()
|
||
self.norm.addItem("不归一化", "none")
|
||
self.norm.addItem("峰值归一化到 -1 dBFS", "peak")
|
||
self.norm.addItem("响度归一化（LUFS）", "lufs")
|
||
form.addRow("归一化", self.norm)
|
||
|
||
self.norm_val = QDoubleSpinBox()
|
||
self.norm_val.setRange(-40, 0)
|
||
self.norm_val.setValue(-16)
|
||
self.norm_val.setSuffix(" LUFS")
|
||
form.addRow("目标响度", self.norm_val)
|
||
|
||
self.fade_in = QDoubleSpinBox()
|
||
self.fade_in.setRange(0, 5)
|
||
self.fade_in.setSingleStep(0.01)
|
||
self.fade_in.setValue(0.01)
|
||
self.fade_in.setSuffix(" s")
|
||
self.fade_out = QDoubleSpinBox()
|
||
self.fade_out.setRange(0, 5)
|
||
self.fade_out.setSingleStep(0.01)
|
||
self.fade_out.setValue(0.05)
|
||
self.fade_out.setSuffix(" s")
|
||
form.addRow("淡入", self.fade_in)
|
||
form.addRow("淡出", self.fade_out)
|
||
|
||
self.mono = QCheckBox("混合为单声道")
|
||
form.addRow(self.mono)
|
||
|
||
self.depth = QComboBox()
|
||
for item in BIT_DEPTHS:
|
||
self.depth.addItem(item[0], item[1])
|
||
if len(item) > 2 and item[2]:
|
||
self.depth.setItemData(self.depth.count() - 1, item[2],
|
||
Qt.ToolTipRole)
|
||
form.addRow("输出位深", self.depth)
|
||
|
||
self.dither = QCheckBox("启用 TPDF 抖动（降位深时推荐）")
|
||
self.dither.setChecked(True)
|
||
form.addRow(self.dither)
|
||
lay.addLayout(form)
|
||
|
||
self.progress = QProgressBar()
|
||
self.progress.setVisible(False)
|
||
lay.addWidget(self.progress)
|
||
|
||
self.log = QPlainTextEdit()
|
||
self.log.setReadOnly(True)
|
||
self.log.setMaximumHeight(120)
|
||
lay.addWidget(self.log)
|
||
|
||
row = QHBoxLayout()
|
||
self.run_btn = QPushButton("开始处理")
|
||
self.run_btn.clicked.connect(self._run)
|
||
row.addWidget(self.run_btn)
|
||
self.open_btn = QPushButton("打开输出目录")
|
||
self.open_btn.setEnabled(False)
|
||
self.open_btn.clicked.connect(self._open_output)
|
||
row.addWidget(self.open_btn)
|
||
row.addStretch(1)
|
||
close = QPushButton("关闭")
|
||
close.clicked.connect(self.reject)
|
||
row.addWidget(close)
|
||
lay.addLayout(row)
|
||
self._task: Task | None = None
|
||
self._last_output = ""
|
||
|
||
def _open_output(self) -> None:
|
||
if self._last_output and os.path.exists(self._last_output):
|
||
QDesktopServices.openUrl(QUrl.fromLocalFile(
|
||
os.path.dirname(os.path.abspath(self._last_output))))
|
||
|
||
def _options(self) -> post.ProcessOptions:
|
||
return post.ProcessOptions(
|
||
trim_silence=self.trim.isChecked(),
|
||
trim_threshold_dbfs=self.trim_thr.value(),
|
||
remove_dc=self.dc.isChecked(),
|
||
lowcut_hz=LOWCUT_CHOICES[self.lowcut.currentIndex()][1],
|
||
noise_gate=self.gate.isChecked(),
|
||
gate_threshold_dbfs=self.gate_thr.value(),
|
||
normalize=self.norm.currentData(),
|
||
normalize_target_lufs=self.norm_val.value(),
|
||
fade_in=self.fade_in.value(),
|
||
fade_out=self.fade_out.value(),
|
||
mono=self.mono.isChecked(),
|
||
bit_depth=self.depth.currentData(),
|
||
dither=self.dither.isChecked(),
|
||
)
|
||
|
||
def _run(self) -> None:
|
||
opts = self._options()
|
||
self.run_btn.setEnabled(False)
|
||
self.progress.setVisible(True)
|
||
self.progress.setRange(0, 0)
|
||
self.log.appendPlainText("开始处理…")
|
||
|
||
def job(progress=None):
|
||
return post.process_file(self.src, opts, progress=progress)
|
||
|
||
self._task = Task(job)
|
||
self._task.progress.connect(lambda m: self.log.appendPlainText(m))
|
||
self._task.done.connect(self._finished)
|
||
self._task.failed.connect(self._failed)
|
||
self._task.start()
|
||
|
||
def _finished(self, report: object) -> None:
|
||
self.progress.setRange(0, 100)
|
||
self.progress.setValue(100)
|
||
self.run_btn.setEnabled(True)
|
||
rep = report if isinstance(report, dict) else {}
|
||
self.log.appendPlainText("完成：" + os.path.basename(str(rep.get("output", ""))))
|
||
for s in rep.get("steps", []):
|
||
self.log.appendPlainText(" · " + str(s))
|
||
an = rep.get("analysis") or {}
|
||
if an:
|
||
self.log.appendPlainText(
|
||
f" 输出：峰值 {an.get('peak_dbfs')} dBFS，"
|
||
f"真峰值 {an.get('true_peak_dbtp')} dBTP，"
|
||
f"响度 {an.get('integrated_lufs')} LUFS")
|
||
# 注意：这里不弹模态对话框——本函数由工作线程的跨线程信号触发，
|
||
# 在信号回调里开嵌套事件循环容易造成重入崩溃。改为启用按钮由用户点击。
|
||
if rep.get("output"):
|
||
self._last_output = str(rep["output"])
|
||
self.open_btn.setEnabled(True)
|
||
self.log.appendPlainText(" → 点「打开输出目录」可直接查看结果")
|
||
|
||
def _failed(self, msg: str) -> None:
|
||
self.progress.setVisible(False)
|
||
self.run_btn.setEnabled(True)
|
||
self.log.appendPlainText("失败：" + msg.splitlines()[0])
|
||
QMessageBox.critical(self, "处理失败", msg)
|
||
|
||
|
||
class ExportDialog(QDialog):
|
||
"""导出为 FLAC / MP3 / Opus / AAC 等格式。"""
|
||
|
||
def __init__(self, src: str, parent=None):
|
||
super().__init__(parent)
|
||
self.src = src
|
||
self.setWindowTitle(f"导出 · {os.path.basename(src)}")
|
||
self.resize(520, 300)
|
||
lay = QVBoxLayout(self)
|
||
|
||
ff = post.find_ffmpeg()
|
||
info = QLabel(("已检测到 ffmpeg：" + ff) if ff else
|
||
"未检测到 ffmpeg：FLAC/MP3/Opus/AAC 导出不可用，"
|
||
"请安装 ffmpeg 并加入 PATH（WAV 导出不受影响）。")
|
||
info.setWordWrap(True)
|
||
info.setStyleSheet(f"color:{OK if ff else WARN};")
|
||
lay.addWidget(info)
|
||
|
||
form = QFormLayout()
|
||
self.preset = QComboBox()
|
||
for key, meta in post.EXPORT_PRESETS.items():
|
||
self.preset.addItem(meta["label"], key)
|
||
self.preset.setCurrentIndex(1)
|
||
form.addRow("目标格式", self.preset)
|
||
lay.addLayout(form)
|
||
|
||
self.progress = QProgressBar()
|
||
self.progress.setVisible(False)
|
||
lay.addWidget(self.progress)
|
||
self.log = QPlainTextEdit()
|
||
self.log.setReadOnly(True)
|
||
lay.addWidget(self.log, 1)
|
||
|
||
row = QHBoxLayout()
|
||
self.run_btn = QPushButton("开始导出")
|
||
self.run_btn.clicked.connect(self._run)
|
||
row.addWidget(self.run_btn)
|
||
row.addStretch(1)
|
||
close = QPushButton("关闭")
|
||
close.clicked.connect(self.reject)
|
||
row.addWidget(close)
|
||
lay.addLayout(row)
|
||
self._task: Task | None = None
|
||
|
||
def _run(self) -> None:
|
||
key = self.preset.currentData()
|
||
dst = post.default_export_path(self.src, key)
|
||
self.run_btn.setEnabled(False)
|
||
self.progress.setVisible(True)
|
||
self.progress.setRange(0, 0)
|
||
self.log.appendPlainText(f"目标文件：{os.path.basename(dst)}")
|
||
|
||
def job():
|
||
return post.export_audio(self.src, dst, key)
|
||
|
||
self._task = Task(job)
|
||
self._task.done.connect(self._finished)
|
||
# 同样不在跨线程信号里弹模态框，只写日志，由用户决定下一步
|
||
self._task.failed.connect(self._failed)
|
||
self._task.start()
|
||
|
||
def _failed(self, msg: str) -> None:
|
||
self.progress.setVisible(False)
|
||
self.run_btn.setEnabled(True)
|
||
self.log.appendPlainText("失败：" + msg.splitlines()[0])
|
||
|
||
def _finished(self, res: object) -> None:
|
||
ok, msg = res if isinstance(res, tuple) else (False, "未知结果")
|
||
self.progress.setVisible(False)
|
||
self.run_btn.setEnabled(True)
|
||
self.log.appendPlainText(("成功：" if ok else "失败：") + str(msg))
|
||
|
||
|
||
# ------------------------------------------------------------------- 主窗口
|
||
class MainWindow(QMainWindow):
|
||
def __init__(self):
|
||
super().__init__()
|
||
self.setWindowTitle("RecorderStudio · 高清录音机")
|
||
self.resize(1240, 860)
|
||
self.setMinimumSize(1120, 700)
|
||
|
||
self.recorder: Recorder | None = None
|
||
self.result: TakeResult | None = None
|
||
self._devices: list[engine.DeviceInfo] = []
|
||
self._settings = QSettings("RecorderStudio", "RecorderStudio")
|
||
self._cap_task: Task | None = None
|
||
self._calib_task: Task | None = None
|
||
self._last_files: list[str] = []
|
||
self._pulse = 0.0
|
||
self._device_signature = ""
|
||
self._peak_alerted = False
|
||
self._tick_count = 0
|
||
self._history_ready = False
|
||
|
||
self._build_ui()
|
||
self._load_settings()
|
||
self.refresh_devices()
|
||
self._install_shortcuts()
|
||
|
||
self.timer = QTimer(self)
|
||
self.timer.setInterval(33)
|
||
self.timer.timeout.connect(self._tick)
|
||
self.timer.start()
|
||
|
||
self.watchdog = QTimer(self)
|
||
self.watchdog.setInterval(3000)
|
||
self.watchdog.timeout.connect(self._check_devices)
|
||
self.watchdog.start()
|
||
|
||
self._set_idle_ui()
|
||
|
||
# ------------------------------------------------------------------ UI
|
||
def _build_ui(self) -> None:
|
||
central = QWidget()
|
||
self.setCentralWidget(central)
|
||
outer = QVBoxLayout(central)
|
||
outer.setContentsMargins(12, 10, 12, 8)
|
||
outer.setSpacing(10)
|
||
|
||
split = QSplitter(Qt.Horizontal)
|
||
split.addWidget(self._build_left_panel())
|
||
split.addWidget(self._build_right_panel())
|
||
split.setStretchFactor(0, 0)
|
||
split.setStretchFactor(1, 1)
|
||
split.setSizes([372, 808])
|
||
outer.addWidget(split, 1)
|
||
|
||
# 底部状态栏
|
||
bar = QStatusBar()
|
||
self.setStatusBar(bar)
|
||
self.status_left = QLabel("就绪")
|
||
self.status_right = QLabel("")
|
||
bar.addWidget(self.status_left, 1)
|
||
bar.addPermanentWidget(self.status_right)
|
||
|
||
self._build_menu()
|
||
|
||
def _build_menu(self) -> None:
|
||
m_file = self.menuBar().addMenu("文件(&F)")
|
||
act_dir = QAction("选择输出目录…", self)
|
||
act_dir.triggered.connect(self.choose_output_dir)
|
||
m_file.addAction(act_dir)
|
||
act_open = QAction("打开输出目录", self)
|
||
act_open.setShortcut(QKeySequence("Ctrl+O"))
|
||
act_open.triggered.connect(self.open_output_dir)
|
||
m_file.addAction(act_open)
|
||
m_file.addSeparator()
|
||
act_quit = QAction("退出", self)
|
||
act_quit.setShortcut(QKeySequence("Ctrl+Q"))
|
||
act_quit.triggered.connect(self.close)
|
||
m_file.addAction(act_quit)
|
||
|
||
m_tool = self.menuBar().addMenu("工具(&T)")
|
||
act_probe = QAction("检测设备支持能力", self)
|
||
act_probe.triggered.connect(self.probe_device)
|
||
m_tool.addAction(act_probe)
|
||
act_calib = QAction("校准本底噪声（测 3 秒静音）", self)
|
||
act_calib.triggered.connect(self.calibrate_noise)
|
||
m_tool.addAction(act_calib)
|
||
m_tool.addSeparator()
|
||
act_post = QAction("对最近的录音做后期处理…", self)
|
||
act_post.triggered.connect(self.open_process_last)
|
||
m_tool.addAction(act_post)
|
||
|
||
m_help = self.menuBar().addMenu("帮助(&H)")
|
||
act_keys = QAction("快捷键", self)
|
||
act_keys.triggered.connect(self.show_shortcuts)
|
||
m_help.addAction(act_keys)
|
||
act_about = QAction("关于 / 高清晰度录音指南", self)
|
||
act_about.triggered.connect(self.show_about)
|
||
m_help.addAction(act_about)
|
||
|
||
def _group(self, title: str) -> tuple[QGroupBox, QFormLayout]:
|
||
box = QGroupBox(title)
|
||
form = QFormLayout(box)
|
||
form.setLabelAlignment(Qt.AlignLeft | Qt.AlignVCenter)
|
||
form.setFormAlignment(Qt.AlignLeft | Qt.AlignTop)
|
||
form.setContentsMargins(10, 14, 10, 10)
|
||
form.setSpacing(7)
|
||
return box, form
|
||
|
||
def _combo(self, items: list[tuple], *, tip: str = "") -> QComboBox:
|
||
"""创建下拉框。
|
||
|
||
关键点：默认的 ``sizeHint`` 会按最长条目计算最小宽度，
|
||
在窄侧栏里会把整个面板撑宽导致右侧被裁切；
|
||
这里改用 ``AdjustToMinimumContentsLengthWithIcon`` 并限定长度，
|
||
窗口变窄时由 Qt 自动省略号处理。
|
||
"""
|
||
c = QComboBox()
|
||
c.setSizeAdjustPolicy(QComboBox.AdjustToMinimumContentsLengthWithIcon)
|
||
c.setMinimumContentsLength(9)
|
||
for item in items:
|
||
label, value = item[0], item[1]
|
||
c.addItem(label, value)
|
||
if len(item) > 2 and item[2]:
|
||
c.setItemData(c.count() - 1, item[2], Qt.ToolTipRole)
|
||
if tip:
|
||
c.setToolTip(tip)
|
||
return c
|
||
|
||
def _spin(self, *, suffix: str = "", rng: tuple[float, float] = (0, 100),
|
||
value: float = 0.0, special: str = "", decimals: int = 2,
|
||
step: float = 0.5, tip: str = "") -> QDoubleSpinBox:
|
||
"""创建数值输入框。
|
||
|
||
QDoubleSpinBox 的 ``minimumSizeHint`` 会按"最大值+后缀+微调按钮"预留宽度
|
||
（动辄 160~180 px），在窄侧栏里会把面板撑宽。这里把水平尺寸策略设为
|
||
``Ignored`` 并给一个明确的最小宽度，让它既能收缩到 88 px 又不至于不可用。
|
||
"""
|
||
s = QDoubleSpinBox()
|
||
s.setRange(*rng)
|
||
s.setDecimals(decimals)
|
||
s.setSingleStep(step)
|
||
if suffix:
|
||
s.setSuffix(suffix)
|
||
if special:
|
||
s.setSpecialValueText(special)
|
||
s.setValue(value)
|
||
s.setSizePolicy(QSizePolicy.Ignored, QSizePolicy.Fixed)
|
||
s.setMinimumWidth(88)
|
||
if tip:
|
||
s.setToolTip(tip)
|
||
return s
|
||
|
||
def _build_left_panel(self) -> QWidget:
|
||
"""左侧设置栏：四组参数 + 滚动容器。"""
|
||
panel = QWidget()
|
||
v = QVBoxLayout(panel)
|
||
v.setContentsMargins(0, 0, 6, 0)
|
||
v.setSpacing(10)
|
||
v.addWidget(self._panel_device())
|
||
v.addWidget(self._panel_format())
|
||
v.addWidget(self._panel_processing())
|
||
v.addWidget(self._panel_files())
|
||
v.addStretch(1)
|
||
|
||
scroll = QScrollArea()
|
||
scroll.setWidget(panel)
|
||
scroll.setWidgetResizable(True)
|
||
scroll.setFrameShape(QFrame.NoFrame)
|
||
scroll.setHorizontalScrollBarPolicy(Qt.ScrollBarAlwaysOff)
|
||
scroll.setMinimumWidth(330)
|
||
scroll.setMaximumWidth(430)
|
||
return scroll
|
||
|
||
def _panel_device(self) -> QGroupBox:
|
||
box, form = self._group("① 输入设备")
|
||
self.device_combo = self._combo([], tip="设备按保真度排序：WASAPI 最优先")
|
||
self.device_combo.currentIndexChanged.connect(self._on_device_changed)
|
||
form.addRow("设备", self.device_combo)
|
||
|
||
row = QHBoxLayout()
|
||
self.refresh_btn = QPushButton("刷新设备")
|
||
self.refresh_btn.clicked.connect(self.refresh_devices)
|
||
self.probe_btn = QPushButton("检测能力")
|
||
self.probe_btn.clicked.connect(self.probe_device)
|
||
row.addWidget(self.refresh_btn)
|
||
row.addWidget(self.probe_btn)
|
||
wrap = QWidget()
|
||
wrap.setLayout(row)
|
||
form.addRow("", wrap)
|
||
|
||
self.device_info = QLabel("—")
|
||
self.device_info.setWordWrap(True)
|
||
self.device_info.setStyleSheet(f"color:{TEXT_DIM}; font-size:11px;")
|
||
form.addRow(self.device_info)
|
||
return box
|
||
|
||
def _panel_format(self) -> QGroupBox:
|
||
box, form = self._group("② 录音格式")
|
||
self.rate_combo = self._combo([(f"{r} Hz", r) for r in SAMPLE_RATES],
|
||
tip="设备原生采样率无需重采样，保真度最高")
|
||
self.rate_combo.setCurrentIndex(1)
|
||
self.rate_combo.currentIndexChanged.connect(self._update_estimate)
|
||
form.addRow("采样率", self.rate_combo)
|
||
|
||
self.depth_combo = self._combo([(l, v, t) for l, v, t in BIT_DEPTHS],
|
||
tip="24 位是录音的甜点：动态范围 144 dB")
|
||
idx = self.depth_combo.findData("24")
|
||
if idx >= 0:
|
||
self.depth_combo.setCurrentIndex(idx)
|
||
self.depth_combo.currentIndexChanged.connect(self._update_estimate)
|
||
form.addRow("位深", self.depth_combo)
|
||
|
||
self.channels_spin = QSpinBox()
|
||
self.channels_spin.setRange(1, 8)
|
||
self.channels_spin.setValue(2)
|
||
self.channels_spin.setSizePolicy(QSizePolicy.Ignored, QSizePolicy.Fixed)
|
||
self.channels_spin.setMinimumWidth(88)
|
||
self.channels_spin.valueChanged.connect(self._update_estimate)
|
||
form.addRow("声道数", self.channels_spin)
|
||
|
||
self.exclusive_check = QCheckBox("WASAPI 独占模式")
|
||
self.exclusive_check.setChecked(True)
|
||
self.exclusive_check.setToolTip(
|
||
"绕开 Windows 系统混音器，不做重采样、不被其它程序改动音量，\n"
|
||
"是 Windows 上最接近“原始信号”的采集方式。")
|
||
form.addRow(self.exclusive_check)
|
||
|
||
self.blocksize_combo = self._combo(
|
||
[("自动（推荐）", 0), ("512 采样", 512), ("1024 采样", 1024),
|
||
("2048 采样", 2048), ("4096 采样（最稳）", 4096)],
|
||
tip="录音不需要低延迟，缓冲区大一点更不容易出现爆音/掉采样")
|
||
form.addRow("缓冲区", self.blocksize_combo)
|
||
|
||
self.latency_combo = self._combo(
|
||
[("高（更抗掉采样，推荐）", "high"), ("低（追求实时监听）", "low")])
|
||
form.addRow("延迟策略", self.latency_combo)
|
||
return box
|
||
|
||
def _panel_processing(self) -> QGroupBox:
|
||
box, form = self._group("③ 信号处理")
|
||
self.lowcut_combo = self._combo(LOWCUT_CHOICES,
|
||
tip="线性相位 FIR，不引入相位失真，且输出已对齐时间轴")
|
||
form.addRow("低切滤波", self.lowcut_combo)
|
||
|
||
self.gain_slider = QSlider(Qt.Horizontal)
|
||
self.gain_slider.setRange(-240, 240) # 0.1 dB 步进
|
||
self.gain_slider.setValue(0)
|
||
self.gain_slider.valueChanged.connect(self._on_gain_changed)
|
||
self.gain_label = QLabel("+0.0 dB")
|
||
self.gain_label.setMinimumWidth(66)
|
||
grow = QHBoxLayout()
|
||
grow.addWidget(self.gain_slider, 1)
|
||
grow.addWidget(self.gain_label)
|
||
gw = QWidget()
|
||
gw.setLayout(grow)
|
||
form.addRow("软件增益", gw)
|
||
|
||
self.dither_check = QCheckBox("启用 TPDF 抖动")
|
||
self.dither_check.setChecked(True)
|
||
self.dither_check.setToolTip(
|
||
"量化到 16/24 位前加入 1 LSB 峰峰值的三角分布抖动，\n"
|
||
"把“与信号相关的非线性失真”变成无关的宽带白噪，弱信号细节更干净。")
|
||
form.addRow(self.dither_check)
|
||
|
||
self.rf64_check = QCheckBox("RF64 大文件容器")
|
||
self.rf64_check.setToolTip(
|
||
"RIFF 容器上限 4 GB。开启后使用 RF64（EBU Tech 3306），\n"
|
||
"单文件可超过 4 GB；不开启时程序会在接近上限处自动分段。")
|
||
form.addRow(self.rf64_check)
|
||
|
||
self.target_combo = self._combo(LUFS_TARGETS,
|
||
tip="在响度条上画一条目标参考线")
|
||
self.target_combo.currentIndexChanged.connect(self._on_target_changed)
|
||
form.addRow("响度参考", self.target_combo)
|
||
return box
|
||
|
||
def _panel_files(self) -> QGroupBox:
|
||
box, form = self._group("④ 文件与分段")
|
||
self.dir_edit = QLineEdit()
|
||
self.dir_edit.textChanged.connect(self._update_estimate)
|
||
browse = QPushButton("…")
|
||
browse.setFixedWidth(32)
|
||
browse.clicked.connect(self.choose_output_dir)
|
||
drow = QHBoxLayout()
|
||
drow.addWidget(self.dir_edit, 1)
|
||
drow.addWidget(browse)
|
||
dw = QWidget()
|
||
dw.setLayout(drow)
|
||
form.addRow("输出目录", dw)
|
||
|
||
self.name_edit = QLineEdit("{datetime}_{device}")
|
||
self.name_edit.setToolTip("可用变量：{datetime} {date} {time} {device} "
|
||
"{sr} {bits} {ch} {seq}")
|
||
form.addRow("命名模板", self.name_edit)
|
||
|
||
self.split_time = self._spin(suffix=" 秒", rng=(0, 7200), value=0.0,
|
||
special="不分段", decimals=0, step=60,
|
||
tip="达到该时长自动切到新文件继续录")
|
||
form.addRow("按时长分段", self.split_time)
|
||
|
||
self.split_size = self._spin(suffix=" MB", rng=(0, 20480), value=0.0,
|
||
special="不限制", decimals=0, step=100,
|
||
tip="单个文件达到该体积自动分段")
|
||
form.addRow("按体积分段", self.split_size)
|
||
|
||
self.split_silence = QCheckBox("静音处自动分段")
|
||
self.split_silence.setToolTip("检测到持续静音时立即切到新文件，"
|
||
"适合长时间会议/采访自动拆分")
|
||
form.addRow(self.split_silence)
|
||
|
||
self.silence_thr = self._spin(suffix=" dBFS", rng=(-90, -20), value=-50.0,
|
||
decimals=1, step=1.0,
|
||
tip="低于该电平即视为静音")
|
||
form.addRow("静音阈值", self.silence_thr)
|
||
|
||
self.silence_gap = self._spin(suffix=" 秒", rng=(0.2, 60), value=2.0,
|
||
decimals=1, step=0.5,
|
||
tip="静音持续超过该时长才触发分段")
|
||
form.addRow("静音时长", self.silence_gap)
|
||
|
||
self.auto_stop = self._spin(suffix=" 秒", rng=(0, 600), value=0.0,
|
||
special="不自动停止", decimals=0, step=5,
|
||
tip="持续静音达到该时长后自动停止录音")
|
||
form.addRow("静音自动停止", self.auto_stop)
|
||
|
||
self.estimate_label = QLabel("—")
|
||
self.estimate_label.setStyleSheet(f"color:{TEXT_DIM}; font-size:11px;")
|
||
self.estimate_label.setWordWrap(True)
|
||
self.estimate_label.setMinimumWidth(0)
|
||
form.addRow(self.estimate_label)
|
||
|
||
self.auto_report_check = QCheckBox("录音后显示体检报告")
|
||
self.auto_report_check.setChecked(True)
|
||
form.addRow(self.auto_report_check)
|
||
|
||
calib_btn = QPushButton("校准本底噪声")
|
||
calib_btn.setToolTip("播放 3 秒静音测量房间/设备本底噪声，"
|
||
"并给出低切与噪声门建议")
|
||
calib_btn.clicked.connect(self.calibrate_noise)
|
||
form.addRow(calib_btn)
|
||
return box
|
||
|
||
def _build_right_panel(self) -> QWidget:
|
||
panel = QWidget()
|
||
v = QVBoxLayout(panel)
|
||
v.setContentsMargins(0, 0, 0, 0)
|
||
v.setSpacing(10)
|
||
|
||
# 传输控制条
|
||
transport = QFrame()
|
||
transport.setObjectName("card")
|
||
t = QHBoxLayout(transport)
|
||
t.setContentsMargins(14, 12, 14, 12)
|
||
t.setSpacing(14)
|
||
|
||
self.rec_button = RecordButton()
|
||
self.rec_button.clicked.connect(self.toggle_record)
|
||
t.addWidget(self.rec_button)
|
||
|
||
tcol = QVBoxLayout()
|
||
tcol.setSpacing(2)
|
||
self.time_label = QLabel("00:00.0")
|
||
tf = QFont("Consolas")
|
||
tf.setStyleHint(QFont.Monospace)
|
||
tf.setPointSize(26)
|
||
tf.setBold(True)
|
||
self.time_label.setFont(tf)
|
||
self.time_label.setMinimumWidth(280)
|
||
self.time_label.setStyleSheet(f"color:{TEXT};")
|
||
tcol.addWidget(self.time_label)
|
||
self.state_label = QLabel("待机")
|
||
self.state_label.setStyleSheet(f"color:{TEXT_DIM};")
|
||
tcol.addWidget(self.state_label)
|
||
t.addLayout(tcol)
|
||
|
||
t.addStretch(1)
|
||
|
||
self.pause_btn = QPushButton("暂停")
|
||
self.pause_btn.setToolTip("暂停 / 继续（快捷键 P）；暂停期间的数据不写入文件")
|
||
self.pause_btn.clicked.connect(self.toggle_pause)
|
||
self.marker_btn = QPushButton("标记")
|
||
self.marker_btn.setToolTip("在当前时间点打标记（快捷键 M），写入元数据")
|
||
self.marker_btn.clicked.connect(self.add_marker)
|
||
self.split_btn = QPushButton("分段")
|
||
self.split_btn.setToolTip("立即切到新文件继续录音（快捷键 Ctrl+S）")
|
||
self.split_btn.clicked.connect(self.split_now)
|
||
self.stop_btn = QPushButton("停止")
|
||
self.stop_btn.setToolTip("停止录音并收尾文件（快捷键 空格）")
|
||
self.stop_btn.clicked.connect(self.stop_record)
|
||
for b in (self.pause_btn, self.marker_btn, self.split_btn, self.stop_btn):
|
||
b.setMinimumHeight(34)
|
||
b.setMinimumWidth(72)
|
||
b.setFocusPolicy(Qt.NoFocus)
|
||
t.addWidget(b)
|
||
self.rec_button.setFocusPolicy(Qt.NoFocus)
|
||
v.addWidget(transport)
|
||
|
||
# 电平表
|
||
box = QGroupBox("电平表（RMS / 峰值保持 / 真峰值）")
|
||
bl = QVBoxLayout(box)
|
||
bl.setContentsMargins(10, 14, 10, 8)
|
||
self.meter = LevelMeter(channels=2)
|
||
bl.addWidget(self.meter)
|
||
v.addWidget(box)
|
||
|
||
# 中段：示波器 + 响度
|
||
box2 = QGroupBox("实时波形")
|
||
s2 = QVBoxLayout(box2)
|
||
s2.setContentsMargins(10, 14, 10, 8)
|
||
self.scope = Scope()
|
||
s2.addWidget(self.scope)
|
||
v.addWidget(box2, 2)
|
||
|
||
box3 = QGroupBox("响度（ITU-R BS.1770 / EBU R128）")
|
||
l3 = QVBoxLayout(box3)
|
||
l3.setContentsMargins(10, 14, 10, 8)
|
||
self.loudness = LoudnessBar()
|
||
self.loudness.set_target(-16.0)
|
||
l3.addWidget(self.loudness)
|
||
v.addWidget(box3)
|
||
|
||
# 总览
|
||
box4 = QGroupBox("整段录音总览")
|
||
o4 = QVBoxLayout(box4)
|
||
o4.setContentsMargins(10, 14, 10, 8)
|
||
self.history = HistoryPlot()
|
||
self.rail = MarkerRail()
|
||
o4.addWidget(self.history)
|
||
o4.addWidget(self.rail)
|
||
v.addWidget(box4, 1)
|
||
|
||
# 状态标签
|
||
pills = QGridLayout()
|
||
pills.setHorizontalSpacing(6)
|
||
self.pill_file = StatusPill("未开始", "dim")
|
||
self.pill_disk = StatusPill("磁盘 —", "dim")
|
||
self.pill_queue = StatusPill("缓冲队列 0", "dim")
|
||
self.pill_health = StatusPill("驱动 xrun 0", "dim")
|
||
pills.addWidget(self.pill_file, 0, 0)
|
||
pills.addWidget(self.pill_disk, 0, 1)
|
||
pills.addWidget(self.pill_queue, 0, 2)
|
||
pills.addWidget(self.pill_health, 0, 3)
|
||
v.addLayout(pills)
|
||
return panel
|
||
|
||
def _install_shortcuts(self) -> None:
|
||
QShortcut(QKeySequence(Qt.Key_Space), self, activated=self.toggle_record)
|
||
QShortcut(QKeySequence("P"), self, activated=self.toggle_pause)
|
||
QShortcut(QKeySequence("M"), self, activated=self.add_marker)
|
||
QShortcut(QKeySequence("Ctrl+S"), self, activated=self.split_now)
|
||
|
||
# ------------------------------------------------------------ 设备逻辑
|
||
def refresh_devices(self) -> None:
|
||
current = self.device_combo.currentData()
|
||
self._devices = list_input_devices()
|
||
self.device_combo.blockSignals(True)
|
||
self.device_combo.clear()
|
||
last_api = None
|
||
for d in self._devices:
|
||
if d.hostapi != last_api:
|
||
self.device_combo.addItem(f"── {d.hostapi} ──", None)
|
||
idx = self.device_combo.count() - 1
|
||
self.device_combo.model().item(idx).setEnabled(False)
|
||
last_api = d.hostapi
|
||
mark = "★ " if d.is_default else ""
|
||
self.device_combo.addItem(
|
||
f"{mark}{d.name} ({d.max_input_channels} ch)", d.index)
|
||
# 恢复原选择或选择默认设备
|
||
target = current if current is not None else engine.default_device_index()
|
||
for i in range(self.device_combo.count()):
|
||
if self.device_combo.itemData(i) == target:
|
||
self.device_combo.setCurrentIndex(i)
|
||
break
|
||
else:
|
||
for i in range(self.device_combo.count()):
|
||
if self.device_combo.itemData(i) is not None:
|
||
self.device_combo.setCurrentIndex(i)
|
||
break
|
||
self.device_combo.blockSignals(False)
|
||
self._on_device_changed()
|
||
self._device_signature = self._signature()
|
||
|
||
def _signature(self) -> str:
|
||
return "|".join(f"{d.index}:{d.name}:{d.hostapi}" for d in self._devices)
|
||
|
||
def _current_device(self) -> engine.DeviceInfo | None:
|
||
idx = self.device_combo.currentData()
|
||
for d in self._devices:
|
||
if d.index == idx:
|
||
return d
|
||
return None
|
||
|
||
def _on_device_changed(self) -> None:
|
||
d = self._current_device()
|
||
if d is None:
|
||
self.device_info.setText("未选择设备")
|
||
return
|
||
self.channels_spin.setMaximum(max(1, d.max_input_channels))
|
||
if self.channels_spin.value() > d.max_input_channels:
|
||
self.channels_spin.setValue(d.max_input_channels)
|
||
# 采样率对齐设备默认值
|
||
for i in range(self.rate_combo.count()):
|
||
if self.rate_combo.itemData(i) == int(d.default_samplerate):
|
||
self.rate_combo.setCurrentIndex(i)
|
||
break
|
||
hint = ""
|
||
rank = d.quality_rank
|
||
if rank >= 3:
|
||
hint = ("\n提示：MME/DirectSound 会经过 Windows 混音器，可能被重采样，"
|
||
"建议改用同一设备的 WASAPI 条目以获得更高保真度。")
|
||
elif rank == 0 and "wasapi" in d.hostapi.lower():
|
||
hint = "\n提示：WASAPI + 独占模式可完全绕开系统混音器，是最高保真的选择。"
|
||
self.device_info.setText(
|
||
f"宿主 API：{d.hostapi} 最大输入通道：{d.max_input_channels}\n"
|
||
f"驱动默认采样率：{d.default_samplerate:.0f} Hz "
|
||
f"报告延迟：低 {d.default_low_input_latency * 1000:.1f} ms / "
|
||
f"高 {d.default_high_input_latency * 1000:.1f} ms{hint}")
|
||
self._update_estimate()
|
||
|
||
def _check_devices(self) -> None:
|
||
devs = list_input_devices()
|
||
sig = "|".join(f"{d.index}:{d.name}:{d.hostapi}" for d in devs)
|
||
if sig != self._device_signature:
|
||
self._device_signature = sig
|
||
self._devices = devs
|
||
if self.recorder is None:
|
||
self.status_left.setText("检测到设备变化，已刷新设备列表")
|
||
self.refresh_devices()
|
||
|
||
def probe_device(self) -> None:
|
||
d = self._current_device()
|
||
if d is None:
|
||
QMessageBox.warning(self, "提示", "请先选择一个输入设备。")
|
||
return
|
||
self.probe_btn.setEnabled(False)
|
||
self.status_left.setText("正在检测设备支持能力…")
|
||
|
||
def job(progress=None):
|
||
return engine.probe_capabilities(d.index, channels=self.channels_spin.value())
|
||
|
||
self._cap_task = Task(job)
|
||
self._cap_task.done.connect(self._show_capabilities)
|
||
self._cap_task.failed.connect(lambda m: (
|
||
self.probe_btn.setEnabled(True),
|
||
QMessageBox.critical(self, "检测失败", m)))
|
||
self._cap_task.start()
|
||
|
||
def _show_capabilities(self, res: object) -> None:
|
||
self.probe_btn.setEnabled(True)
|
||
r = res if isinstance(res, dict) else {}
|
||
lines = [f"设备：{self.device_combo.currentText().strip()}",
|
||
f"宿主 API：{r.get('hostapi')} 检测通道数：{r.get('channels')}", ""]
|
||
for rate, entry in (r.get("rates") or {}).items():
|
||
if entry.get("supported"):
|
||
mode = "独占" if entry.get("exclusive") else "共享"
|
||
lines.append(f" ✓ {rate:>6} Hz {mode} 支持格式："
|
||
f"{', '.join(entry.get('dtypes', []))}")
|
||
else:
|
||
lines.append(f" ✗ {rate:>6} Hz 不支持（{entry.get('error', '')[:60]}）")
|
||
lines += ["", "结论：优先选择标注 ✓ 且为「独占」的采样率；",
|
||
"若想录 96 kHz 却没有 ✓，说明设备本身不支持该采样率。"]
|
||
QMessageBox.information(self, "设备能力检测", "\n".join(lines))
|
||
self.status_left.setText("设备能力检测完成")
|
||
|
||
def calibrate_noise(self) -> None:
|
||
d = self._current_device()
|
||
if d is None:
|
||
QMessageBox.warning(self, "提示", "请先选择一个输入设备。")
|
||
return
|
||
if self.recorder is not None:
|
||
QMessageBox.warning(self, "提示", "请先停止当前录音再做校准。")
|
||
return
|
||
QMessageBox.information(self, "校准本底噪声",
|
||
"接下来 3 秒请保持安静（不要说话、不要敲键盘）。\n"
|
||
"将测量房间/设备的本底噪声，并给出低切与噪声门建议。")
|
||
self.status_left.setText("正在测量本底噪声（保持安静）…")
|
||
|
||
rate = int(self.rate_combo.currentData())
|
||
ch = int(self.channels_spin.value())
|
||
dev = d.index
|
||
excl = self.exclusive_check.isChecked()
|
||
|
||
def job(progress=None):
|
||
import sounddevice as sd
|
||
extra = sd.WasapiSettings(exclusive=excl) \
|
||
if "wasapi" in d.hostapi.lower() else None
|
||
data = sd.rec(int(3.0 * rate), samplerate=rate, channels=ch,
|
||
dtype="float32", device=dev, extra_settings=extra,
|
||
blocking=True)
|
||
sd.wait()
|
||
return dsp.analyze_data(np.asarray(data, dtype=np.float64), rate)
|
||
|
||
self._calib_task = Task(job)
|
||
self._calib_task.done.connect(self._show_calibration)
|
||
self._calib_task.failed.connect(lambda m: (
|
||
self.status_left.setText("校准失败"), QMessageBox.critical(self, "校准失败", m)))
|
||
self._calib_task.start()
|
||
|
||
def _show_calibration(self, res: object) -> None:
|
||
an = res if isinstance(res, dict) else {}
|
||
floor = an.get("noise_floor_dbfs") or []
|
||
peak = an.get("peak_dbfs") or []
|
||
rms = an.get("rms_dbfs") or []
|
||
lines = ["本底噪声测量结果（3 秒）", ""]
|
||
lines.append(f" 采样峰值：{peak if peak else '—'} dBFS")
|
||
lines.append(f" RMS 电平：{rms if rms else '—'} dBFS")
|
||
lines.append(f" 本底噪声：{floor if floor else '—（3 秒内没有检测到安静段）'}")
|
||
lines.append(f" 直流偏移：{an.get('dc_offset')}")
|
||
lines.append("")
|
||
suggest_gate = None
|
||
if floor:
|
||
f0 = float(floor[0])
|
||
suggest_gate = round(min(-30.0, f0 + 6.0), 1)
|
||
lines.append(f"建议噪声门阈值：{suggest_gate} dBFS（本底 +6 dB）")
|
||
lines.append("建议低切：若本底噪声在 100 Hz 以下明显更高，选择 80~120 Hz 低切")
|
||
lines.append("建议增益：让正常说话时峰值落在 -12 ~ -6 dBFS 之间最理想")
|
||
if peak and peak[0] is not None and peak[0] < -30:
|
||
lines.append(f" 当前峰值偏低（{peak[0]:.1f} dBFS），"
|
||
"可在「软件增益」里提升 10~20 dB")
|
||
if peak and peak[0] is not None and peak[0] > -3:
|
||
lines.append(f" 当前峰值偏高（{peak[0]:.1f} dBFS），建议降低系统麦克风音量")
|
||
box = QMessageBox(self)
|
||
box.setWindowTitle("本底噪声校准")
|
||
box.setText("\n".join(lines))
|
||
apply_btn = box.addButton("应用建议", QMessageBox.AcceptRole)
|
||
box.addButton("关闭", QMessageBox.RejectRole)
|
||
box.exec_()
|
||
if box.clickedButton() is apply_btn and suggest_gate is not None:
|
||
self.silence_thr.setValue(suggest_gate)
|
||
self.status_left.setText(f"已把静音阈值设为 {suggest_gate} dBFS")
|
||
|
||
# ------------------------------------------------------------ 参数收集
|
||
def _config_from_ui(self) -> RecordConfig:
|
||
lowcut = self.lowcut_combo.currentData() or 0.0
|
||
return RecordConfig(
|
||
device=self.device_combo.currentData(),
|
||
samplerate=int(self.rate_combo.currentData()),
|
||
channels=int(self.channels_spin.value()),
|
||
bit_depth=self.depth_combo.currentData(),
|
||
gain_db=self.gain_slider.value() / 10.0,
|
||
exclusive=self.exclusive_check.isChecked(),
|
||
blocksize=int(self.blocksize_combo.currentData() or 0),
|
||
latency=self.latency_combo.currentData(),
|
||
lowcut_hz=float(lowcut),
|
||
dither=self.dither_check.isChecked(),
|
||
rf64=self.rf64_check.isChecked(),
|
||
output_dir=self.dir_edit.text().strip(),
|
||
name_template=self.name_edit.text().strip() or "{datetime}_{device}",
|
||
split_seconds=float(self.split_time.value()),
|
||
split_megabytes=float(self.split_size.value()),
|
||
split_on_silence=self.split_silence.isChecked(),
|
||
silence_threshold_dbfs=float(self.silence_thr.value()),
|
||
silence_gap_seconds=float(self.silence_gap.value()),
|
||
auto_stop_silence_seconds=float(self.auto_stop.value()),
|
||
)
|
||
|
||
def _on_gain_changed(self, value: int) -> None:
|
||
db = value / 10.0
|
||
self.gain_label.setText(f"{db:+.1f} dB")
|
||
if self.recorder is not None and self.recorder.state == RecorderState.RECORDING:
|
||
self.recorder.config.gain_db = db # 软件增益支持录音中实时调整
|
||
|
||
def _on_target_changed(self) -> None:
|
||
target = self.target_combo.currentData()
|
||
if target is not None and hasattr(self, "loudness"):
|
||
self.loudness.set_target(float(target))
|
||
|
||
def _update_estimate(self) -> None:
|
||
try:
|
||
cfg = self._config_from_ui()
|
||
except Exception:
|
||
return
|
||
bps = cfg.estimate_bytes_per_hour()
|
||
free = free_space_bytes(cfg.output_dir or os.getcwd())
|
||
bits = cfg.bit_depth
|
||
note = ""
|
||
if bits == "24":
|
||
note = "24 位/48 kHz 是录音的甜点：动态范围 144 dB，兼容性极好。"
|
||
elif bits == "16":
|
||
note = "16 位适合直接发布，但后期处理余量小，建议录 24 位再降。"
|
||
elif bits == "float32":
|
||
note = "32 位浮点不会因超 0 dBFS 而削波，适合需要大量后期的素材。"
|
||
else:
|
||
note = "32 位整数相对少见，一般选 24 位或 32 位浮点更实用。"
|
||
self.estimate_label.setText(
|
||
f"约 {format_bytes(bps)} / 小时 · 剩余空间 {format_bytes(free) if free > 0 else '—'}"
|
||
f" · 可录约 {(free / bps):.1f} 小时\n{note}"
|
||
if free > 0 else f"约 {format_bytes(bps)} / 小时\n{note}")
|
||
|
||
# ------------------------------------------------------------ 录音控制
|
||
def toggle_record(self) -> None:
|
||
if self.recorder is None:
|
||
self.start_record()
|
||
else:
|
||
self.stop_record()
|
||
|
||
def start_record(self) -> None:
|
||
try:
|
||
cfg = self._config_from_ui()
|
||
except Exception as exc:
|
||
QMessageBox.critical(self, "参数错误", str(exc))
|
||
return
|
||
if not cfg.output_dir:
|
||
cfg.output_dir = os.path.join(os.path.expanduser("~"), "Recordings")
|
||
try:
|
||
os.makedirs(cfg.output_dir, exist_ok=True)
|
||
except Exception as exc:
|
||
QMessageBox.critical(self, "无法创建输出目录", str(exc))
|
||
return
|
||
|
||
self.recorder = Recorder(cfg)
|
||
try:
|
||
self.recorder.start()
|
||
except Exception as exc:
|
||
self.recorder = None
|
||
QMessageBox.critical(self, "无法开始录音", str(exc))
|
||
self._set_idle_ui()
|
||
return
|
||
|
||
self._peak_alerted = False
|
||
self._history_ready = False
|
||
self.meter.set_channels(cfg.channels)
|
||
self.scope.set_format(cfg.samplerate, cfg.channels)
|
||
self.meter.reset()
|
||
self.rail.set_marks([], 1.0)
|
||
self._set_recording_ui(True)
|
||
self.status_left.setText(f"正在录音：{cfg.output_dir}")
|
||
notes = self.recorder.notes()
|
||
if notes:
|
||
self.status_left.setText(notes[-1])
|
||
|
||
def stop_record(self) -> None:
|
||
if self.recorder is None:
|
||
return
|
||
self.status_left.setText("正在收尾文件…")
|
||
QApplication.processEvents()
|
||
try:
|
||
result = self.recorder.stop()
|
||
except Exception as exc:
|
||
QMessageBox.critical(self, "停止录音出错", str(exc))
|
||
result = None
|
||
self.recorder = None
|
||
self._set_idle_ui()
|
||
if result is None or not result.files:
|
||
self.status_left.setText("录音已停止（没有产生文件）")
|
||
return
|
||
self.result = result
|
||
self._last_files = list(result.files)
|
||
self.status_left.setText(
|
||
f"已保存：{os.path.basename(result.primary_file)} "
|
||
f"{format_duration(result.duration)} {format_bytes(result.bytes_written)}")
|
||
self.rail.set_marks([(m.seconds, m.label) for m in result.markers],
|
||
max(1e-6, result.duration))
|
||
if self.auto_report_check.isChecked():
|
||
ReportDialog(result, self).exec_()
|
||
|
||
def toggle_pause(self) -> None:
|
||
if self.recorder is None:
|
||
return
|
||
self.recorder.toggle_pause()
|
||
paused = self.recorder.state == RecorderState.PAUSED
|
||
self.rec_button.set_state("paused" if paused else "recording")
|
||
self.pause_btn.setText("▶ 继续" if paused else "⏸ 暂停")
|
||
self.state_label.setText("已暂停（此段不写入文件）" if paused else "录音中")
|
||
|
||
def add_marker(self) -> None:
|
||
if self.recorder is None:
|
||
return
|
||
m = self.recorder.add_marker()
|
||
if m:
|
||
self.status_left.setText(f"已添加标记：{m.seconds:.2f} 秒")
|
||
|
||
def split_now(self) -> None:
|
||
if self.recorder is None:
|
||
return
|
||
self.recorder.split_now()
|
||
self.status_left.setText("已请求分段，正在切换新文件…")
|
||
|
||
def choose_output_dir(self) -> None:
|
||
start = self.dir_edit.text().strip() or os.path.expanduser("~")
|
||
path = QFileDialog.getExistingDirectory(self, "选择录音保存目录", start)
|
||
if path:
|
||
self.dir_edit.setText(path)
|
||
self._save_settings()
|
||
|
||
def open_output_dir(self) -> None:
|
||
path = self.dir_edit.text().strip()
|
||
if path and os.path.isdir(path):
|
||
QDesktopServices.openUrl(QUrl.fromLocalFile(path))
|
||
else:
|
||
QMessageBox.information(self, "提示", "输出目录还不存在。")
|
||
|
||
def open_process_last(self) -> None:
|
||
src = self.result.primary_file if self.result else (
|
||
self._last_files[0] if self._last_files else "")
|
||
if not src or not os.path.exists(src):
|
||
path, _ = QFileDialog.getOpenFileName(self, "选择要处理的音频", "",
|
||
"WAV 文件 (*.wav)")
|
||
if not path:
|
||
return
|
||
src = path
|
||
ProcessDialog(src, self).exec_()
|
||
|
||
# ------------------------------------------------------------ 界面状态
|
||
def _set_recording_ui(self, recording: bool) -> None:
|
||
self.rec_button.set_state("recording" if recording else "idle")
|
||
self.pause_btn.setText("⏸ 暂停")
|
||
for w in (self.device_combo, self.refresh_btn, self.rate_combo,
|
||
self.depth_combo, self.channels_spin, self.exclusive_check,
|
||
self.blocksize_combo, self.latency_combo, self.lowcut_combo,
|
||
self.rf64_check, self.dir_edit, self.name_edit,
|
||
self.split_time, self.split_size, self.split_silence,
|
||
self.silence_thr, self.silence_gap, self.auto_stop):
|
||
w.setEnabled(not recording)
|
||
for w in (self.pause_btn, self.marker_btn, self.split_btn, self.stop_btn):
|
||
w.setEnabled(recording)
|
||
self.state_label.setText("录音中…" if recording else "待机")
|
||
|
||
def _set_idle_ui(self) -> None:
|
||
self._set_recording_ui(False)
|
||
self.rec_button.set_state("idle")
|
||
self.pause_btn.setEnabled(False)
|
||
self.marker_btn.setEnabled(False)
|
||
self.split_btn.setEnabled(False)
|
||
self.stop_btn.setEnabled(False)
|
||
|
||
def _tick(self) -> None:
|
||
self._pulse = (self._pulse + 0.033) % 1.0
|
||
self.rec_button.set_pulse(self._pulse)
|
||
if self.recorder is None:
|
||
return
|
||
snap = self.recorder.live()
|
||
# 引擎可能自己停了（静音自动停止 / 写入失败）：此时要替用户走完收尾流程，
|
||
# 否则界面会一直停在"录音中"而文件其实已经收尾。
|
||
if snap.state == RecorderState.IDLE.value \
|
||
and self.recorder.state != RecorderState.IDLE:
|
||
self.status_left.setText("录音已由引擎自动结束，正在收尾…")
|
||
self.stop_record()
|
||
return
|
||
self.meter.update_levels(snap.meter)
|
||
self.scope.update_data(self.recorder.scope_data(),
|
||
self.recorder.state == RecorderState.RECORDING)
|
||
self.loudness.update_levels(snap.meter.momentary_lufs,
|
||
snap.meter.short_term_lufs,
|
||
snap.meter.integrated_lufs)
|
||
self.time_label.setText(_clock(snap.elapsed))
|
||
|
||
# 总览图是整段聚合，刷新率降到约 3 Hz，长录音也不会拖累界面
|
||
self._tick_count += 1
|
||
if self._tick_count % 8 == 0 or not self._history_ready:
|
||
mins, maxs, bucket = self.recorder.envelope()
|
||
self.history.update_envelope(mins, maxs, bucket,
|
||
self.recorder.config.samplerate,
|
||
snap.elapsed)
|
||
self._history_ready = True
|
||
marks = [(m.seconds, m.label) for m in self.recorder.current_markers()]
|
||
self.rail.set_marks(marks, max(1e-6, snap.elapsed))
|
||
|
||
if snap.current_file:
|
||
self.pill_file.set(f"{os.path.basename(snap.current_file)} "
|
||
f"{format_bytes(snap.bytes_written)}", "info")
|
||
free = snap.free_space
|
||
if free > 0:
|
||
lvl = "ok" if free > 2 * 1024 ** 3 else ("warn" if free > 512 * 1024 ** 2
|
||
else "err")
|
||
self.pill_disk.set(f"剩余空间 {format_bytes(free)}", lvl)
|
||
self.pill_queue.set(
|
||
f"缓冲队列 {snap.queue_backlog}（{snap.queue_backlog_ms:.0f} ms）"
|
||
f" 写盘 {snap.disk_write_mbps:.1f} MB/s",
|
||
"ok" if snap.queue_backlog < 8 else ("warn" if snap.queue_backlog < 32
|
||
else "err"))
|
||
health_lvl = "ok" if (snap.xruns == 0 and snap.overflow_blocks == 0) else "warn"
|
||
self.pill_health.set(
|
||
f"驱动 xrun {snap.xruns} 溢出块 {snap.overflow_blocks}", health_lvl)
|
||
if snap.meter.clipped and not self._peak_alerted:
|
||
self._peak_alerted = True
|
||
self.status_left.setText("⚠ 检测到削波：请降低系统麦克风音量或软件增益")
|
||
self.status_right.setText(
|
||
f"{self.recorder.config.wav_format().describe()} "
|
||
f"{'独占' if self.recorder.config.exclusive else '共享'} "
|
||
f"标记 {snap.markers}")
|
||
|
||
# ------------------------------------------------------------ 设置持久化
|
||
def _settings_path(self) -> str:
|
||
return os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
|
||
"recordings")
|
||
|
||
def _load_settings(self) -> None:
|
||
s = self._settings
|
||
default_dir = s.value("output_dir", self._settings_path(), type=str)
|
||
self.dir_edit.setText(str(default_dir))
|
||
self.name_edit.setText(s.value("name_template", "{datetime}_{device}", type=str))
|
||
idx = self.depth_combo.findData(s.value("bit_depth", "24", type=str))
|
||
if idx >= 0:
|
||
self.depth_combo.setCurrentIndex(idx)
|
||
idx = self.rate_combo.findData(s.value("samplerate", 48000, type=int))
|
||
if idx >= 0:
|
||
self.rate_combo.setCurrentIndex(idx)
|
||
self.channels_spin.setValue(s.value("channels", 2, type=int))
|
||
self.exclusive_check.setChecked(s.value("exclusive", True, type=bool))
|
||
self.dither_check.setChecked(s.value("dither", True, type=bool))
|
||
idx = self.lowcut_combo.findData(s.value("lowcut", 0.0, type=float))
|
||
if idx >= 0:
|
||
self.lowcut_combo.setCurrentIndex(idx)
|
||
self.gain_slider.setValue(int(s.value("gain_db_x10", 0, type=int)))
|
||
idx = self.latency_combo.findData(s.value("latency", "high", type=str))
|
||
if idx >= 0:
|
||
self.latency_combo.setCurrentIndex(idx)
|
||
self.window_state = s.value("geometry")
|
||
|
||
def _save_settings(self) -> None:
|
||
s = self._settings
|
||
s.setValue("output_dir", self.dir_edit.text().strip())
|
||
s.setValue("name_template", self.name_edit.text().strip())
|
||
s.setValue("bit_depth", self.depth_combo.currentData())
|
||
s.setValue("samplerate", int(self.rate_combo.currentData()))
|
||
s.setValue("channels", int(self.channels_spin.value()))
|
||
s.setValue("exclusive", self.exclusive_check.isChecked())
|
||
s.setValue("dither", self.dither_check.isChecked())
|
||
s.setValue("lowcut", float(self.lowcut_combo.currentData() or 0.0))
|
||
s.setValue("gain_db_x10", int(self.gain_slider.value()))
|
||
s.setValue("latency", self.latency_combo.currentData())
|
||
s.sync()
|
||
|
||
def closeEvent(self, event) -> None: # noqa: N802
|
||
if self.recorder is not None:
|
||
answer = QMessageBox.question(
|
||
self, "正在录音", "录音尚未停止，是否停止并保存后退出？",
|
||
QMessageBox.Yes | QMessageBox.No, QMessageBox.Yes)
|
||
if answer == QMessageBox.No:
|
||
event.ignore()
|
||
return
|
||
self.recorder.shutdown()
|
||
self._save_settings()
|
||
event.accept()
|
||
|
||
# ------------------------------------------------------------ 帮助信息
|
||
def show_shortcuts(self) -> None:
|
||
QMessageBox.information(self, "快捷键", "\n".join([
|
||
"空格 开始 / 停止录音",
|
||
"P 暂停 / 继续（暂停段不写入文件）",
|
||
"M 添加标记（写入元数据）",
|
||
"Ctrl+S 立即分段（切到新文件继续录）",
|
||
"Ctrl+O 打开输出目录",
|
||
"Ctrl+Q 退出",
|
||
]))
|
||
|
||
def show_about(self) -> None:
|
||
dlg = QDialog(self)
|
||
dlg.setWindowTitle("关于 / 高清晰度录音指南")
|
||
dlg.resize(680, 620)
|
||
lay = QVBoxLayout(dlg)
|
||
text = QPlainTextEdit()
|
||
text.setReadOnly(True)
|
||
text.setPlainText(HIGH_QUALITY_GUIDE)
|
||
lay.addWidget(text)
|
||
btns = QDialogButtonBox(QDialogButtonBox.Ok)
|
||
btns.accepted.connect(dlg.accept)
|
||
lay.addWidget(btns)
|
||
dlg.exec_()
|
||
|
||
|
||
HIGH_QUALITY_GUIDE = """RecorderStudio 高清晰度录音指南
|
||
=====================================================
|
||
|
||
一、这条链路为什么"清晰"
|
||
-----------------------------------------------------
|
||
1. 采集端默认走 WASAPI 独占模式：绕开 Windows 混音器，
|
||
不会被系统重采样、不会被其它程序的音量条动过手脚。
|
||
2. 内部全程 32 位浮点运算，写盘再按你选的位深量化；
|
||
16/24 位写入时可选 TPDF 抖动，把量化失真从"与信号相关的
|
||
非线性失真"变成无关的宽带白噪，弱信号细节听得更干净。
|
||
3. 低切滤波使用 5000+ 抽头的线性相位 FIR：不引入相位失真，
|
||
并对输出做群延迟补偿，不会丢头掉尾。
|
||
4. 音频回调线程只做内存拷贝，滤波/写盘/FFT 都在写入线程，
|
||
从根上避免 xrun（爆音、掉采样）。
|
||
5. 计量按国际标准：真峰值 4 倍过采样（ITU-R BS.1770 附录 2），
|
||
响度按 K 加权 + 门限算法（EBU R128）。本程序的自检里
|
||
已用 ffmpeg 的 ebur128 交叉验证，误差 < 0.1 LU。
|
||
|
||
二、参数怎么选
|
||
-----------------------------------------------------
|
||
· 采样率：48000 Hz。它同时满足语音、音乐与视频后期；
|
||
设备原生就是 48 kHz 时选它可完全避免重采样。
|
||
· 位深：24 位。动态范围 144 dB，远超人耳与话筒本底；
|
||
需要发给别人听就录 24 位、导出 16 位（带抖动）。
|
||
· 声道：话筒用 1 声道；立体声乐器/环境音用 2 声道。
|
||
· 缓冲区：优先"自动/4096 + 高延迟策略"，录音不需要低延迟，
|
||
稳定不出 xrun 才是第一优先级。
|
||
· 软件增益：只在系统麦克风音量已经拉满还不够时使用；
|
||
优先调节系统输入音量，因为它作用在更靠前的增益级。
|
||
|
||
三、电平与房间
|
||
-----------------------------------------------------
|
||
· 目标：正常说话时峰值落在 -12 ~ -6 dBFS，绝不碰到 0 dBFS。
|
||
数字削波是不可逆的，且比模拟过载难听得多。
|
||
· 房间噪声比设备更影响"清晰度"：关空调、关风扇、
|
||
在桌上垫软布、话筒离嘴 15~25 cm 并稍偏轴（避开喷气）。
|
||
· 先用「校准本底噪声」测一下：如果本底高于 -55 dBFS，
|
||
先改善环境，再考虑噪声门。
|
||
|
||
四、录制电脑内部声音
|
||
-----------------------------------------------------
|
||
在设备列表里选择「立体声混音 / Stereo Mix」或
|
||
「主声音捕获驱动程序」（通常在 WASAPI/WDM-KS 分组下）。
|
||
若列表中没有，需要在 Windows「声音设置 → 录制 →
|
||
显示禁用的设备」里启用"立体声混音"。
|
||
|
||
五、录完之后
|
||
-----------------------------------------------------
|
||
· 每次录音都会生成体检报告：采样峰值、真峰值、RMS、
|
||
整体响度、动态范围、直流偏移、本底噪声、削波统计。
|
||
· 「后期处理」可做：DC 去除、低切、噪声门、裁剪首尾静音、
|
||
峰值/响度归一化、淡入淡出、混单声道 —— 全部另存为新文件，
|
||
原始母版永不被覆盖。
|
||
· 「导出」支持 FLAC（无损压缩）、MP3、Opus、AAC；
|
||
需要 ffmpeg，程序会自动检测。
|
||
|
||
六、常见问题
|
||
-----------------------------------------------------
|
||
· 打不开设备：多半被其它程序独占，关掉浏览器/会议软件再试；
|
||
或关闭"独占模式"改用共享模式。
|
||
· 提示采样率不支持：看「检测能力」的结果，选一个带 ✓ 的采样率。
|
||
· 录出来是静音：检查 Windows 隐私设置是否允许桌面应用访问麦克风，
|
||
以及系统输入设备是否选对。
|
||
"""
|