Compare commits
1
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
1276d95708 |
@@ -2,6 +2,14 @@
|
||||
|
||||
All notable changes to ComfyUI-AusBoss are documented here.
|
||||
|
||||
## Unreleased
|
||||
|
||||
- **Load Audio 🆎: Clean up.** Turn on **Clean up** for a voice recording
|
||||
and it comes out as one channel, with the quiet bits at the start and end
|
||||
cut, less hum and hiss, and an even volume that never clips. A video model
|
||||
that copies the sound it is given copies its noise too, so this helps a
|
||||
phone recording. It is off by default, so saved workflows are unchanged.
|
||||
|
||||
## 2.25.0 - 2026-11-12
|
||||
|
||||
- **New node: Load Image.** Loads a picture and sets its size in the same
|
||||
|
||||
@@ -27,6 +27,22 @@ Max length can be wired. When the wire comes from a number node such as
|
||||
Float 🆎, the wave shows the cut for that number; otherwise the length's tooltip says
|
||||
`max wired` and the run still cuts the sound to the wired length.
|
||||
|
||||
## Clean up
|
||||
|
||||
Turn **Clean up** on for a voice recording you want a video model to say,
|
||||
such as a line recorded on a phone. The sound that comes out:
|
||||
|
||||
- is one channel,
|
||||
- has the quiet bits at the start and end cut off, so the voice starts right
|
||||
away,
|
||||
- has less hum, rumble and hiss,
|
||||
- is at an even volume, never so loud that it clips.
|
||||
|
||||
A model that copies the sound it is given copies its noise too, so a clean
|
||||
recording gives a cleaner result. Clean up works on the part you picked, so
|
||||
the sound can come out a little shorter than the length shown under the wave;
|
||||
`seconds` always gives the real length. Leave it off for music.
|
||||
|
||||
## Controls
|
||||
|
||||
- **Source** (`audio`): a sound file in ComfyUI's input folder. **Upload**
|
||||
@@ -36,6 +52,8 @@ Float 🆎, the wave shows the cut for that number; otherwise the length's toolt
|
||||
- **IN** (`start_seconds`): where the sound starts.
|
||||
- **OUT** (`end_seconds`): where the sound stops. `0` means the end of the
|
||||
file.
|
||||
- **Clean up** (`clean_up`): on cleans up a voice recording, as above. Off
|
||||
hands on the sound as it is.
|
||||
|
||||
## Outputs
|
||||
|
||||
@@ -49,4 +67,11 @@ Float 🆎, the wave shows the cut for that number; otherwise the length's toolt
|
||||
- The sound keeps its own sample rate and channels.
|
||||
- A start past the end of the file stops the run with a message that says so.
|
||||
- The wave is drawn in the browser from the file itself; nothing extra is
|
||||
stored with the workflow besides the four values above.
|
||||
stored with the workflow besides the five values above.
|
||||
- Clean up runs in this order: the channels are averaged into one; a
|
||||
high-pass at 80 Hz takes out rumble and mains hum; a noise print is
|
||||
taken from the quietest moments and whatever sits close to it is turned
|
||||
down by up to 12 dB (hiss, fans, room tone); quiet ends are cut, keeping
|
||||
0.1 s before the first sound and 0.2 s after the last; then the volume is
|
||||
set to -18 LUFS (the loudness broadcast meters read) with a limiter that
|
||||
keeps every peak under -1 dBFS. The sample rate stays the file's own.
|
||||
|
||||
@@ -431,11 +431,15 @@ const CARDS = {
|
||||
AUSBOSS_NODES_LoadAudio: {
|
||||
// The wave, IN and OUT live in the panel under the card (js/load_audio).
|
||||
minWidth: 280, first: true, hide: ["upload"],
|
||||
// Older saves hand clean_up the upload button's empty value.
|
||||
resetUnknown: { clean_up: false },
|
||||
rows: [
|
||||
{ widget: "audio", label: "Source", kind: "select",
|
||||
button: { text: PANEL_LABELS.load, title: "Upload a sound file into ComfyUI's input folder", onClick: (node) => node.widgets?.find((w) => w.name === "upload")?.callback?.() } },
|
||||
{ widget: "max_seconds", label: "Max length", suffix: "s", decimals: 1, trimZeros: true,
|
||||
title: "The longest the sound may be. Anything past it is cut off and shows in red. 0 means no limit." },
|
||||
{ widget: "clean_up", label: "Clean up", onText: PANEL_LABELS.on, offText: PANEL_LABELS.off,
|
||||
title: "For a voice recording. On cuts the quiet bits at the start and end, takes out hum and hiss, and evens out the volume. Off hands on the sound as it is." },
|
||||
],
|
||||
},
|
||||
// Load Image: the file, then where it is in the folder and the size it
|
||||
|
||||
@@ -0,0 +1,212 @@
|
||||
"""Voice cleanup for Load Audio's clean_up switch.
|
||||
|
||||
One pass, in this order, on the part of the file the node hands on:
|
||||
|
||||
1. one channel (the channels are averaged);
|
||||
2. a high-pass at 80 Hz for rumble and mains hum (a man's voice starts
|
||||
around 85 Hz, so it keeps its depth);
|
||||
3. light hiss reduction: a noise print from the quietest moments, and every
|
||||
sound close to it is turned down by at most ``HISS_REDUCTION_DB``;
|
||||
4. quiet stretches at the start and the end are cut, keeping a short pad;
|
||||
5. an even loudness (``TARGET_LUFS``, measured the way broadcast meters do),
|
||||
with a limiter that keeps every peak under ``PEAK_CEILING_DB``.
|
||||
|
||||
Everything is torch on the CPU, at the file's own sample rate.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import math
|
||||
|
||||
import torch
|
||||
import torch.nn.functional as F
|
||||
|
||||
from ._execution_helpers import raise_if_interrupted
|
||||
|
||||
HIGH_PASS_HZ = 80.0
|
||||
HISS_REDUCTION_DB = 12.0
|
||||
# A sound counts as hiss below this many dB over the noise print.
|
||||
GATE_MARGIN_DB = 6.0
|
||||
TARGET_LUFS = -18.0
|
||||
PEAK_CEILING_DB = -1.0
|
||||
MAX_BOOST_DB = 30.0
|
||||
# Quiet ends: what stays before the first and after the last sound.
|
||||
LEAD_PAD_SECONDS = 0.1
|
||||
TAIL_PAD_SECONDS = 0.2
|
||||
# Quieter than this everywhere is treated as silence and left alone.
|
||||
SILENCE_DB = -60.0
|
||||
|
||||
_EPS = 1e-10
|
||||
|
||||
|
||||
def _db(value: torch.Tensor) -> torch.Tensor:
|
||||
return 20.0 * torch.log10(value.clamp_min(_EPS))
|
||||
|
||||
|
||||
def _shape_spectrum(signal: torch.Tensor, rate: int, gain_for_hz) -> torch.Tensor:
|
||||
"""Multiply the whole signal's spectrum by ``gain_for_hz(frequencies)``.
|
||||
|
||||
Zero phase. The signal is padded first so the filter's ring at one end
|
||||
does not wrap around into the other.
|
||||
"""
|
||||
length = signal.shape[-1]
|
||||
pad = int(rate * 0.1)
|
||||
padded = F.pad(signal, (pad, pad))
|
||||
spectrum = torch.fft.rfft(padded.double())
|
||||
hz = torch.fft.rfftfreq(padded.shape[-1], d=1.0 / rate, dtype=torch.float64)
|
||||
shaped = torch.fft.irfft(spectrum * gain_for_hz(hz), n=padded.shape[-1])
|
||||
return shaped[..., pad : pad + length].to(signal.dtype)
|
||||
|
||||
|
||||
def _high_pass_gain(hz: torch.Tensor) -> torch.Tensor:
|
||||
# A fourth-order Butterworth's magnitude: -3 dB at the corner, -10 dB at
|
||||
# 60 Hz hum, -16 dB at 50 Hz, -24 dB an octave under. Zero at DC.
|
||||
ratio = (HIGH_PASS_HZ / hz.clamp_min(1e-6)) ** 8
|
||||
return torch.where(hz > 0, 1.0 / torch.sqrt(1.0 + ratio), torch.zeros_like(hz))
|
||||
|
||||
|
||||
def _biquad_magnitude(b, a, hz: torch.Tensor, rate: int) -> torch.Tensor:
|
||||
z = torch.exp(-1j * 2.0 * math.pi * hz / rate)
|
||||
top = b[0] + b[1] * z + b[2] * z * z
|
||||
bottom = a[0] + a[1] * z + a[2] * z * z
|
||||
return (top / bottom).abs()
|
||||
|
||||
|
||||
def _k_weighting_gain(rate: int):
|
||||
"""The loudness meter's ear curve (ITU-R BS.1770) at any sample rate."""
|
||||
# Stage 1: a high shelf, +4 dB above about 1.7 kHz.
|
||||
f0, gain_db, q = 1681.974450955533, 3.999843853973347, 0.7071752369554196
|
||||
k = math.tan(math.pi * f0 / rate)
|
||||
vh = 10.0 ** (gain_db / 20.0)
|
||||
vb = vh ** 0.4996667741545416
|
||||
a0 = 1.0 + k / q + k * k
|
||||
shelf_b = ((vh + vb * k / q + k * k) / a0, 2.0 * (k * k - vh) / a0, (vh - vb * k / q + k * k) / a0)
|
||||
shelf_a = (1.0, 2.0 * (k * k - 1.0) / a0, (1.0 - k / q + k * k) / a0)
|
||||
# Stage 2: a high-pass at about 38 Hz.
|
||||
f0, q = 38.13547087602444, 0.5003270373238773
|
||||
k = math.tan(math.pi * f0 / rate)
|
||||
a0 = 1.0 + k / q + k * k
|
||||
pass_b = (1.0, -2.0, 1.0)
|
||||
pass_a = (1.0, 2.0 * (k * k - 1.0) / a0, (1.0 - k / q + k * k) / a0)
|
||||
|
||||
def gain(hz: torch.Tensor) -> torch.Tensor:
|
||||
return _biquad_magnitude(shelf_b, shelf_a, hz, rate) * _biquad_magnitude(pass_b, pass_a, hz, rate)
|
||||
|
||||
return gain
|
||||
|
||||
|
||||
def integrated_loudness(mono: torch.Tensor, rate: int) -> float:
|
||||
"""Integrated loudness in LUFS of a 1-D signal, or -inf for silence.
|
||||
|
||||
400 ms blocks every 100 ms, gated at -70 LUFS and then 10 LU under the
|
||||
mean, as broadcast meters measure it.
|
||||
"""
|
||||
weighted = _shape_spectrum(mono, rate, _k_weighting_gain(rate)).double()
|
||||
block = int(round(0.4 * rate))
|
||||
hop = int(round(0.1 * rate))
|
||||
if weighted.shape[-1] < block:
|
||||
blocks = weighted.square().mean().reshape(1)
|
||||
else:
|
||||
blocks = weighted.unfold(-1, block, hop).square().mean(dim=-1)
|
||||
|
||||
def lufs(power: torch.Tensor) -> torch.Tensor:
|
||||
return -0.691 + 10.0 * torch.log10(power.clamp_min(1e-20))
|
||||
|
||||
kept = blocks[lufs(blocks) > -70.0]
|
||||
if kept.numel() == 0:
|
||||
return float("-inf")
|
||||
relative = lufs(kept.mean()) - 10.0
|
||||
kept = kept[lufs(kept) > relative]
|
||||
return float(lufs(kept.mean()))
|
||||
|
||||
|
||||
def _fft_size(rate: int) -> int:
|
||||
# About 40 ms: fine enough to tell a voice's harmonics from hiss.
|
||||
return max(256, 1 << int(round(math.log2(rate * 0.04))))
|
||||
|
||||
|
||||
def _reduce_hiss(mono: torch.Tensor, rate: int) -> torch.Tensor:
|
||||
"""Spectral gate: a noise print from the quietest frames, then a soft,
|
||||
smoothed gain that turns down whatever sits close to it."""
|
||||
n_fft = _fft_size(rate)
|
||||
hop = n_fft // 4
|
||||
if mono.shape[-1] < n_fft * 4:
|
||||
return mono # too short to learn what the noise is
|
||||
window = torch.hann_window(n_fft, dtype=mono.dtype)
|
||||
spec = torch.stft(mono, n_fft, hop, window=window, center=True, return_complex=True)
|
||||
level = _db(spec.abs()) # bins x frames
|
||||
# The quietest tenth of the frames, by their total energy, is the print.
|
||||
energy = spec.abs().square().sum(dim=0)
|
||||
count = max(4, energy.shape[0] // 10)
|
||||
quiet = torch.topk(energy, count, largest=False).indices
|
||||
noise = level[:, quiet]
|
||||
threshold = noise.mean(dim=1, keepdim=True) + noise.std(dim=1, keepdim=True) + GATE_MARGIN_DB
|
||||
# 0 for hiss, 1 for anything well over it, a 3 dB soft edge between.
|
||||
keep = torch.sigmoid((level - threshold) / 1.5)
|
||||
# Smooth across nearby bins and frames so the gain does not twinkle.
|
||||
kernel = torch.outer(torch.hann_window(7, periodic=False), torch.hann_window(5, periodic=False))
|
||||
kernel = (kernel / kernel.sum()).to(keep.dtype)[None, None]
|
||||
keep = F.conv2d(keep[None, None], kernel, padding=(3, 2))[0, 0]
|
||||
floor = 10.0 ** (-HISS_REDUCTION_DB / 20.0)
|
||||
gain = floor + (1.0 - floor) * keep
|
||||
return torch.istft(spec * gain, n_fft, hop, window=window, center=True, length=mono.shape[-1])
|
||||
|
||||
|
||||
def _sound_span(mono: torch.Tensor, rate: int) -> tuple[int, int]:
|
||||
"""First and last sample worth keeping, quiet ends cut, pads kept."""
|
||||
frame = max(1, int(rate * 0.02))
|
||||
usable = (mono.shape[-1] // frame) * frame
|
||||
if usable == 0:
|
||||
return 0, mono.shape[-1]
|
||||
levels = _db(mono[:usable].reshape(-1, frame).square().mean(dim=1).sqrt())
|
||||
loudest = float(levels.max())
|
||||
if loudest < SILENCE_DB:
|
||||
return 0, mono.shape[-1]
|
||||
floor = float(torch.quantile(levels, 0.1))
|
||||
# 10 dB over the quiet floor, or halfway to the loudest moment when the
|
||||
# noise sits closer than that, and never more than 50 dB under it.
|
||||
threshold = max(loudest - 50.0, min(floor + 10.0, (floor + loudest) / 2.0))
|
||||
loud = torch.nonzero(levels > threshold).flatten()
|
||||
first = max(0, int(loud[0]) * frame - int(rate * LEAD_PAD_SECONDS))
|
||||
last = min(mono.shape[-1], (int(loud[-1]) + 1) * frame + int(rate * TAIL_PAD_SECONDS))
|
||||
return first, last
|
||||
|
||||
|
||||
def _limit(mono: torch.Tensor, rate: int, ceiling: float) -> torch.Tensor:
|
||||
"""Look-ahead peak limiter: the gain dips just before a peak and comes
|
||||
back after it, so nothing passes the ceiling and nothing is clipped."""
|
||||
needed = (ceiling / mono.abs().clamp_min(_EPS)).clamp(max=1.0)
|
||||
if float(needed.min()) >= 1.0:
|
||||
return mono
|
||||
reach = max(1, int(rate * 0.005))
|
||||
width = 2 * reach + 1
|
||||
# The lowest gain any sample within reach needs, then averaged over the
|
||||
# same reach: every average is at or under what its own sample needs.
|
||||
lowest = -F.max_pool1d(-needed[None, None], width, stride=1, padding=reach)
|
||||
smooth = F.avg_pool1d(lowest, width, stride=1, padding=reach, count_include_pad=False)[0, 0]
|
||||
return (mono * smooth).clamp(-ceiling, ceiling)
|
||||
|
||||
|
||||
def clean_up_voice(sound: dict) -> dict:
|
||||
"""ComfyUI AUDIO in, the cleaned-up voice out as one channel."""
|
||||
waveform = sound["waveform"]
|
||||
rate = int(sound["sample_rate"])
|
||||
mono = waveform[0].float().mean(dim=0)
|
||||
if mono.numel() == 0:
|
||||
return {"waveform": mono.reshape(1, 1, 0), "sample_rate": rate}
|
||||
raise_if_interrupted()
|
||||
mono = _shape_spectrum(mono, rate, _high_pass_gain)
|
||||
raise_if_interrupted()
|
||||
mono = _reduce_hiss(mono, rate)
|
||||
raise_if_interrupted()
|
||||
first, last = _sound_span(mono, rate)
|
||||
mono = mono[first:last]
|
||||
loudness = integrated_loudness(mono, rate)
|
||||
if math.isfinite(loudness):
|
||||
gain_db = min(TARGET_LUFS - loudness, MAX_BOOST_DB)
|
||||
mono = mono * (10.0 ** (gain_db / 20.0))
|
||||
mono = _limit(mono, rate, 10.0 ** (PEAK_CEILING_DB / 20.0))
|
||||
return {"waveform": mono.contiguous().reshape(1, 1, -1), "sample_rate": rate}
|
||||
|
||||
|
||||
__all__ = ["clean_up_voice", "integrated_loudness"]
|
||||
@@ -4,6 +4,7 @@ from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
|
||||
from ._audio_cleanup_helpers import clean_up_voice
|
||||
from ._audio_trim_helpers import decode_audio_window, list_input_audio
|
||||
from ._execution_helpers import ignore_extra_inputs
|
||||
from ._media_helpers import resolve_input_path
|
||||
@@ -17,7 +18,8 @@ class AusBossLoadAudio:
|
||||
"handles to set where it starts and stops, and press play to hear just that "
|
||||
"part. Set Max length to the longest sound the next step takes: anything "
|
||||
"longer shows in red as cut off, so you see exactly what comes out before "
|
||||
"you run anything."
|
||||
"you run anything. Turn on Clean up for a voice recording: it comes out as one "
|
||||
"channel, with the quiet ends cut, less hum and hiss, and an even volume."
|
||||
)
|
||||
SEARCH_ALIASES = [
|
||||
"load audio",
|
||||
@@ -28,6 +30,7 @@ class AusBossLoadAudio:
|
||||
"voice",
|
||||
"music",
|
||||
"waveform",
|
||||
"clean up voice",
|
||||
"ausboss",
|
||||
]
|
||||
|
||||
@@ -77,6 +80,17 @@ class AusBossLoadAudio:
|
||||
),
|
||||
},
|
||||
),
|
||||
"clean_up": (
|
||||
"BOOLEAN",
|
||||
{
|
||||
"default": False,
|
||||
"tooltip": (
|
||||
"For a voice recording. On cuts the quiet bits at the start and end, "
|
||||
"takes out hum and hiss, and evens out the volume. Off hands on the "
|
||||
"sound as it is."
|
||||
),
|
||||
},
|
||||
),
|
||||
},
|
||||
}
|
||||
|
||||
@@ -88,7 +102,7 @@ class AusBossLoadAudio:
|
||||
)
|
||||
FUNCTION = "load_audio"
|
||||
|
||||
async def load_audio(self, audio, start_seconds, end_seconds, max_seconds=0.0):
|
||||
async def load_audio(self, audio, start_seconds, end_seconds, max_seconds=0.0, clean_up=False):
|
||||
path = resolve_input_path(audio)
|
||||
# Decoding blocks for as long as the pick is; a worker thread keeps
|
||||
# ComfyUI's event loop answering, and carries the context ComfyUI
|
||||
@@ -96,6 +110,9 @@ class AusBossLoadAudio:
|
||||
sound = await asyncio.to_thread(
|
||||
decode_audio_window, path, start_seconds, end_seconds, max_seconds
|
||||
)
|
||||
if clean_up is True:
|
||||
# After the trim, so seconds is the length of what is handed on.
|
||||
sound = await asyncio.to_thread(clean_up_voice, sound)
|
||||
seconds = sound["waveform"].shape[-1] / float(sound["sample_rate"])
|
||||
return (sound, float(seconds))
|
||||
|
||||
|
||||
Vendored
+2
-1
@@ -570,7 +570,8 @@
|
||||
"end_seconds"
|
||||
],
|
||||
"optional": [
|
||||
"max_seconds"
|
||||
"max_seconds",
|
||||
"clean_up"
|
||||
],
|
||||
"outputs": [
|
||||
[
|
||||
|
||||
@@ -0,0 +1,169 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import math
|
||||
import sys
|
||||
import tempfile
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
from unittest import mock
|
||||
|
||||
import av
|
||||
import numpy as np
|
||||
import torch
|
||||
|
||||
ROOT = Path(__file__).resolve().parent.parent
|
||||
sys.path.insert(0, str(ROOT))
|
||||
if "nodes" in sys.modules and not hasattr(sys.modules["nodes"], "__path__"):
|
||||
del sys.modules["nodes"]
|
||||
|
||||
from nodes._audio_cleanup_helpers import (
|
||||
PEAK_CEILING_DB,
|
||||
TARGET_LUFS,
|
||||
clean_up_voice,
|
||||
integrated_loudness,
|
||||
)
|
||||
|
||||
RATE = 16000
|
||||
|
||||
|
||||
def seconds(n: float) -> int:
|
||||
return int(n * RATE)
|
||||
|
||||
|
||||
def voice_like(length: float, rng: np.random.Generator) -> np.ndarray:
|
||||
"""A buzzy vowel at 150 Hz with a syllable rhythm: something with the
|
||||
harmonics and the stops a voice has."""
|
||||
t = np.arange(seconds(length)) / RATE
|
||||
tone = sum(np.sin(2 * np.pi * 150 * k * t) / k for k in range(1, 20))
|
||||
rhythm = np.clip(np.sin(2 * np.pi * 3.0 * t), 0, None) ** 0.5
|
||||
return (0.2 * tone * rhythm).astype(np.float32)
|
||||
|
||||
|
||||
def sound(*channels: np.ndarray, rate: int = RATE) -> dict:
|
||||
data = np.stack(channels).astype(np.float32)
|
||||
return {"waveform": torch.from_numpy(data).unsqueeze(0), "sample_rate": rate}
|
||||
|
||||
|
||||
def band_level(signal: np.ndarray, low: float, high: float) -> float:
|
||||
spectrum = np.abs(np.fft.rfft(signal)) ** 2
|
||||
hz = np.fft.rfftfreq(signal.size, 1 / RATE)
|
||||
return 10 * math.log10(spectrum[(hz >= low) & (hz < high)].sum() + 1e-20)
|
||||
|
||||
|
||||
class CleanUpTests(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.rng = np.random.default_rng(7)
|
||||
|
||||
def test_comes_out_as_one_channel_at_the_same_rate(self):
|
||||
voice = voice_like(2.0, self.rng)
|
||||
out = clean_up_voice(sound(voice, voice * 0.5, rate=RATE))
|
||||
self.assertEqual(out["sample_rate"], RATE)
|
||||
self.assertEqual(out["waveform"].shape[:2], (1, 1))
|
||||
|
||||
def test_cuts_quiet_ends_and_keeps_a_short_pad(self):
|
||||
quiet = np.zeros(seconds(1.5), dtype=np.float32)
|
||||
voice = voice_like(2.0, self.rng)
|
||||
noisy = np.concatenate([quiet, voice, quiet, quiet])
|
||||
noisy = noisy + self.rng.normal(0, 1e-4, noisy.size).astype(np.float32)
|
||||
out = clean_up_voice(sound(noisy))["waveform"][0, 0]
|
||||
length = out.shape[-1] / RATE
|
||||
self.assertLess(length, 2.6)
|
||||
self.assertGreater(length, 1.9)
|
||||
|
||||
def test_level_is_even_and_never_clips(self):
|
||||
ceiling = 10 ** (PEAK_CEILING_DB / 20)
|
||||
for scale in (0.05, 0.2, 4.0): # very quiet, quiet, clipping hot
|
||||
out = clean_up_voice(sound(voice_like(3.0, self.rng) * scale))["waveform"][0, 0]
|
||||
self.assertLessEqual(float(out.abs().max()), ceiling + 1e-6)
|
||||
self.assertAlmostEqual(integrated_loudness(out, RATE), TARGET_LUFS, delta=1.5)
|
||||
|
||||
def test_takes_out_rumble_and_hum(self):
|
||||
voice = voice_like(3.0, self.rng)
|
||||
t = np.arange(voice.size) / RATE
|
||||
hum = (0.05 * np.sin(2 * np.pi * 30 * t) + 0.05 * np.sin(2 * np.pi * 50 * t)).astype(np.float32)
|
||||
before = voice + hum
|
||||
out = clean_up_voice(sound(before))["waveform"][0, 0].numpy()
|
||||
# Under 60 Hz against the voice's own 150 Hz and up, before and after.
|
||||
gap_before = band_level(before, 140, 4000) - band_level(before, 20, 60)
|
||||
gap_after = band_level(out, 140, 4000) - band_level(out, 20, 60)
|
||||
self.assertGreater(gap_after - gap_before, 12)
|
||||
|
||||
def test_turns_down_hiss_between_words(self):
|
||||
voice = voice_like(4.0, self.rng)
|
||||
hiss = self.rng.normal(0, 0.01, voice.size).astype(np.float32)
|
||||
before = voice + hiss
|
||||
out = clean_up_voice(sound(before))["waveform"][0, 0].numpy()
|
||||
|
||||
def floor_gap(signal: np.ndarray) -> float:
|
||||
frames = signal[: signal.size // 320 * 320].reshape(-1, 320)
|
||||
levels = 10 * np.log10((frames**2).mean(axis=1) + 1e-20)
|
||||
return float(levels.max() - np.quantile(levels, 0.1))
|
||||
|
||||
self.assertGreater(floor_gap(out) - floor_gap(before), 6)
|
||||
|
||||
def test_silence_and_tiny_sounds_pass_without_error(self):
|
||||
silent = clean_up_voice(sound(np.zeros(seconds(1.0), dtype=np.float32)))["waveform"]
|
||||
self.assertEqual(float(silent.abs().max()), 0.0)
|
||||
short = clean_up_voice(sound(voice_like(0.05, self.rng)))["waveform"]
|
||||
self.assertTrue(torch.isfinite(short).all())
|
||||
empty = clean_up_voice(sound(np.zeros(0, dtype=np.float32)))["waveform"]
|
||||
self.assertEqual(tuple(empty.shape), (1, 1, 0))
|
||||
|
||||
def test_loudness_meter_reads_a_full_scale_tone_as_broadcast_meters_do(self):
|
||||
# A 997 Hz sine at 0 dBFS reads -3.01 LUFS on a BS.1770 meter.
|
||||
t = np.arange(seconds(3.0)) / RATE
|
||||
tone = torch.from_numpy(np.sin(2 * np.pi * 997 * t).astype(np.float32))
|
||||
self.assertAlmostEqual(integrated_loudness(tone, RATE), -3.01, delta=0.1)
|
||||
|
||||
|
||||
class NodeTests(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.folder = tempfile.TemporaryDirectory()
|
||||
self.path = Path(self.folder.name, "voice.wav")
|
||||
rng = np.random.default_rng(3)
|
||||
voice = voice_like(2.0, rng)
|
||||
quiet = np.zeros(seconds(1.0), dtype=np.float32)
|
||||
data = np.stack([np.concatenate([quiet, voice, quiet])] * 2)
|
||||
with av.open(str(self.path), "w") as container:
|
||||
stream = container.add_stream("pcm_f32le", rate=RATE, layout="stereo")
|
||||
frame = av.AudioFrame.from_ndarray(data, format="fltp", layout="stereo")
|
||||
frame.sample_rate = RATE
|
||||
for packet in stream.encode(frame):
|
||||
container.mux(packet)
|
||||
for packet in stream.encode(None):
|
||||
container.mux(packet)
|
||||
|
||||
def tearDown(self):
|
||||
self.folder.cleanup()
|
||||
|
||||
def run_node(self, **values):
|
||||
from nodes import node_load_audio
|
||||
|
||||
node = node_load_audio.AusBossLoadAudio()
|
||||
with mock.patch.object(node_load_audio, "resolve_input_path", return_value=self.path):
|
||||
return asyncio.run(node.load_audio("voice.wav", 0.0, 0.0, **values))
|
||||
|
||||
def test_off_hands_on_the_sound_as_it_is(self):
|
||||
audio, length = self.run_node()
|
||||
self.assertEqual(tuple(audio["waveform"].shape), (1, 2, seconds(4.0)))
|
||||
self.assertAlmostEqual(length, 4.0, places=3)
|
||||
audio, _length = self.run_node(clean_up=False)
|
||||
self.assertEqual(audio["waveform"].shape[1], 2)
|
||||
|
||||
def test_on_cleans_after_the_trim_and_seconds_matches(self):
|
||||
audio, length = self.run_node(clean_up=True, max_seconds=3.0)
|
||||
waveform = audio["waveform"]
|
||||
self.assertEqual(waveform.shape[1], 1)
|
||||
self.assertAlmostEqual(length, waveform.shape[-1] / RATE, places=6)
|
||||
# Max length took the first 3 s (1 s quiet, 2 s voice), then the
|
||||
# quiet start went: about 2 s of voice and its pads are left.
|
||||
self.assertLess(length, 2.5)
|
||||
|
||||
def test_an_older_saves_empty_value_leaves_it_off(self):
|
||||
audio, _length = self.run_node(clean_up="")
|
||||
self.assertEqual(audio["waveform"].shape[1], 2)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user