470 lines
19 KiB
Python
470 lines
19 KiB
Python
"""后期处理与导出：归一化、裁剪、门限、淡入淡出、转码、元数据、波形预览。
|
||
|
||
设计取向：**原始 WAV 母版永远保留且不被覆盖**。所有后期处理默认写出新文件
|
||
（``*_processed.wav``），源文件只读。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import os
|
||
import shutil
|
||
import struct
|
||
import subprocess
|
||
import threading
|
||
import zlib
|
||
from dataclasses import asdict, dataclass
|
||
from datetime import datetime
|
||
|
||
import numpy as np
|
||
|
||
from . import dsp
|
||
from .engine import TakeResult, format_bytes, format_duration
|
||
from .wavfile import WavReader, write_wav
|
||
|
||
# ---------------------------------------------------------------- ffmpeg
|
||
_FFMPEG_CACHE: list[str | None] = []
|
||
_FFMPEG_LOCK = threading.Lock()
|
||
|
||
|
||
def find_ffmpeg() -> str | None:
|
||
"""定位 ffmpeg（PATH 或常见安装位置）。"""
|
||
with _FFMPEG_LOCK:
|
||
if _FFMPEG_CACHE:
|
||
return _FFMPEG_CACHE[0]
|
||
candidates: list[str | None] = []
|
||
exe = shutil.which("ffmpeg")
|
||
if exe:
|
||
candidates.append(exe)
|
||
for p in (
|
||
r"C:\ffmpeg\bin\ffmpeg.exe",
|
||
r"C:\Program Files\ffmpeg\bin\ffmpeg.exe",
|
||
os.path.expanduser(r"~\scoop\shims\ffmpeg.exe"),
|
||
os.path.expanduser(r"~\AppData\Local\Microsoft\WinGet\Links\ffmpeg.exe"),
|
||
):
|
||
if os.path.exists(p):
|
||
candidates.append(p)
|
||
_FFMPEG_CACHE.append(candidates[0] if candidates else None)
|
||
return _FFMPEG_CACHE[0]
|
||
|
||
|
||
EXPORT_PRESETS: dict[str, dict] = {
|
||
"wav_16": {"label": "WAV 16-bit（兼容性最好）", "ext": ".wav", "kind": "wav"},
|
||
"wav_24": {"label": "WAV 24-bit（无损母版）", "ext": ".wav", "kind": "wav"},
|
||
"wav_f32": {"label": "WAV 32-bit float（后期制作）", "ext": ".wav", "kind": "wav"},
|
||
"flac": {"label": "FLAC（无损压缩，约 50% 体积）", "ext": ".flac", "kind": "ffmpeg"},
|
||
"mp3_320": {"label": "MP3 320 kbps（高码率有损）", "ext": ".mp3", "kind": "ffmpeg"},
|
||
"mp3_v0": {"label": "MP3 V0（VBR 约 245 kbps）", "ext": ".mp3", "kind": "ffmpeg"},
|
||
"opus": {"label": "Opus 128 kbps（语音/播客首选）", "ext": ".opus", "kind": "ffmpeg"},
|
||
"m4a": {"label": "AAC/M4A 256 kbps（苹果生态）", "ext": ".m4a", "kind": "ffmpeg"},
|
||
}
|
||
|
||
|
||
def _ffmpeg_args(preset: str, src: str, dst: str) -> list[str]:
|
||
ff = find_ffmpeg() or "ffmpeg"
|
||
base = [ff, "-hide_banner", "-loglevel", "error", "-y", "-i", src]
|
||
if preset == "flac":
|
||
return base + ["-c:a", "flac", "-compression_level", "8", dst]
|
||
if preset == "mp3_320":
|
||
return base + ["-c:a", "libmp3lame", "-b:a", "320k", dst]
|
||
if preset == "mp3_v0":
|
||
return base + ["-c:a", "libmp3lame", "-q:a", "0", dst]
|
||
if preset == "opus":
|
||
return base + ["-c:a", "libopus", "-b:a", "128k", dst]
|
||
if preset == "m4a":
|
||
return base + ["-c:a", "aac", "-b:a", "256k", dst]
|
||
return base + [dst]
|
||
|
||
|
||
def default_export_path(src: str, preset: str) -> str:
|
||
"""给出导出的默认目标路径；**绝不允许覆盖源文件**。"""
|
||
meta = EXPORT_PRESETS.get(preset) or {"ext": ".wav"}
|
||
root, _ext = os.path.splitext(src)
|
||
dst = f"{root}{meta['ext']}"
|
||
if os.path.abspath(dst) == os.path.abspath(src):
|
||
dst = f"{root}_{preset}{meta['ext']}"
|
||
return dst
|
||
|
||
|
||
def export_audio(src: str, dst: str, preset: str) -> tuple[bool, str]:
|
||
"""把 WAV 转成目标格式。返回 ``(是否成功, 说明)``。
|
||
|
||
安全约束：如果 ``dst`` 指向源文件本身，会自动改名，避免把母版覆盖掉。
|
||
"""
|
||
info = EXPORT_PRESETS.get(preset)
|
||
if info is None:
|
||
return False, f"未知的导出预设：{preset}"
|
||
if os.path.abspath(dst) == os.path.abspath(src):
|
||
dst = default_export_path(src, preset)
|
||
if info["kind"] == "wav":
|
||
try:
|
||
data, sr = _read_whole(src)
|
||
bits = {"wav_16": "16", "wav_24": "24", "wav_f32": "float32"}[preset]
|
||
write_wav(dst, data, sr, bit_depth=bits,
|
||
dither=(bits in ("16", "24")))
|
||
return True, f"{os.path.basename(dst)}（{format_bytes(os.path.getsize(dst))}）"
|
||
except Exception as exc:
|
||
return False, f"写入失败：{exc}"
|
||
ff = find_ffmpeg()
|
||
if ff is None:
|
||
return False, "未找到 ffmpeg：请安装 ffmpeg 并加入 PATH，或改用 WAV 导出"
|
||
try:
|
||
proc = subprocess.run(_ffmpeg_args(preset, src, dst),
|
||
capture_output=True, text=True, timeout=1800)
|
||
except Exception as exc:
|
||
return False, f"调用 ffmpeg 失败：{exc}"
|
||
if proc.returncode != 0:
|
||
return False, f"ffmpeg 出错：{(proc.stderr or '').strip()[:300]}"
|
||
return True, f"{os.path.basename(dst)}（{format_bytes(os.path.getsize(dst))}）"
|
||
|
||
|
||
# ------------------------------------------------------------ 处理选项
|
||
@dataclass
|
||
class ProcessOptions:
|
||
trim_silence: bool = False
|
||
trim_threshold_dbfs: float = -50.0
|
||
trim_min_silence: float = 0.4
|
||
remove_dc: bool = False
|
||
lowcut_hz: float = 0.0
|
||
noise_gate: bool = False
|
||
gate_threshold_dbfs: float = -60.0
|
||
normalize: str = "none" # 'none' | 'peak' | 'lufs'
|
||
normalize_target_dbfs: float = -1.0
|
||
normalize_target_lufs: float = -16.0
|
||
fade_in: float = 0.0
|
||
fade_out: float = 0.0
|
||
mono: bool = False
|
||
bit_depth: str = "24"
|
||
dither: bool = True
|
||
|
||
@property
|
||
def is_identity(self) -> bool:
|
||
return (not self.trim_silence and not self.remove_dc and self.lowcut_hz <= 0
|
||
and not self.noise_gate and self.normalize == "none"
|
||
and self.fade_in <= 0 and self.fade_out <= 0 and not self.mono)
|
||
|
||
|
||
def process_array(data: np.ndarray, samplerate: int, opts: ProcessOptions
|
||
) -> tuple[np.ndarray, dict]:
|
||
"""在内存中执行后期处理链，返回 ``(处理后的数据, 处理报告)``。"""
|
||
x = np.asarray(data, dtype=np.float64)
|
||
report: dict = {"steps": []}
|
||
|
||
if opts.remove_dc:
|
||
x = dsp.remove_dc(x, samplerate)
|
||
report["steps"].append("去除直流偏移（1 秒滑动平均，兼顾漂移）")
|
||
|
||
if opts.lowcut_hz > 0:
|
||
x = dsp.highpass_offline(x, samplerate, opts.lowcut_hz)
|
||
report["steps"].append(f"{opts.lowcut_hz:.0f} Hz 线性相位低切")
|
||
|
||
if opts.noise_gate:
|
||
x, gate_info = dsp.noise_gate(x, samplerate,
|
||
threshold_dbfs=opts.gate_threshold_dbfs)
|
||
report["noise_gate"] = gate_info
|
||
report["steps"].append(f"噪声门（{opts.gate_threshold_dbfs:.0f} dBFS）")
|
||
|
||
if opts.trim_silence:
|
||
x, trim_info = dsp.trim_silence(x, samplerate,
|
||
threshold_dbfs=opts.trim_threshold_dbfs,
|
||
min_silence=opts.trim_min_silence)
|
||
report["trim"] = trim_info
|
||
report["steps"].append(f"裁剪首尾静音（{trim_info.get('trimmed_seconds', 0)} s）")
|
||
|
||
if opts.mono and x.ndim > 1 and x.shape[1] > 1:
|
||
x = dsp.mixdown_mono(x)
|
||
report["steps"].append("混合为单声道")
|
||
|
||
if opts.normalize == "peak":
|
||
x, gain = dsp.normalize_peak(x, opts.normalize_target_dbfs)
|
||
report["normalize"] = {"mode": "peak", "gain_db": round(gain, 3),
|
||
"target_dbfs": opts.normalize_target_dbfs}
|
||
report["steps"].append(f"峰值归一化到 {opts.normalize_target_dbfs} dBFS")
|
||
elif opts.normalize == "lufs":
|
||
x, lufs_info = dsp.normalize_loudness(x, samplerate,
|
||
opts.normalize_target_lufs)
|
||
report["normalize"] = {"mode": "lufs", **lufs_info}
|
||
report["steps"].append(f"响度归一化到 {opts.normalize_target_lufs} LUFS")
|
||
|
||
if opts.fade_in > 0 or opts.fade_out > 0:
|
||
x = dsp.fade_edges(x, samplerate, fade_in=opts.fade_in, fade_out=opts.fade_out)
|
||
report["steps"].append(f"淡入 {opts.fade_in}s / 淡出 {opts.fade_out}s")
|
||
|
||
report["output_peak_dbfs"] = [round(float(v), 3) for v in
|
||
np.atleast_1d(dsp.dbfs(dsp.peak(x, axis=0)))]
|
||
return x, report
|
||
|
||
|
||
def _read_whole(path: str) -> tuple[np.ndarray, int]:
|
||
"""读取整个 WAV；超大文件自动降级为 float32 以节省内存。"""
|
||
with WavReader(path) as r:
|
||
frames = r.frames
|
||
need = frames * r.channels * 8
|
||
dtype = np.float64 if need < (1 << 31) else np.float32
|
||
chunks = []
|
||
for blk in r.iter_blocks(1 << 20):
|
||
chunks.append(blk.astype(dtype))
|
||
sr = r.samplerate
|
||
if not chunks:
|
||
return np.zeros((0, 1), dtype=dtype), sr
|
||
return np.concatenate(chunks, axis=0), sr
|
||
|
||
|
||
def process_file(src: str, opts: ProcessOptions, *,
|
||
dst: str | None = None,
|
||
progress=None) -> dict:
|
||
"""对录音文件执行后期处理并写出新文件（源文件保持不变）。"""
|
||
if dst is None:
|
||
root, ext = os.path.splitext(src)
|
||
dst = f"{root}_processed{ext or '.wav'}"
|
||
if progress:
|
||
progress("读取音频…")
|
||
data, sr = _read_whole(src)
|
||
if progress:
|
||
progress(f"处理 {len(data) / max(1, sr):.1f} 秒音频…")
|
||
out, report = process_array(data, sr, opts)
|
||
if progress:
|
||
progress("写出文件…")
|
||
stats = write_wav(dst, out.astype(np.float32), sr,
|
||
bit_depth=opts.bit_depth, dither=opts.dither)
|
||
try:
|
||
report["analysis"] = dsp.analyze_file(dst)
|
||
except Exception:
|
||
report["analysis"] = None
|
||
report.update({"source": src, "output": dst, "format": stats.get("format", "")})
|
||
return report
|
||
|
||
|
||
# --------------------------------------------------------------- 元数据
|
||
def write_metadata(result: TakeResult, *, extra: dict | None = None) -> list[str]:
|
||
"""写出 JSON 元数据 + 人类可读文本日志，返回生成的文件列表。"""
|
||
written: list[str] = []
|
||
cfg = result.config
|
||
meta = {
|
||
"app": "RecorderStudio",
|
||
"version": "1.0",
|
||
"recorded_at": result.started_at,
|
||
"finished_at": result.ended_at,
|
||
"device": result.device_label,
|
||
"format": result.format_label,
|
||
"sample_rate": cfg.samplerate if cfg else None,
|
||
"bit_depth": cfg.bit_depth if cfg else None,
|
||
"channels": cfg.channels if cfg else None,
|
||
"gain_db": cfg.gain_db if cfg else 0.0,
|
||
"lowcut_hz": cfg.lowcut_hz if cfg else 0.0,
|
||
"exclusive_mode": cfg.exclusive if cfg else None,
|
||
"dither": cfg.dither if cfg else None,
|
||
"duration_seconds": round(result.duration, 3),
|
||
"frames": result.frames,
|
||
"bytes": result.bytes_written,
|
||
"peak_dbfs": round(result.peak_dbfs, 3)
|
||
if np.isfinite(result.peak_dbfs) else None,
|
||
"clipped_samples": result.clipped_samples,
|
||
"xruns": result.xruns,
|
||
"queue_overflows": result.overflow_blocks,
|
||
"files": [os.path.basename(f) for f in result.files],
|
||
"markers": [asdict(m) for m in result.markers],
|
||
"analysis": result.analysis,
|
||
"notes": result.notes,
|
||
}
|
||
if extra:
|
||
meta.update(extra)
|
||
base = os.path.splitext(result.primary_file)[0] if result.primary_file else None
|
||
if not base:
|
||
return written
|
||
json_path = f"{base}.json"
|
||
with open(json_path, "w", encoding="utf-8") as fh:
|
||
json.dump(meta, fh, ensure_ascii=False, indent=2)
|
||
written.append(json_path)
|
||
|
||
txt_path = f"{base}.txt"
|
||
with open(txt_path, "w", encoding="utf-8") as fh:
|
||
fh.write(render_report(result))
|
||
written.append(txt_path)
|
||
return written
|
||
|
||
|
||
def render_report(result: TakeResult) -> str:
|
||
"""生成人类可读的录音报告（也用于界面上的"体检"面板）。"""
|
||
cfg = result.config
|
||
an = result.analysis or {}
|
||
lines = [
|
||
"RecorderStudio 录音报告",
|
||
"=" * 46,
|
||
f"开始时间 : {result.started_at}",
|
||
f"结束时间 : {result.ended_at}",
|
||
f"输入设备 : {result.device_label or '—'}",
|
||
f"录制格式 : {result.format_label or '—'}",
|
||
f"独占模式 : {'是' if (cfg and cfg.exclusive) else '否'}",
|
||
f"软件增益 : {cfg.gain_db:+.1f} dB" if cfg else "",
|
||
f"低切滤波 : {cfg.lowcut_hz:.0f} Hz" if cfg and cfg.lowcut_hz > 0 else "低切滤波 : 关闭",
|
||
f"抖动 : {'开启 (TPDF)' if (cfg and cfg.dither) else '关闭'}",
|
||
"",
|
||
f"总时长 : {format_duration(result.duration)}",
|
||
f"总采样帧 : {result.frames}",
|
||
f"数据量 : {format_bytes(result.bytes_written)}",
|
||
f"文件数 : {len(result.files)}",
|
||
]
|
||
for f in result.files:
|
||
try:
|
||
sz = os.path.getsize(f)
|
||
except OSError:
|
||
sz = 0
|
||
lines.append(f" · {os.path.basename(f)} ({format_bytes(sz)})")
|
||
lines += [
|
||
"",
|
||
"音质体检",
|
||
"-" * 46,
|
||
f"采样峰值 : {_fmt_list(an.get('peak_dbfs'))} dBFS",
|
||
f"真峰值 : {_fmt_list(an.get('true_peak_dbtp'))} dBTP"
|
||
+ ("（已达上限，建议降低增益）"
|
||
if _maxf(an.get("true_peak_dbtp")) is not None
|
||
and _maxf(an.get("true_peak_dbtp")) > -0.1 else ""),
|
||
f"RMS 电平 : {_fmt_list(an.get('rms_dbfs'))} dBFS",
|
||
f"整体响度 : {an.get('integrated_lufs')} LUFS",
|
||
f"动态范围 : {an.get('loudness_range_lu')} LU",
|
||
f"直流偏移 : {an.get('dc_offset')}",
|
||
f"本底噪声 : {_fmt_list(an.get('noise_floor_dbfs'))} dBFS"
|
||
+ ("" if an.get("noise_floor_available") else "（录音中未检测到静音段，无法测定）"),
|
||
f"削波样本 : {an.get('clipped_total', result.clipped_samples)}",
|
||
f"丢弃块/溢出: {result.overflow_blocks}",
|
||
f"驱动层 xrun: {result.xruns}",
|
||
]
|
||
if result.markers:
|
||
lines += ["", "标记", "-" * 46]
|
||
for m in result.markers:
|
||
lines.append(f" {m.seconds:8.3f} s {m.label} ({m.file})")
|
||
if result.notes:
|
||
lines += ["", "运行日志", "-" * 46] + [f" · {n}" for n in result.notes]
|
||
if an.get("segment_count"):
|
||
lines += ["", f"注：本次录音共 {an['segment_count']} 个分段，以上为合并统计。"]
|
||
return "\n".join(l for l in lines if l is not None)
|
||
|
||
|
||
def _fmt_list(v) -> str:
|
||
if v is None:
|
||
return "—"
|
||
if isinstance(v, list):
|
||
return ", ".join("—" if x is None else f"{x:+.2f}" for x in v)
|
||
return f"{v:+.2f}"
|
||
|
||
|
||
def _maxf(v) -> float | None:
|
||
if v is None:
|
||
return None
|
||
if isinstance(v, list):
|
||
vals = [x for x in v if x is not None]
|
||
return max(vals) if vals else None
|
||
return float(v)
|
||
|
||
|
||
# ------------------------------------------------------- 波形预览 (PNG)
|
||
def _png_chunk(tag: bytes, data: bytes) -> bytes:
|
||
return (struct.pack(">I", len(data)) + tag + data
|
||
+ struct.pack(">I", zlib.crc32(tag + data) & 0xFFFFFFFF))
|
||
|
||
|
||
def write_waveform_png(path: str, data: np.ndarray, samplerate: int,
|
||
*, width: int = 1600, height: int = 320,
|
||
bg=(18, 20, 26), wave=(90, 200, 255), mid=(70, 78, 96),
|
||
rms_color=(255, 190, 80)) -> str:
|
||
"""用纯 numpy + zlib 画一张波形预览图（不依赖 PIL）。"""
|
||
x = np.asarray(data, dtype=np.float32)
|
||
if x.ndim == 1:
|
||
x = x[:, None]
|
||
n, ch = x.shape
|
||
if n == 0:
|
||
x = np.zeros((1, 1), np.float32)
|
||
n, ch = 1, 1
|
||
canvas = np.zeros((height, width, 3), dtype=np.uint8)
|
||
canvas[:, :] = bg
|
||
|
||
lanes = ch if ch <= 2 else 2
|
||
lane_h = height // lanes
|
||
per = max(1, n // width)
|
||
usable = (n // per) * per
|
||
block = x[:usable].reshape(-1, per, ch)
|
||
mn = block.min(axis=1)
|
||
mx = block.max(axis=1)
|
||
rms = np.sqrt(np.mean(block.astype(np.float64) ** 2, axis=1))
|
||
cols = mn.shape[0]
|
||
|
||
for lane in range(lanes):
|
||
y0 = lane * lane_h
|
||
yc = y0 + lane_h // 2
|
||
canvas[max(0, yc - 1):yc + 1, :] = mid
|
||
src = lane if ch <= 2 else 0
|
||
for c in range(cols):
|
||
xx = int(c * width / max(1, cols))
|
||
if xx >= width:
|
||
continue
|
||
top = int(yc - mx[c, src] * (lane_h / 2 - 4))
|
||
bot = int(yc - mn[c, src] * (lane_h / 2 - 4))
|
||
top = max(y0, min(y0 + lane_h - 1, top))
|
||
bot = max(y0, min(y0 + lane_h - 1, bot))
|
||
if bot < top:
|
||
top, bot = bot, top
|
||
canvas[top:bot + 1, xx] = wave
|
||
r = float(rms[c, src]) * (lane_h / 2 - 4)
|
||
canvas[max(y0, int(yc - r)):min(y0 + lane_h, int(yc + r) + 1), xx] = rms_color
|
||
|
||
raw = b"".join(b"\x00" + canvas[y].tobytes() for y in range(height))
|
||
png = (b"\x89PNG\r\n\x1a\n"
|
||
+ _png_chunk(b"IHDR", struct.pack(">IIBBBBB", width, height, 8, 2, 0, 0, 0))
|
||
+ _png_chunk(b"IDAT", zlib.compress(raw, 6))
|
||
+ _png_chunk(b"IEND", b""))
|
||
with open(path, "wb") as fh:
|
||
fh.write(png)
|
||
return path
|
||
|
||
|
||
def make_preview_for(result: TakeResult, *, seconds: float | None = None) -> str | None:
|
||
"""为录音结果生成波形预览图 + 报告文本。"""
|
||
if not result.primary_file or not os.path.exists(result.primary_file):
|
||
return None
|
||
try:
|
||
data, sr = _read_whole(result.primary_file)
|
||
if seconds is not None and len(data) > seconds * sr:
|
||
data = data[:int(seconds * sr)]
|
||
base = os.path.splitext(result.primary_file)[0]
|
||
return write_waveform_png(f"{base}_waveform.png", data, sr)
|
||
except Exception:
|
||
return None
|
||
|
||
|
||
def write_readme_for_session(result: TakeResult) -> list[str]:
|
||
"""额外的"每次录音都留一份说明"的兜底函数（供 CLI 使用）。"""
|
||
out: list[str] = []
|
||
if not result.primary_file:
|
||
return out
|
||
base = os.path.splitext(result.primary_file)[0]
|
||
p = f"{base}_info.txt"
|
||
with open(p, "w", encoding="utf-8") as fh:
|
||
fh.write(render_report(result) + "\n")
|
||
out.append(p)
|
||
return out
|
||
|
||
|
||
def regenerate_report(path: str) -> str:
|
||
"""对已有录音重新生成报告（界面上的"重新体检"）。"""
|
||
an = dsp.analyze_file(path)
|
||
lines = [
|
||
"RecorderStudio 文件体检",
|
||
"=" * 46,
|
||
f"文件 : {os.path.basename(path)}",
|
||
f"生成时间 : {datetime.now().isoformat(timespec='seconds')}",
|
||
f"格式 : {an.get('format')}",
|
||
f"时长 : {format_duration(an.get('duration', 0.0))}",
|
||
f"数据量 : {format_bytes(an.get('data_bytes', 0))}",
|
||
"",
|
||
f"采样峰值 : {_fmt_list(an.get('peak_dbfs'))} dBFS",
|
||
f"真峰值 : {_fmt_list(an.get('true_peak_dbtp'))} dBTP",
|
||
f"RMS 电平 : {_fmt_list(an.get('rms_dbfs'))} dBFS",
|
||
f"整体响度 : {an.get('integrated_lufs')} LUFS",
|
||
f"动态范围 : {an.get('loudness_range_lu')} LU",
|
||
f"直流偏移 : {an.get('dc_offset')}",
|
||
f"本底噪声 : {_fmt_list(an.get('noise_floor_dbfs'))} dBFS",
|
||
f"削波样本 : {an.get('clipped_total')}",
|
||
]
|
||
return "\n".join(lines)
|