Compare commits

...
Author SHA1 Message Date
ausboss 1276d95708 Load Audio: Clean up switch for voice recordings
- New optional clean_up input (off by default, so saved workflows run
  as before). On, the sound that comes out is one channel, with rumble
  and mains hum under 80 Hz taken out, hiss turned down by up to 12 dB
  from a noise print of the quietest moments, the quiet start and end
  cut (0.1 s kept before the first sound, 0.2 s after the last), and the
  volume set to -18 LUFS with a look-ahead limiter that keeps peaks under
  -1 dBFS.
- It runs after the start/end/max trim, so seconds is the length of the
  sound handed on.
- Torch only, at the file's own sample rate. No new dependency.
- The card gets a Clean up off | on row; older saves, which hand the new
  input the upload button's empty value, open with it off.
2026-11-12 22:17:32 -06:00
7 changed files with 440 additions and 4 deletions
+8
View File
@@ -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
+26 -1
View File
@@ -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.
+4
View File
@@ -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
+212
View File
@@ -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"]
+19 -2
View File
@@ -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))
+2 -1
View File
@@ -570,7 +570,8 @@
"end_seconds"
],
"optional": [
"max_seconds"
"max_seconds",
"clean_up"
],
"outputs": [
[
+169
View File
@@ -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()