From c2b0f7e4fb4738afcae1705db8f983dd90a669a4 Mon Sep 17 00:00:00 2001 From: historia Date: Sun, 6 Sep 2026 15:17:57 -0400 Subject: inital commit --- lib/src/__main__.py | 4 + lib/src/producer/__init__.py | 1 + lib/src/producer/cli.py | 206 +++++++++++++++++++ lib/src/producer/config.py | 258 +++++++++++++++++++++++ lib/src/producer/doctor.py | 78 +++++++ lib/src/producer/dsp.py | 273 +++++++++++++++++++++++++ lib/src/producer/engines/__init__.py | 0 lib/src/producer/engines/base.py | 34 +++ lib/src/producer/engines/denoise_dfn.py | 98 +++++++++ lib/src/producer/engines/denoise_zip.py | 22 ++ lib/src/producer/engines/enhance_mossformer.py | 38 ++++ lib/src/producer/engines/enhance_resemble.py | 74 +++++++ lib/src/producer/engines/resemble_worker.py | 37 ++++ lib/src/producer/io.py | 87 ++++++++ lib/src/producer/lazy.py | 85 ++++++++ lib/src/producer/loudness.py | 30 +++ lib/src/producer/meters.py | 75 +++++++ lib/src/producer/pipeline.py | 169 +++++++++++++++ lib/src/producer/report.py | 56 +++++ 19 files changed, 1625 insertions(+) create mode 100644 lib/src/__main__.py create mode 100644 lib/src/producer/__init__.py create mode 100644 lib/src/producer/cli.py create mode 100644 lib/src/producer/config.py create mode 100644 lib/src/producer/doctor.py create mode 100644 lib/src/producer/dsp.py create mode 100644 lib/src/producer/engines/__init__.py create mode 100644 lib/src/producer/engines/base.py create mode 100644 lib/src/producer/engines/denoise_dfn.py create mode 100644 lib/src/producer/engines/denoise_zip.py create mode 100644 lib/src/producer/engines/enhance_mossformer.py create mode 100644 lib/src/producer/engines/enhance_resemble.py create mode 100644 lib/src/producer/engines/resemble_worker.py create mode 100644 lib/src/producer/io.py create mode 100644 lib/src/producer/lazy.py create mode 100644 lib/src/producer/loudness.py create mode 100644 lib/src/producer/meters.py create mode 100644 lib/src/producer/pipeline.py create mode 100644 lib/src/producer/report.py (limited to 'lib/src') diff --git a/lib/src/__main__.py b/lib/src/__main__.py new file mode 100644 index 0000000..cfe0046 --- /dev/null +++ b/lib/src/__main__.py @@ -0,0 +1,4 @@ +from producer.cli import main + +if __name__ == "__main__": + main() diff --git a/lib/src/producer/__init__.py b/lib/src/producer/__init__.py new file mode 100644 index 0000000..3dc1f76 --- /dev/null +++ b/lib/src/producer/__init__.py @@ -0,0 +1 @@ +__version__ = "0.1.0" diff --git a/lib/src/producer/cli.py b/lib/src/producer/cli.py new file mode 100644 index 0000000..745b797 --- /dev/null +++ b/lib/src/producer/cli.py @@ -0,0 +1,206 @@ +from __future__ import annotations + +import argparse +import sys +from pathlib import Path + +from . import __version__, pipeline +from . import config as cfgmod +from . import dsp as pdsp +from . import io as pio +from . import report as repmod +from .config import DSP_KEYS, PROFILES, Options + +DATA_DIR = Path(__file__).resolve().parents[2] +EXTS = {".wav", ".flac", ".mp3", ".m4a", ".aac", ".ogg", ".opus", ".aif", ".aiff", ".wma"} + + +def build_parser() -> argparse.ArgumentParser: + p = argparse.ArgumentParser( + prog="producer", + description="One-click narration/podcast mastering: denoise, enhance, voice DSP, loudness.", + ) + p.add_argument("inputs", nargs="*", help="audio file(s) to process") + p.add_argument("doctor", nargs="?", help=argparse.SUPPRESS) + p.add_argument("-o", "--output", help="output file, or directory for batch") + p.add_argument("--profile", choices=tuple(PROFILES), help="audiobook (default) or podcast") + p.add_argument( + "--denoise", choices=["dfn3", "zipenhancer", "off"], help="denoise engine (default dfn3)" + ) + p.add_argument("--denoise-strength", type=float, help="0-1 blend of denoised signal") + p.add_argument( + "--enhance", choices=["off", "mossformer2", "resemble"], help="speech enhancement engine" + ) + p.add_argument("--enhance-strength", type=float, help="0-1 blend of enhanced signal") + p.add_argument( + "--no-dsp", action="store_false", dest="dsp", help="skip EQ/compression/de-ess chain" + ) + p.add_argument( + "--no-levelling", action="store_false", dest="levelling", help="skip loudness stage" + ) + for key in DSP_KEYS: + p.add_argument(f"--{key}", type=float, metavar="0-1", help=f"strength for {key} stage") + p.add_argument("--hpf-hz", type=float, help="high-pass corner (default 80)") + p.add_argument("--target", type=float, help="loudness target (RMS dB or LUFS per profile)") + p.add_argument("--ceiling", type=float, dest="ceiling_db", help="true-peak ceiling in dB") + p.add_argument("--sample-rate", type=int, help="output sample rate (44100/48000)") + p.add_argument( + "--bit-depth", type=int, choices=[16, 24, 32], default=None, help="wav/flac bit depth" + ) + p.add_argument( + "--format", choices=["wav", "flac", "mp3"], dest="out_format", help="output format" + ) + p.add_argument("--device", choices=["auto", "cuda", "cpu"], help="compute device") + p.add_argument("--batch", action="store_true", help="inputs are directories/globs to expand") + p.add_argument("--report", action="store_true", help="write .report.json") + p.add_argument("--report-path", help="explicit report file path") + p.add_argument("--dry-run", action="store_true", help="show the processing chain and exit") + p.add_argument("--config", help="config file (default config.toml in the project root)") + p.add_argument("-v", "--verbose", action="count", default=0) + p.add_argument("--version", action="store_true") + return p + + +def _apply_args(opts: Options, args: argparse.Namespace) -> None: + m = { + "profile": "profile", + "denoise": "denoise", + "denoise_strength": "denoise_strength", + "enhance": "enhance", + "enhance_strength": "enhance_strength", + "output": "output", + "hpf_hz": "hpf_hz", + "target": "target", + "ceiling_db": "ceiling_db", + "sample_rate": "sample_rate", + "bit_depth": "bit_depth", + "out_format": "out_format", + "device": "device", + "report": "report", + "report_path": "report_path", + "dry_run": "dry_run", + "verbose": "verbose", + } + for arg_name, field_name in m.items(): + v = getattr(args, arg_name, None) + if v is not None: + setattr(opts, field_name, v) + if args.dsp is False: + opts.dsp = False + if args.levelling is False: + opts.levelling = False + for key in DSP_KEYS: + v = getattr(args, key, None) + if v is not None: + opts.strengths[key] = float(v) + if opts.profile not in PROFILES: + raise SystemExit(f"unknown profile: {opts.profile}") + if not 0.0 <= opts.denoise_strength <= 1.0 or not 0.0 <= opts.enhance_strength <= 1.0: + raise SystemExit("engine strengths must be within 0-1") + for k in DSP_KEYS: + v = opts.strengths[k] + if v is not None and not 0.0 <= float(v) <= 1.0: + raise SystemExit(f"--{k} must be within 0-1") + + +def _expand_inputs(args: argparse.Namespace) -> list[Path]: + inputs: list[Path] = [] + for raw in args.inputs: + p = Path(raw) + if args.batch and p.is_dir(): + inputs.extend(sorted(q for q in p.iterdir() if q.suffix.lower() in EXTS)) + elif args.batch and not p.exists(): + import glob + + for q in sorted(glob.glob(raw)): + q = Path(q) + if q.suffix.lower() in EXTS: + inputs.append(q) + else: + inputs.append(p) + return inputs + + +def _resolve_output(inp: Path, opts: Options, single: bool) -> Path: + ext = opts.out_format.lower() + if opts.output: + o = Path(opts.output) + if single: + return o if o.suffix else o.with_suffix("." + ext) + o.mkdir(parents=True, exist_ok=True) + return o / f"{inp.stem}_master.{ext}" + return inp.with_name(f"{inp.stem}_master.{ext}") + + +def process_one(inp: Path, opts: Options, single: bool, quiet: bool = False) -> Path: + x, sr = pio.decode(inp) + out_path = _resolve_output(inp, opts, single) + target_sr = opts.out_sample_rate() + if not quiet: + print(f"[producer] {inp} ({sr} Hz, {len(x) / sr:.1f}s)") + res = pipeline.run_pipeline(x, sr, opts) + y = res.audio + if target_sr != sr: + y = pdsp.resample(y, sr, target_sr) + pio.encode(y, target_sr, out_path, opts.out_format, opts.bit_depth) + rep = repmod.build(str(inp), str(out_path), opts, res.before, res.after, res.timings, res.notes) + if opts.report or opts.report_path: + rp = ( + Path(opts.report_path) + if opts.report_path + else out_path.with_name(out_path.stem + ".report.json") + ) + repmod.save(rep, rp) + if not quiet: + repmod.print_human(rep) + print(f"[producer] wrote {out_path}") + return out_path + + +def main(argv: list[str] | None = None) -> int: + argv = list(sys.argv[1:] if argv is None else argv) + if argv and argv[0] == "doctor": + from . import doctor + + return doctor.main(argv[1:]) + parser = build_parser() + args = parser.parse_args(argv) + if args.version: + print(f"producer {__version__}") + return 0 + opts = Options() + cfg_path = Path(args.config) if args.config else DATA_DIR.parent / "config.toml" + if not args.config: + cfgmod.write_default_config(cfg_path) + cfgmod.apply_config(opts, cfgmod.load_config(cfg_path)) + _apply_args(opts, args) + inputs = _expand_inputs(args) + if not inputs: + parser.error("no input files given") + if opts.dry_run: + stages = pipeline.build_stages(opts) + print(f"producer dry run (profile={opts.profile}, format={opts.out_format})") + for st in stages: + state = "on " if st.enabled else "off" + print(f" [{state}] {st.name:<10} {st.detail}") + print(f" output: {opts.out_format} @ {opts.out_sample_rate()} Hz, {opts.bit_depth}-bit") + return 0 + (DATA_DIR / "models").mkdir(parents=True, exist_ok=True) + (DATA_DIR / "cache").mkdir(parents=True, exist_ok=True) + failed = 0 + single = len(inputs) == 1 + for inp in inputs: + try: + process_one(inp, opts, single, quiet=len(inputs) > 1) + except Exception as e: + failed += 1 + print(f"[producer] ERROR {inp}: {e}", file=sys.stderr) + if opts.verbose: + import traceback + + traceback.print_exc() + return 1 if failed else 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/lib/src/producer/config.py b/lib/src/producer/config.py new file mode 100644 index 0000000..e88e80b --- /dev/null +++ b/lib/src/producer/config.py @@ -0,0 +1,258 @@ +from __future__ import annotations + +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any + +import tomllib + +DSP_KEYS = ( + "hpf", + "mud", + "warmth", + "soothe", + "compress", + "tape", + "deess", + "presence", + "air", + "breath", +) + + +@dataclass +class Profile: + name: str + loudness_mode: str = "rms" + target: float = -20.0 + ceiling_db: float = -3.0 + sample_rate: int = 44100 + hpf_hz: float = 80.0 + mud: tuple[float, float, float] = (300.0, -2.5, 1.0) + warmth: tuple[float, float] = (150.0, 1.5) + comp1: tuple[float, float, float, float] = (-20.0, 2.0, 15.0, 150.0) + comp2: tuple[float, float, float, float] = (-18.0, 3.0, 5.0, 100.0) + deess: tuple[float, float, float] = (5500.0, 8000.0, 5.0) + presence: tuple[float, float] = (3000.0, 1.5) + air: tuple[float, float] = (10000.0, 1.5) + strengths: dict[str, float] = field( + default_factory=lambda: { + "hpf": 1.0, + "mud": 0.8, + "warmth": 0.8, + "soothe": 0.3, + "compress": 0.8, + "tape": 0.0, + "deess": 0.6, + "presence": 0.8, + "air": 0.6, + "breath": 0.35, + } + ) + + +AUDIOBOOK = Profile("audiobook") +PODCAST = Profile( + "podcast", + loudness_mode="lufs", + target=-16.0, + ceiling_db=-1.5, + sample_rate=48000, + mud=(300.0, -1.5, 1.0), + warmth=(150.0, 1.0), + comp1=(-18.0, 2.0, 15.0, 150.0), + comp2=(-16.0, 3.0, 5.0, 100.0), + deess=(5500.0, 8000.0, 6.0), + presence=(3000.0, 2.5), + air=(10000.0, 2.0), + strengths={ + "hpf": 1.0, + "mud": 0.6, + "warmth": 0.7, + "soothe": 0.5, + "compress": 0.9, + "tape": 0.2, + "deess": 0.7, + "presence": 1.0, + "air": 0.8, + "breath": 0.2, + }, +) +RADIO = Profile( + "radio", + loudness_mode="lufs", + target=-16.0, + ceiling_db=-1.5, + sample_rate=48000, + hpf_hz=70.0, + mud=(280.0, -3.5, 1.3), + warmth=(100.0, 3.0), + comp1=(-16.0, 3.0, 10.0, 120.0), + comp2=(-14.0, 4.0, 3.0, 90.0), + deess=(5000.0, 7500.0, 4.0), + presence=(3500.0, 1.0), + air=(9000.0, 1.0), + strengths={ + "hpf": 1.0, + "mud": 0.9, + "warmth": 1.0, + "soothe": 0.7, + "compress": 1.0, + "tape": 0.55, + "deess": 0.5, + "presence": 0.7, + "air": 0.5, + "breath": 0.4, + }, +) +PROFILES = {"audiobook": AUDIOBOOK, "podcast": PODCAST, "radio": RADIO} + + +@dataclass +class Options: + profile: str = "audiobook" + denoise: str = "dfn3" + denoise_strength: float = 1.0 + enhance: str = "off" + enhance_strength: float = 1.0 + dsp: bool = True + levelling: bool = True + target: float | None = None + ceiling_db: float | None = None + sample_rate: int | None = None + bit_depth: int = 16 + out_format: str = "wav" + device: str = "auto" + strengths: dict[str, float | None] = field(default_factory=lambda: {k: None for k in DSP_KEYS}) + hpf_hz: float | None = None + output: str | None = None + report: bool = False + report_path: str | None = None + dry_run: bool = False + verbose: int = 0 + + def prof(self) -> Profile: + return PROFILES[self.profile] + + def eff(self, key: str) -> float: + v = self.strengths.get(key) + if v is None: + return float(self.prof().strengths[key]) + return float(v) + + def target_value(self) -> float: + return self.prof().target if self.target is None else float(self.target) + + def ceiling(self) -> float: + return self.prof().ceiling_db if self.ceiling_db is None else float(self.ceiling_db) + + def out_sample_rate(self) -> int: + return self.prof().sample_rate if self.sample_rate is None else int(self.sample_rate) + + def loudness_mode(self) -> str: + return self.prof().loudness_mode + + def to_dict(self) -> dict[str, Any]: + return { + "profile": self.profile, + "denoise": self.denoise, + "denoise_strength": self.denoise_strength, + "enhance": self.enhance, + "enhance_strength": self.enhance_strength, + "dsp": self.dsp, + "levelling": self.levelling, + "target": self.target_value(), + "ceiling_db": self.ceiling(), + "sample_rate": self.out_sample_rate(), + "bit_depth": self.bit_depth, + "out_format": self.out_format, + "device": self.device, + "strengths": {k: self.eff(k) for k in DSP_KEYS}, + "hpf_hz": self.hpf_hz if self.hpf_hz is not None else self.prof().hpf_hz, + } + + +_CFG_FIELDS = { + "profile": "profile", + "denoise_strength": "denoise_strength", + "enhance_strength": "enhance_strength", + "format": "out_format", + "bit_depth": "bit_depth", + "device": "device", + "target": "target", + "ceiling": "ceiling_db", + "sample_rate": "sample_rate", +} + + +def _engine_cfg(value: Any) -> tuple[Any, Any]: + if isinstance(value, dict): + return value.get("engine"), value.get("strength") + return value, None + + +def load_config(path: Path) -> dict[str, Any]: + if not path.exists(): + return {} + with path.open("rb") as f: + return tomllib.load(f) + + +def apply_config(opts: Options, cfg: dict[str, Any]) -> None: + for key, field_name in _CFG_FIELDS.items(): + if key in cfg: + setattr(opts, field_name, cfg[key]) + if "denoise" in cfg: + eng, strength = _engine_cfg(cfg["denoise"]) + if eng: + opts.denoise = eng + if strength is not None: + opts.denoise_strength = float(strength) + if "enhance" in cfg: + eng, strength = _engine_cfg(cfg["enhance"]) + if eng: + opts.enhance = eng + if strength is not None: + opts.enhance_strength = float(strength) + prof_cfg = cfg.get(opts.profile) + if isinstance(prof_cfg, dict): + for k in DSP_KEYS: + if k in prof_cfg: + opts.strengths[k] = float(prof_cfg[k]) + if "hpf_hz" in prof_cfg: + opts.hpf_hz = float(prof_cfg["hpf_hz"]) + + +def write_default_config(path: Path) -> None: + if path.exists(): + return + lines = [ + 'profile = "audiobook"', + "", + "[denoise]", + 'engine = "dfn3"', + "strength = 1.0", + "", + "[enhance]", + 'engine = "off"', + "", + "[audiobook]", + "mud = 0.8", + "warmth = 0.8", + "compress = 0.8", + "deess = 0.6", + "presence = 0.8", + "air = 0.6", + "breath = 0.35", + "", + "[podcast]", + "mud = 0.6", + "warmth = 0.7", + "compress = 0.9", + "deess = 0.7", + "presence = 1.0", + "air = 0.8", + "breath = 0.2", + "", + ] + path.write_text("\n".join(lines)) diff --git a/lib/src/producer/doctor.py b/lib/src/producer/doctor.py new file mode 100644 index 0000000..482a0aa --- /dev/null +++ b/lib/src/producer/doctor.py @@ -0,0 +1,78 @@ +from __future__ import annotations + +import importlib.util +import shutil +import sys +from pathlib import Path + +from . import __version__ + +DATA_DIR = Path(__file__).resolve().parents[2] + + +def _check(name: str, ok: bool, detail: str = "") -> bool: + mark = "ok " if ok else "MISS" + line = f" [{mark}] {name}" + if detail: + line += f": {detail}" + print(line) + return ok + + +def _import(name: str): + try: + return importlib.import_module(name) + except Exception: + return None + + +def main(_argv: list[str] | None = None) -> int: + print(f"producer {__version__} doctor") + print(f" data dir: {DATA_DIR}") + critical_ok = True + + v = sys.version_info + critical_ok &= _check("python", v >= (3, 10), f"{v.major}.{v.minor}.{v.micro}") + + for mod in ("numpy", "scipy", "soundfile", "pyloudnorm"): + m = _import(mod) + ok = m is not None + critical_ok &= ok + _check(mod, ok, getattr(m, "__version__", "") if ok else "not installed") + + torch = _import("torch") + if torch is None: + _check("torch", False, "not installed (installs on first denoise/enhance run)") + else: + ver = getattr(torch, "__version__", "?") + cuda = bool(torch.cuda.is_available()) + dev = torch.cuda.get_device_name(0) if cuda else "" + _check("torch", True, f"{ver}, cuda={cuda}" + (f" ({dev})" if dev else "")) + + ffmpeg = shutil.which("ffmpeg") + _check("ffmpeg", ffmpeg is not None, ffmpeg or "missing (needed for mp3 + odd formats)") + + for mod in ("deepfilternet", "zipenhancer", "clearvoice"): + m = _import(mod) or _import(mod.replace("-", "_")) + _check( + mod, m is not None, "installed" if m is not None else "lazy (installed on first use)" + ) + + dfn3 = DATA_DIR / "models" / "DeepFilterNet3" / "config.ini" + _check("dfn3 weights", dfn3.is_file(), str(dfn3)) + + res_venv = DATA_DIR / "venvs" / "resemble" / "bin" / "python" + _check("resemble venv", res_venv.is_file(), str(res_venv)) + + uv = DATA_DIR / "bin" / "uv" + _check("uv", uv.is_file() or shutil.which("uv") is not None, str(uv)) + + cfg = DATA_DIR.parent / "config.toml" + _check("config.toml", cfg.exists(), str(cfg)) + + print("doctor:", "core OK" if critical_ok else "core problems detected") + return 0 if critical_ok else 1 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/lib/src/producer/dsp.py b/lib/src/producer/dsp.py new file mode 100644 index 0000000..60cdbc6 --- /dev/null +++ b/lib/src/producer/dsp.py @@ -0,0 +1,273 @@ +from __future__ import annotations + +import numpy as np +from scipy import ndimage, signal + +from .meters import SILENCE_DB + +_EPS = 1e-12 + + +def _coef(time_ms: float, sr: int) -> float: + return float(np.exp(-1000.0 / (max(time_ms, 1e-4) * sr))) + + +def _onepole(x: np.ndarray, a: float) -> np.ndarray: + return signal.lfilter([1.0 - a], [1.0, -a], x) + + +def rbj(kind: str, sr: int, freq: float, gain_db: float = 0.0, q: float = 0.7071) -> np.ndarray: + a = 10.0 ** (gain_db / 40.0) + w0 = 2.0 * np.pi * freq / sr + cs = np.cos(w0) + sn = np.sin(w0) + alpha = sn / (2.0 * q) + if kind == "highpass": + b = [(1 + cs) / 2, -(1 + cs), (1 + cs) / 2] + aa = [1 + alpha, -2 * cs, 1 - alpha] + elif kind == "lowpass": + b = [(1 - cs) / 2, 1 - cs, (1 - cs) / 2] + aa = [1 + alpha, -2 * cs, 1 - alpha] + elif kind == "peaking": + b = [1 + alpha * a, -2 * cs, 1 - alpha * a] + aa = [1 + alpha / a, -2 * cs, 1 - alpha / a] + elif kind == "lowshelf": + sq = 2 * np.sqrt(a) * alpha + b = [ + a * ((a + 1) - (a - 1) * cs + sq), + 2 * a * ((a - 1) - (a + 1) * cs), + a * ((a + 1) - (a - 1) * cs - sq), + ] + aa = [ + (a + 1) + (a - 1) * cs + sq, + -2 * ((a - 1) + (a + 1) * cs), + (a + 1) + (a - 1) * cs - sq, + ] + elif kind == "highshelf": + sq = 2 * np.sqrt(a) * alpha + b = [ + a * ((a + 1) + (a - 1) * cs + sq), + -2 * a * ((a - 1) + (a + 1) * cs), + a * ((a + 1) + (a - 1) * cs - sq), + ] + aa = [ + (a + 1) - (a - 1) * cs + sq, + 2 * ((a - 1) - (a + 1) * cs), + (a + 1) - (a - 1) * cs - sq, + ] + else: + raise ValueError(f"unknown filter kind: {kind}") + arr = np.array(b + aa, dtype=np.float64) + return arr / arr[3] + + +def biquad(x: np.ndarray, sos: np.ndarray) -> np.ndarray: + y = signal.sosfilt(sos[None, :] if sos.ndim == 1 else sos, x.astype(np.float64)) + return y.astype(np.float32) + + +def _as_sos(sos: np.ndarray) -> np.ndarray: + return sos[None, :] if sos.ndim == 1 else sos + + +def hpf(x: np.ndarray, sr: int, hz: float) -> np.ndarray: + return biquad(x, rbj("highpass", sr, hz)) + + +def shelf(x: np.ndarray, sr: int, freq: float, gain_db: float, low: bool = True) -> np.ndarray: + if abs(gain_db) < 0.01: + return x + return biquad(x, rbj("lowshelf" if low else "highshelf", sr, freq, gain_db)) + + +def peak_eq(x: np.ndarray, sr: int, freq: float, gain_db: float, q: float = 1.0) -> np.ndarray: + if abs(gain_db) < 0.01: + return x + return biquad(x, rbj("peaking", sr, freq, gain_db, q)) + + +def compressor( + x: np.ndarray, + sr: int, + threshold_db: float, + ratio: float, + attack_ms: float, + release_ms: float, + knee_db: float = 6.0, +) -> np.ndarray: + x64 = x.astype(np.float64) + a_att = _coef(attack_ms, sr) + a_rel = _coef(release_ms, sr) + env = np.sqrt(np.clip(_onepole(np.square(x64), a_att), 0.0, None)) + level_db = 20.0 * np.log10(env + _EPS) + over = level_db - threshold_db + k = knee_db + gr = np.where( + over <= -k / 2, + 0.0, + np.where( + over < k / 2, + (1.0 - 1.0 / ratio) * np.square(over + k / 2) / (2.0 * k), + (1.0 - 1.0 / ratio) * over, + ), + ) + gr = _onepole(np.clip(gr, 0.0, None), a_rel) + gain = 10.0 ** (-gr / 20.0) + return (x64 * gain).astype(np.float32) + + +def deesser( + x: np.ndarray, sr: int, lo_hz: float, hi_hz: float, max_reduction_db: float +) -> np.ndarray: + if max_reduction_db < 0.05: + return x + x64 = x.astype(np.float64) + low = signal.sosfiltfilt(signal.butter(4, lo_hz, btype="lowpass", fs=sr, output="sos"), x64) + high = x64 - low + mid = signal.sosfiltfilt( + signal.butter(4, [lo_hz, hi_hz], btype="bandpass", fs=sr, output="sos"), high + ) + win = max(3, int(0.005 * sr) | 1) + env = signal.convolve(np.abs(mid), np.hanning(win) / np.sum(np.hanning(win)), mode="same") + act = env[env > _EPS] + if act.size == 0: + return x + thr = float(np.percentile(act, 95)) * 10.0 ** (-3.0 / 20.0) + over = np.clip(20.0 * np.log10((env + _EPS) / thr), 0.0, None) + gr = np.clip(over * 0.6, 0.0, max_reduction_db) + gr = _onepole(_onepole(gr, _coef(1.0, sr)), _coef(30.0, sr)) + gain = 10.0 ** (-gr / 20.0) + high_out = high - mid + mid * gain + return (low + high_out).astype(np.float32) + + +def expander( + x: np.ndarray, + sr: int, + max_drop_db: float = 6.0, + ratio: float = 2.0, + attack_ms: float = 10.0, + release_ms: float = 120.0, +) -> np.ndarray: + if max_drop_db < 0.05: + return x + n = x.size + frame = max(1, int(0.020 * sr)) + nf = n // frame + if nf < 2: + return x + frames = x[: nf * frame].reshape(nf, frame).astype(np.float64) + frms = np.sqrt(np.mean(np.square(frames), axis=1)) + fdb = 20.0 * np.log10(frms + _EPS) + active = fdb[fdb > SILENCE_DB + 1.0] + if active.size == 0: + return x + speech_ref = float(np.percentile(active, 90)) + floor_ref = float(np.percentile(active, 5)) + thr = 0.5 * (speech_ref + floor_ref) + drop = np.where(fdb < thr, np.minimum((thr - fdb) * (1.0 - 1.0 / ratio), max_drop_db), 0.0) + centers = (np.arange(nf) + 0.5) * frame + drop_s = np.interp(np.arange(n), centers, drop) + drop_s = np.maximum( + _onepole(drop_s, _coef(release_ms, sr)), _onepole(drop_s, _coef(attack_ms, sr)) + ) + gain = 10.0 ** (-np.clip(drop_s, 0.0, max_drop_db) / 20.0) + return (x.astype(np.float64) * gain).astype(np.float32) + + +def limit( + x: np.ndarray, + sr: int, + ceiling_db: float, + release_ms: float = 60.0, + lookahead_ms: float = 1.0, +) -> np.ndarray: + x64 = x.astype(np.float64) + n = x64.size + if n == 0: + return x + lin = 10.0 ** (ceiling_db / 20.0) + win = max(1, int(lookahead_ms * sr / 1000.0) | 1) + env = ndimage.maximum_filter1d(np.abs(x64), size=win, mode="nearest") + over = np.clip(20.0 * np.log10(env + _EPS) - ceiling_db, 0.0, None) + block = max(1, int(0.005 * sr)) + nb = (n + block - 1) // block + padded = np.zeros(nb * block, dtype=np.float64) + padded[:n] = over + block_over = padded.reshape(nb, block).max(axis=1) + decay = float(np.exp(-1000.0 * 0.005 / max(release_ms, 0.1))) + held = np.empty(nb, dtype=np.float64) + prev = 0.0 + for i in range(nb): + prev = max(block_over[i], prev * decay) + held[i] = prev + held_s = np.interp(np.arange(n), (np.arange(nb) + 0.5) * block, held) + held_s = _onepole(held_s, _coef(1.0, sr)) + gain = 10.0 ** (-held_s / 20.0) + y = x64 * gain + bad = np.abs(y) > lin + if np.any(bad): + y[bad] = np.sign(y[bad]) * lin + return y.astype(np.float32) + + +def resample(x: np.ndarray, sr_in: int, sr_out: int) -> np.ndarray: + if sr_in == sr_out or x.size == 0: + return x + from math import gcd + + g = gcd(int(sr_in), int(sr_out)) + y = signal.resample_poly( + x.astype(np.float64), int(sr_out) // g, int(sr_in) // g, window=("kaiser", 10.0) + ) + return y.astype(np.float32) + + +def tape(x: np.ndarray, sr: int, amount: float) -> np.ndarray: + """Gentle asymmetric soft-clip saturation — analog-style even-harmonic warmth.""" + if amount < 0.02: + return x + s = float(np.clip(amount, 0.0, 1.0)) + drive = 1.0 + 2.5 * s + x64 = x.astype(np.float64) + curve = np.where(x64 < 0.0, x64 * (1.0 + 0.4 * s), x64) + y = np.tanh(curve * drive) / np.tanh(drive) + rms_in = np.sqrt(np.mean(np.square(x64))) + rms_out = np.sqrt(np.mean(np.square(y))) + y *= rms_in / max(rms_out, _EPS) + return (x64 * (1.0 - s) + y * s).astype(np.float32) + + +def soothe(x: np.ndarray, sr: int, amount: float) -> np.ndarray: + """Dynamic reducer that digs into resonances only while they stick out. + + Targets boxy 200-450 Hz and harsh 2.5-6 kHz bands; static-only when turned + to zero. Deesser-style: band split -> smoothed envelope -> percentile + threshold -> gain reduce -> subtractive recombination. + """ + if amount < 0.02: + return x + s = float(np.clip(amount, 0.0, 1.0)) + y = x.astype(np.float64) + bands = ((200.0, 450.0, 3.5), (2500.0, 6000.0, 5.0)) + win = max(3, int(0.010 * sr) | 1) + kernel = np.hanning(win) / np.sum(np.hanning(win)) + for lo, hi, max_db in bands: + if max_db * s < 0.1: + continue + band = signal.sosfiltfilt( + signal.butter(4, [lo, hi], btype="bandpass", fs=sr, output="sos"), y + ) + env = signal.convolve(np.abs(band), kernel, mode="same") + act = env[env > _EPS] + if act.size == 0: + continue + thr = float(np.percentile(act, 88)) * 10.0 ** (-6.0 / 20.0) + over = np.clip(20.0 * np.log10((env + _EPS) / thr), 0.0, None) + gr = np.clip(over * 0.7, 0.0, max_db) * s + gr = _onepole(_onepole(gr, _coef(5.0, sr)), _coef(60.0, sr)) + gain = 10.0 ** (-gr / 20.0) + y = y - band + band * gain + if np.array_equal(y, x.astype(np.float64)): + return x + return y.astype(np.float32) diff --git a/lib/src/producer/engines/__init__.py b/lib/src/producer/engines/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/lib/src/producer/engines/base.py b/lib/src/producer/engines/base.py new file mode 100644 index 0000000..2988357 --- /dev/null +++ b/lib/src/producer/engines/base.py @@ -0,0 +1,34 @@ +from __future__ import annotations + +import numpy as np + + +def pick_device(pref: str = "auto") -> str: + if pref == "cpu": + return "cpu" + try: + import torch + except ImportError: + if pref == "cuda": + print("[producer] warning: torch not installed; using cpu", flush=True) + return "cpu" + if torch.cuda.is_available(): + return "cuda" + if pref == "cuda": + print("[producer] warning: CUDA requested but unavailable; using cpu", flush=True) + return "cpu" + + +def device_name(device: str) -> str: + if device == "cuda": + import torch + + return torch.cuda.get_device_name(0) + return "cpu" + + +def blend(x: np.ndarray, y: np.ndarray, strength: float) -> np.ndarray: + s = float(np.clip(strength, 0.0, 1.0)) + if s >= 0.999: + return y + return (x.astype(np.float64) * (1.0 - s) + y.astype(np.float64) * s).astype(np.float32) diff --git a/lib/src/producer/engines/denoise_dfn.py b/lib/src/producer/engines/denoise_dfn.py new file mode 100644 index 0000000..66a8cf6 --- /dev/null +++ b/lib/src/producer/engines/denoise_dfn.py @@ -0,0 +1,98 @@ +from __future__ import annotations + +import sys +import types +import urllib.request +import zipfile +from pathlib import Path + +import numpy as np + +from .. import dsp, lazy +from .base import blend, device_name, pick_device + +MODELS_DIR = lazy.DATA_DIR / "models" +TAG = "v0.5.6" +MODEL_ZIPS = { + "DeepFilterNet3": "models/DeepFilterNet3.zip", + "DeepFilterNet2": "models/DeepFilterNet2.zip", +} +BASE_URL = f"https://raw.githubusercontent.com/Rikorose/DeepFilterNet/{TAG}" + + +def _shim_torchaudio_backend() -> None: + try: + import torchaudio.backend # noqa: F401 + except Exception: + pkg = sys.modules.get("torchaudio") + if pkg is not None and "torchaudio.backend" not in sys.modules: + stub = types.ModuleType("torchaudio.backend") + stub.__path__ = [] + sys.modules["torchaudio.backend"] = stub + pkg.backend = stub + + +def ensure_model(model: str) -> Path: + target = MODELS_DIR / model + if (target / "config.ini").is_file(): + return target + url = f"{BASE_URL}/{MODEL_ZIPS[model]}" + MODELS_DIR.mkdir(parents=True, exist_ok=True) + zpath = MODELS_DIR / f"{model}.zip" + print(f"[producer] downloading {model} weights...", flush=True) + urllib.request.urlretrieve(url, zpath) + with zipfile.ZipFile(zpath) as z: + z.extractall(target) + zpath.unlink(missing_ok=True) + if not (target / "config.ini").is_file(): + inner = list(target.glob(f"**/{model}/config.ini")) + if inner: + src = inner[0].parent + for f in src.iterdir(): + f.rename(target / f.name) + if not (target / "config.ini").is_file(): + raise RuntimeError(f"{model} weights download failed") + return target + + +def denoise( + x: np.ndarray, sr: int, strength: float, device_pref: str = "auto" +) -> tuple[np.ndarray, str, str]: + lazy.ensure(["deepfilternet==0.5.6"], purpose="DeepFilterNet") + lazy.ensure_torch() + _shim_torchaudio_backend() + model_name = "DeepFilterNet3" + try: + model_dir = ensure_model(model_name) + except Exception: + model_name = "DeepFilterNet2" + model_dir = ensure_model(model_name) + from df.enhance import enhance as df_enhance + from df.enhance import init_df + + device = pick_device(device_pref) + model, df_state, _ = init_df( + model_base_dir=str(model_dir), + post_filter=strength >= 0.95, + log_level="error", + log_file=None, + ) + try: + model = model.to(device) + dev = device + except Exception: + dev = "cpu" + sr_df = int(df_state.sr()) + xin = dsp.resample(x, sr, sr_df) + import torch + + xin_t = torch.from_numpy(xin.astype(np.float32)).unsqueeze(0) + y = df_enhance(model, df_state, xin_t) + if isinstance(y, torch.Tensor): + y = y.detach().cpu().numpy() + y = np.asarray(y, dtype=np.float32) + if y.ndim > 1: + y = y.reshape(-1) + y = dsp.resample(y, sr_df, sr) + y = blend(x, y, strength) + return y, f"dfn ({model_name})", f"{device_name(dev)} ({dev})" diff --git a/lib/src/producer/engines/denoise_zip.py b/lib/src/producer/engines/denoise_zip.py new file mode 100644 index 0000000..1794a59 --- /dev/null +++ b/lib/src/producer/engines/denoise_zip.py @@ -0,0 +1,22 @@ +from __future__ import annotations + +import numpy as np + +from .. import dsp, lazy +from .base import blend + + +def denoise( + x: np.ndarray, sr: int, strength: float, device_pref: str = "auto" +) -> tuple[np.ndarray, str, str]: + lazy.ensure(["zipenhancer==0.3.2"], purpose="ZipEnhancer") + from zipenhancer import denoise as z_denoise + + sr_z = 16000 + x16 = dsp.resample(x, sr, sr_z) + result = z_denoise(x16, sr_z, model="zipenhancer", strength=float(np.clip(strength, 0.0, 1.0))) + y = result[0] if isinstance(result, tuple) else result + y = np.asarray(y, dtype=np.float32) + y = dsp.resample(y, sr_z, sr) + y = blend(x, y, strength) + return y, "zipenhancer (16 kHz SOTA, bandwidth restored)", device_pref diff --git a/lib/src/producer/engines/enhance_mossformer.py b/lib/src/producer/engines/enhance_mossformer.py new file mode 100644 index 0000000..f881045 --- /dev/null +++ b/lib/src/producer/engines/enhance_mossformer.py @@ -0,0 +1,38 @@ +from __future__ import annotations + +import numpy as np + +from .. import dsp, io, lazy +from .base import blend, device_name, pick_device + + +def enhance( + x: np.ndarray, sr: int, strength: float, device_pref: str = "auto" +) -> tuple[np.ndarray, str, str]: + lazy.ensure(["clearvoice==0.1.2"], purpose="MossFormer2 (ClearVoice)") + import tempfile + from pathlib import Path + + from clearvoice import ClearVoice + + device = pick_device(device_pref) + sr_target = 48000 + xin = dsp.resample(x, sr, sr_target) + cv = ClearVoice(task="speech_enhancement", model_name="MossFormer2_SE_48K") + with tempfile.TemporaryDirectory(prefix="producer_mf2_") as td: + inp = Path(td) / "in.wav" + outp = Path(td) / "out.wav" + io.encode(xin, sr_target, inp, "wav", 16) + result = cv(input_path=str(inp), output_name=str(outp)) + if isinstance(result, tuple): + y, fs = result[0], int(result[1]) + else: + y, fs = io.decode(outp) + y = np.asarray(y, dtype=np.float32) + if y.ndim > 1: + y = y.mean(axis=-1) if y.shape[-1] <= 2 else y.reshape(-1) + if fs != sr_target: + y = dsp.resample(y, fs, sr_target) + y = dsp.resample(y, sr_target, sr) + y = blend(x, y, strength) + return y, "mossformer2 (MossFormer2_SE_48K)", device_name(device) diff --git a/lib/src/producer/engines/enhance_resemble.py b/lib/src/producer/engines/enhance_resemble.py new file mode 100644 index 0000000..47f8a6c --- /dev/null +++ b/lib/src/producer/engines/enhance_resemble.py @@ -0,0 +1,74 @@ +from __future__ import annotations + +import os +import subprocess +import sys +from pathlib import Path + +import numpy as np + +from .. import dsp, io, lazy +from .base import blend + +VENV_DIR = lazy.DATA_DIR / "venvs" / "resemble" +WORKER = Path(__file__).resolve().parent / "resemble_worker.py" +REQ_HASH_FILE = VENV_DIR / ".req-hash" +REQS = ["resemble-enhance==0.0.1", "librosa", "soundfile"] + + +def _ensure_venv() -> Path: + marker = REQ_HASH_FILE.read_text() if REQ_HASH_FILE.exists() else None + want = "|".join(REQS) + py = VENV_DIR / "bin" / "python" + if py.is_file() and marker == want: + return py + uv = lazy.find_uv() + if not py.is_file(): + print("[producer] creating resemble venv (CPython 3.11)...", flush=True) + subprocess.run([uv, "venv", str(VENV_DIR), "--python", "3.11"], check=True) + print( + "[producer] installing resemble-enhance into isolated venv (one-time, large)...", flush=True + ) + subprocess.run( + [uv, "pip", "install", "--python", str(py), *REQS], + check=True, + ) + REQ_HASH_FILE.write_text(want) + return py + + +def enhance( + x: np.ndarray, sr: int, strength: float, device_pref: str = "auto" +) -> tuple[np.ndarray, str, str]: + import tempfile + + py = _ensure_venv() + from .base import device_name as _dname + from .base import pick_device as _pick + + device = _pick(device_pref) + with tempfile.TemporaryDirectory(prefix="producer_res_") as td: + inp = Path(td) / "in.wav" + outp = Path(td) / "out.wav" + io.encode(x, sr, inp, "wav", 16) + env = dict(os.environ) + env["HF_HOME"] = str(lazy.DATA_DIR / "models" / "hf") + env["TORCH_HOME"] = str(lazy.DATA_DIR / "models" / "torch") + cmd = [ + str(py), + str(WORKER), + str(inp), + str(outp), + device, + str(float(np.clip(strength, 0.0, 1.0))), + ] + subprocess.run(cmd, check=True, env=env) + y, fs = io.decode(outp) + if fs != sr: + y = dsp.resample(y, fs, sr) + y = blend(x, y, strength) + return y, "resemble-enhance (generative, may alter timbre)", _dname(device) + + +if __name__ == "__main__": + sys.exit(0) diff --git a/lib/src/producer/engines/resemble_worker.py b/lib/src/producer/engines/resemble_worker.py new file mode 100644 index 0000000..cdc5e2b --- /dev/null +++ b/lib/src/producer/engines/resemble_worker.py @@ -0,0 +1,37 @@ +from __future__ import annotations + +import contextlib +import sys + +import numpy as np +import soundfile as sf + + +def load(path: str) -> tuple[np.ndarray, int]: + data, sr = sf.read(path, dtype="float32", always_2d=True) + return data.mean(axis=1).astype(np.float32), int(sr) + + +def main() -> int: + import torch + from resemble_enhance.enhancer.inference import denoise as r_denoise + from resemble_enhance.enhancer.inference import enhance as r_enhance + + inp, outp, device, strength = sys.argv[1], sys.argv[2], sys.argv[3], float(sys.argv[4]) + x, sr = load(inp) + t = torch.from_numpy(x) + with contextlib.suppress(Exception): + t, sr = r_denoise(t, sr, device) + try: + y, sr = r_enhance(t, sr, device, nfe=64, solver="midpoint", lambd=1.0 - 0.1 * strength) + except TypeError: + y, sr = r_enhance(t, sr, device) + if isinstance(y, torch.Tensor): + y = y.detach().cpu().numpy() + y = np.asarray(y, dtype=np.float32).reshape(-1) + sf.write(outp, y, int(sr), subtype="FLOAT") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/lib/src/producer/io.py b/lib/src/producer/io.py new file mode 100644 index 0000000..3e9e769 --- /dev/null +++ b/lib/src/producer/io.py @@ -0,0 +1,87 @@ +from __future__ import annotations + +import shutil +import subprocess +import tempfile +from pathlib import Path + +import numpy as np +import soundfile as sf + +SUBTYPES = {16: "PCM_16", 24: "PCM_24", 32: "FLOAT"} + + +def ffmpeg_available() -> bool: + return shutil.which("ffmpeg") is not None + + +def decode(path: str | Path) -> tuple[np.ndarray, int]: + path = Path(path) + try: + data, sr = sf.read(str(path), dtype="float32", always_2d=True) + except RuntimeError: + if not ffmpeg_available(): + raise + with tempfile.TemporaryDirectory(prefix="producer_dec_") as tmpdir: + tmp = Path(tmpdir) / "dec.wav" + subprocess.run( + ["ffmpeg", "-v", "error", "-y", "-i", str(path), "-vn", "-ac", "1", str(tmp)], + check=True, + ) + data, sr = sf.read(str(tmp), dtype="float32", always_2d=True) + mono = data.mean(axis=1).astype(np.float32) + return mono, int(sr) + + +def encode( + x: np.ndarray, + sr: int, + path: str | Path, + out_format: str = "wav", + bit_depth: int = 16, +) -> Path: + path = Path(path) + fmt = out_format.lower() + if fmt in ("wav", "flac"): + subtype = SUBTYPES.get(bit_depth, "PCM_16") + sf.write(str(path), x, sr, subtype=subtype, format=fmt.upper()) + return path + if fmt == "mp3": + if not ffmpeg_available(): + raise RuntimeError("mp3 output requires ffmpeg on PATH") + with tempfile.TemporaryDirectory(prefix="producer_enc_") as td: + tmp_wav = Path(td) / "enc.wav" + sf.write(str(tmp_wav), x, sr, subtype="PCM_16") + subprocess.run( + [ + "ffmpeg", + "-v", + "error", + "-y", + "-i", + str(tmp_wav), + "-codec:a", + "libmp3lame", + "-b:a", + "192k", + str(path), + ], + check=True, + ) + return path + raise ValueError(f"unsupported output format: {out_format}") + + +def supported_globs() -> list[str]: + return ( + "*.wav", + "*.flac", + "*.mp3", + "*.m4a", + "*.aac", + "*.ogg", + "*.opus", + "*.aif", + "*.aiff", + "*.wma", + ) diff --git a/lib/src/producer/lazy.py b/lib/src/producer/lazy.py new file mode 100644 index 0000000..0ad664d --- /dev/null +++ b/lib/src/producer/lazy.py @@ -0,0 +1,85 @@ +from __future__ import annotations + +import importlib.util +import shutil +import subprocess +import sys +from importlib import metadata +from pathlib import Path + +DATA_DIR = Path(__file__).resolve().parents[2] +TORCH_GPU_INDEX = "https://download.pytorch.org/whl/cu126" +DIST_ALIASES = { + "deepfilternet": "df", + "deepfilterlib": "libdf", +} + + +class EngineUnavailable(RuntimeError): + pass + + +def find_uv() -> str: + uv = shutil.which("uv") + if uv: + return uv + local = DATA_DIR / "bin" / "uv" + if local.is_file(): + return str(local) + raise EngineUnavailable("uv not found; run the producer script to bootstrap") + + +def has_module(name: str) -> bool: + return importlib.util.find_spec(name) is not None + + +def dist_name(spec: str) -> str: + return spec.split("==")[0].split(">=")[0].split("<")[0].split("[")[0] + + +def is_installed(spec: str) -> bool: + dist = dist_name(spec) + try: + metadata.distribution(dist) + return True + except metadata.PackageNotFoundError: + return has_module(DIST_ALIASES.get(dist, dist)) + + +def gpu_present() -> bool: + return shutil.which("nvidia-smi") is not None + + +def ensure_torch() -> None: + if has_module("torch"): + return + cmd = [find_uv(), "pip", "install", "--python", sys.executable] + if gpu_present(): + cmd += ["torch==2.7.1+cu126", "torchaudio==2.7.1+cu126", "--index-url", TORCH_GPU_INDEX] + else: + cmd += ["torch==2.7.1", "torchaudio==2.7.1"] + print("[producer] installing torch (one-time, large download)...", flush=True) + subprocess.run(cmd, check=True) + import importlib + + importlib.invalidate_caches() + + +def ensure(packages: list[str], purpose: str) -> None: + missing = [p for p in packages if not is_installed(p)] + if not missing: + return + if any(m.startswith("torch") for m in missing): + ensure_torch() + missing = [m for m in missing if not m.startswith("torch") and not is_installed(m)] + if not missing: + return + print(f"[producer] installing {purpose} dependencies (one-time)...", flush=True) + cmd = [find_uv(), "pip", "install", "--python", sys.executable, *missing] + subprocess.run(cmd, check=True) + import importlib + + importlib.invalidate_caches() + for spec in missing: + if not is_installed(spec): + raise EngineUnavailable(f"failed to install {dist_name(spec)} for {purpose}") diff --git a/lib/src/producer/loudness.py b/lib/src/producer/loudness.py new file mode 100644 index 0000000..8ceae97 --- /dev/null +++ b/lib/src/producer/loudness.py @@ -0,0 +1,30 @@ +from __future__ import annotations + +import numpy as np + +from . import dsp +from .meters import meter, true_peak_db + + +def normalize( + x: np.ndarray, + sr: int, + mode: str, + target: float, + ceiling_db: float, + max_iters: int = 4, +) -> np.ndarray: + if not np.any(x): + return x + y = x + for _ in range(max_iters): + cur = meter(y, sr, mode) + gain = target - cur + if abs(gain) < 0.05: + break + y = y * (10.0 ** (gain / 20.0)) + y = dsp.limit(y, sr, ceiling_db) + tp = true_peak_db(y, sr) + if tp > ceiling_db + 0.05: + y = y * (10.0 ** ((ceiling_db - tp) / 20.0)) + return y.astype(np.float32) diff --git a/lib/src/producer/meters.py b/lib/src/producer/meters.py new file mode 100644 index 0000000..0cfc3e3 --- /dev/null +++ b/lib/src/producer/meters.py @@ -0,0 +1,75 @@ +from __future__ import annotations + +import numpy as np +from scipy import signal + +SILENCE_DB = -120.0 + + +def _db(v: float) -> float: + if v <= 0.0 or not np.isfinite(v): + return SILENCE_DB + return max(20.0 * np.log10(v), SILENCE_DB) + + +def rms_db(x: np.ndarray) -> float: + x64 = x.astype(np.float64) + return _db(float(np.sqrt(np.mean(np.square(x64))))) + + +def sample_peak_db(x: np.ndarray) -> float: + x64 = np.abs(x.astype(np.float64)) + return _db(float(np.max(x64)) if x64.size else 0.0) + + +def true_peak_db(x: np.ndarray, sr: int, oversample: int = 4) -> float: + if x.size < 2: + return sample_peak_db(x) + y = signal.resample_poly(x.astype(np.float64), oversample, 1, window=("kaiser", 8.0)) + return sample_peak_db(y.astype(np.float32)) + + +def noise_floor_db(x: np.ndarray, sr: int, block_ms: float = 50.0, pct: float = 5.0) -> float: + n = max(1, int(sr * block_ms / 1000.0)) + nb = x.size // n + if nb < 1: + return sample_peak_db(x) if x.size else SILENCE_DB + blocks = x[: nb * n].reshape(nb, n).astype(np.float64) + block_rms = np.sqrt(np.mean(np.square(blocks), axis=1)) + active = block_rms[block_rms > 0.0] + if active.size == 0: + return SILENCE_DB + return _db(float(np.percentile(active, pct))) + + +def lufs(x: np.ndarray, sr: int) -> float: + import pyloudnorm as pyln + + x64 = x.astype(np.float64) + if not np.any(x64): + return SILENCE_DB + if x64.size < int(sr * 0.2): + return SILENCE_DB + meter = pyln.Meter(sr) + try: + val = meter.integrated_loudness(x64) + except ValueError: + return SILENCE_DB + if not np.isfinite(val): + return SILENCE_DB + return float(val) + + +def meter(x: np.ndarray, sr: int, mode: str) -> float: + return lufs(x, sr) if mode == "lufs" else rms_db(x) + + +def all_meters(x: np.ndarray, sr: int) -> dict[str, float]: + return { + "rms_db": rms_db(x), + "sample_peak_db": sample_peak_db(x), + "true_peak_db": true_peak_db(x, sr), + "lufs": lufs(x, sr), + "noise_floor_db": noise_floor_db(x, sr), + "duration_s": round(x.size / sr, 3) if sr else 0.0, + } diff --git a/lib/src/producer/pipeline.py b/lib/src/producer/pipeline.py new file mode 100644 index 0000000..0a22bf8 --- /dev/null +++ b/lib/src/producer/pipeline.py @@ -0,0 +1,169 @@ +from __future__ import annotations + +import time +from collections.abc import Callable +from dataclasses import dataclass, field + +import numpy as np + +from . import dsp, loudness +from .config import Options +from .meters import all_meters + + +@dataclass +class Stage: + name: str + detail: str + enabled: bool = True + fn: Callable[[np.ndarray, int], np.ndarray] | None = None + + +@dataclass +class RunResult: + audio: np.ndarray + sr: int + stages: list[Stage] + timings: dict[str, float] = field(default_factory=dict) + notes: list[str] = field(default_factory=list) + before: dict[str, float] = field(default_factory=dict) + after: dict[str, float] = field(default_factory=dict) + + +def _denoise_fn(opts: Options, notes: list[str]): + engine = opts.denoise + + def fn(x: np.ndarray, sr: int) -> np.ndarray: + if engine == "off": + return x + if engine == "dfn3": + from .engines import denoise_dfn + + y, eng, dev = denoise_dfn.denoise(x, sr, opts.denoise_strength, opts.device) + elif engine == "zipenhancer": + from .engines import denoise_zip + + y, eng, dev = denoise_zip.denoise(x, sr, opts.denoise_strength, opts.device) + else: + raise ValueError(f"unknown denoise engine: {engine}") + notes.append(f"denoise: {eng} on {dev}") + return y + + return fn + + +def _enhance_fn(opts: Options, notes: list[str]): + engine = opts.enhance + + def fn(x: np.ndarray, sr: int) -> np.ndarray: + if engine == "off": + return x + if engine == "mossformer2": + from .engines import enhance_mossformer + + y, eng, dev = enhance_mossformer.enhance(x, sr, opts.enhance_strength, opts.device) + elif engine == "resemble": + from .engines import enhance_resemble + + y, eng, dev = enhance_resemble.enhance(x, sr, opts.enhance_strength, opts.device) + else: + raise ValueError(f"unknown enhance engine: {engine}") + notes.append(f"enhance: {eng} on {dev}") + return y + + return fn + + +def _dsp_fn(opts: Options, notes: list[str]): + p = opts.prof() + + def fn(x: np.ndarray, sr: int) -> np.ndarray: + hpf_hz = opts.hpf_hz if opts.hpf_hz is not None else p.hpf_hz + if opts.eff("hpf") > 0.001: + x = dsp.hpf(x, sr, hpf_hz) + f, g, q = p.mud + x = dsp.peak_eq(x, sr, f, g * opts.eff("mud"), q) + wf, wg = p.warmth + x = dsp.shelf(x, sr, wf, wg * opts.eff("warmth"), low=True) + x = dsp.soothe(x, sr, opts.eff("soothe")) + c1 = p.comp1 + y = dsp.compressor(x, sr, c1[0], c1[1], c1[2], c1[3]) + x = _blend(x, y, opts.eff("compress") * 0.7) + c2 = p.comp2 + y = dsp.compressor(x, sr, c2[0], c2[1], c2[2], c2[3]) + x = _blend(x, y, opts.eff("compress") * 0.5) + x = dsp.tape(x, sr, opts.eff("tape")) + d = p.deess + x = dsp.deesser(x, sr, d[0], d[1], d[2] * opts.eff("deess")) + pf, pg = p.presence + x = dsp.peak_eq(x, sr, pf, pg * opts.eff("presence")) + af, ag = p.air + x = dsp.shelf(x, sr, af, ag * opts.eff("air"), low=False) + x = dsp.expander(x, sr, max_drop_db=6.0 * opts.eff("breath")) + return x + + return fn + + +def _blend(x: np.ndarray, y: np.ndarray, s: float) -> np.ndarray: + s = float(np.clip(s, 0.0, 1.0)) + if s >= 0.999: + return y + return (x.astype(np.float64) * (1.0 - s) + y.astype(np.float64) * s).astype(np.float32) + + +def _level_fn(opts: Options, notes: list[str]): + def fn(x: np.ndarray, sr: int) -> np.ndarray: + return loudness.normalize(x, sr, opts.loudness_mode(), opts.target_value(), opts.ceiling()) + + return fn + + +def build_stages(opts: Options, notes: list[str] | None = None) -> list[Stage]: + notes = notes if notes is not None else [] + stages: list[Stage] = [] + stages.append( + Stage( + "denoise", + f"engine={opts.denoise} strength={opts.denoise_strength:.2f}", + opts.denoise != "off", + _denoise_fn(opts, notes), + ) + ) + stages.append( + Stage( + "enhance", + f"engine={opts.enhance} strength={opts.enhance_strength:.2f}", + opts.enhance != "off", + _enhance_fn(opts, notes), + ) + ) + dsp_desc = "hpf→mud→warmth→soothe→comp2x→tape→deess→presence→air→breath" + stages.append(Stage("dsp", dsp_desc, opts.dsp, _dsp_fn(opts, notes) if opts.dsp else None)) + stages.append( + Stage( + "levelling", + f"mode={opts.loudness_mode()} target={opts.target_value()} ceiling={opts.ceiling()}", + opts.levelling, + _level_fn(opts, notes) if opts.levelling else None, + ) + ) + return stages + + +def run_pipeline(x: np.ndarray, sr: int, opts: Options) -> RunResult: + notes: list[str] = [] + stages = build_stages(opts, notes) + before = all_meters(x, sr) + y = x + timings: dict[str, float] = {} + for st in stages: + if not st.enabled or st.fn is None: + continue + t0 = time.perf_counter() + y = st.fn(y, sr) + timings[st.name] = round(time.perf_counter() - t0, 3) + after = all_meters(y, sr) + return RunResult( + audio=y, sr=sr, stages=stages, timings=timings, notes=notes, before=before, after=after + ) diff --git a/lib/src/producer/report.py b/lib/src/producer/report.py new file mode 100644 index 0000000..27fbb8f --- /dev/null +++ b/lib/src/producer/report.py @@ -0,0 +1,56 @@ +from __future__ import annotations + +import json +from pathlib import Path + +from . import __version__ +from .config import Options + + +def build( + input_path: str, + out_path: str, + opts: Options, + before: dict[str, float], + after: dict[str, float], + timings: dict[str, float], + notes: list[str], +) -> dict: + return { + "tool": "producer", + "version": __version__, + "input": str(input_path), + "output": str(out_path), + "settings": opts.to_dict(), + "before": before, + "after": after, + "stage_seconds": timings, + "engine_notes": notes, + } + + +def print_human(rep: dict) -> None: + b = rep["before"] + a = rep["after"] + print(" before -> after:") + for key, label in ( + ("rms_db", "RMS"), + ("true_peak_db", "true peak"), + ("lufs", "LUFS"), + ("noise_floor_db", "noise floor"), + ): + print(f" {label:<12} {b[key]:>8.1f} dB -> {a[key]:>8.1f} dB") + if rep["engine_notes"]: + for note in rep["engine_notes"]: + print(f" {note}") + times = rep["stage_seconds"] + if times: + total = sum(times.values()) + detail = ", ".join(f"{k} {v:.1f}s" for k, v in times.items()) + print(f" stages: {detail} (total {total:.1f}s)") + + +def save(rep: dict, path: str | Path) -> Path: + path = Path(path) + path.write_text(json.dumps(rep, indent=2)) + return path -- cgit v1.2.3