"""
Audio Helper — implementation module.
Utilities for working with audio files: format conversion, duration probing,
load/save in numpy form, regular-interval splitting, concatenation, room-tone
mixing, silent-segment generation, and optional Demucs-based source separation.
Dependencies
------------
- ffmpeg-python (wraps the system ``ffmpeg`` binary)
- numpy
- soundfile
- scipy
- tqdm
- os-helper
Optional (``demucs`` extra)
---------------------------
- torch, torchaudio — only required by :func:`separate_sources` and loaded
lazily via :func:`_require_torch` / :func:`_require_torchaudio`. Install with:
``pip install 'audio-helper[demucs]'``.
Usage Example
-------------
>>> import audio_helper as ah
>>> ah.sound_converter("in.mp3", "out.wav", freq=44100, channels=1)
>>> duration = ah.get_audio_duration("in.mp3")
>>> ah.mix_room_tone("narration.wav", "narration-rt.wav", noise_db=-42)
Author
------
Warith Harchaoui, Ph.D. — https://linkedin.com/in/warith-harchaoui/
"""
from __future__ import annotations
# Standard library first (PEP 8 import grouping): concurrency.
import concurrent.futures
import os
import re
from typing import TYPE_CHECKING
# Third-party: ffmpeg wrapper, numpy, scipy signal/IO helpers, progress bar.
import ffmpeg
import numpy as np
import os_helper as osh
import scipy.io.wavfile as wav
from scipy.fftpack import dct
from scipy.signal import get_window, resample
from tqdm import tqdm
# ``torch`` / ``torchaudio`` are optional (demucs extra) dependencies. Import
# them only for the type checker so annotations can name ``torch.Tensor``
# without forcing every user of this module to install torch just to convert
# an audio file. Under ``from __future__ import annotations`` these names live
# purely in the (never-evaluated) annotation strings.
if TYPE_CHECKING:
import torch
import torchaudio
# Shared, actionable hint raised whenever a torch-only feature is reached
# without the optional extra installed. Kept as a module constant so the
# wording stays identical across the two lazy-import helpers below.
_DEMUCS_EXTRA_HINT: str = (
"This feature requires the optional 'demucs' extra. "
"Install with: pip install 'audio-helper[demucs]'"
)
def _require_torch() -> torch:
"""Import and return the optional :mod:`torch` module lazily.
Torch is only needed by the Demucs-backed source-separation path, so we
avoid importing it at module load time and surface a clear, actionable
error when it is missing instead of a raw ``ImportError``.
Returns
-------
module
The imported :mod:`torch` module.
Raises
------
ImportError
If :mod:`torch` is not installed, re-raised with the install hint.
"""
# Defer the import so the base package works without the demucs extra.
try:
import torch
return torch
except ImportError as exc:
# Replace the terse stdlib message with an install instruction.
raise ImportError(_DEMUCS_EXTRA_HINT) from exc
def _require_torchaudio() -> torchaudio:
"""Import and return the optional :mod:`torchaudio` module lazily.
Mirror of :func:`_require_torch` for the :mod:`torchaudio` dependency,
which supplies the Demucs pipeline and the ``Fade`` transform.
Returns
-------
module
The imported :mod:`torchaudio` module.
Raises
------
ImportError
If :mod:`torchaudio` is not installed, re-raised with the hint.
"""
# Same lazy pattern as _require_torch: keep the optional dep optional.
try:
import torchaudio
return torchaudio
except ImportError as exc:
raise ImportError(_DEMUCS_EXTRA_HINT) from exc
# Container/codec extensions we treat as "audio-bearing" when validating a
# file. Video containers are appended below because their audio track is a
# perfectly valid input for every operation in this module.
audio_extensions: list[str] = [
"aif",
"aiff",
"alac",
"amr",
"ape",
"au",
"flac",
"gsm",
"iff",
"m4a",
"m4b",
"m4p",
"mp3",
"ogg",
"oga",
"opus",
"ra",
"ram",
"raw",
"sln",
"tta",
"voc",
"vox",
"wav",
"wma",
"wv",
"webm",
"rmi",
]
# Video containers whose embedded audio stream we can decode via ffmpeg.
# These get folded into ``audio_extensions`` so callers can pass a ``.mp4``
# or ``.mkv`` straight through without pre-extracting the audio.
video_extensions: list[str] = [
"mp4",
"avi",
"mov",
"wmv",
"flv",
"mkv",
"webm",
"mpeg",
"mpg",
"m4v",
"3gp",
"ogv",
"mxf",
"ts",
"vob",
"m2ts",
"mts",
"rm",
"asf",
]
audio_extensions += video_extensions
def _refuse_output_over_input(output_audio_filename: str, inputs: list[str]) -> None:
"""Refuse a destination that is also one of the sources.
Every writer here settles its destination and then calls
:func:`_overwrite_audio_file`, which with ``overwrite=True`` **deletes** the
existing file so it can be rewritten. When the destination is one of the
sources, that deletes the source before ffmpeg ever opens it: the caller
loses the input and gets an exception instead of a result. Measured on four
of this module's five writers -- ``sound_converter``,
``extract_audio_chunk``, ``mix_room_tone`` and ``audio_concatenation`` -- a
one-second input was gone from disk afterwards.
With ``overwrite=False`` the same confusion is quieter and worse:
``_overwrite_audio_file`` finds a valid audio file at the destination,
returns it as "already done", and the caller receives the *first input*
believing it is the result. Concatenating a 1 s file with a 2 s file
returned the 1 s file and raised nothing at all.
The check therefore belongs *before* the overwrite decision rather than
inside it: once that function has run, the safe answer is already gone.
Paths are compared through :func:`os_helper.relative2absolute_path`, so
``a.wav``, ``./a.wav`` and ``../here/a.wav`` are recognised as one file.
Parameters
----------
output_audio_filename : str
The destination about to be written.
inputs : list of str
Every path this operation reads.
Raises
------
ValueError
When the destination resolves to one of the inputs.
Examples
--------
>>> _refuse_output_over_input("out.wav", ["a.wav", "b.wav"]) is None
True
>>> _refuse_output_over_input("a.wav", ["a.wav"])
Traceback (most recent call last):
ValueError: a.wav is both an input and the output; refusing to overwrite a source file
"""
target = osh.relative2absolute_path(output_audio_filename)
for source in inputs:
if osh.relative2absolute_path(source) == target:
raise ValueError(
f"{source} is both an input and the output; refusing to overwrite a source file"
)
def _output_is_stale(output_audio_filename: str, inputs: list[str] | None) -> bool:
"""True when *output* predates any of *inputs*, so it cannot be their result.
A missing input is not staleness -- the caller's own validation reports
that, with a better message than this function could. An unreadable
timestamp is treated as fresh: refusing to reuse a cache because `stat`
failed would turn a filesystem quirk into repeated re-encoding.
"""
if not inputs:
return False
try:
produced = os.path.getmtime(output_audio_filename)
except OSError:
return False
for source in inputs:
try:
if os.path.getmtime(source) > produced:
return True
except OSError:
continue
return False
def _overwrite_audio_file(
output_audio_filename: str,
overwrite: bool = True,
inputs: list[str] | None = None,
) -> str | None:
"""
Decide what to do with an existing output audio file before (re)writing it.
"Skip" means **skip if up to date**, not merely "skip if a file is there".
When *inputs* is given and the existing output is older than any of them,
it cannot be their result, so it is deleted and the caller regenerates.
That distinction is not pedantry. `mix_room_tone`, `audio_concatenation`
and `split_audio_regularly` all default to ``overwrite=False``, so the
caching path is the one a caller gets without asking for it. Measured
before this check existed: concatenating a 1 s and a 2 s file gave a 3 s
join, replacing the 2 s input with a 5 s one and re-running gave **the old
3 s join back**, with nothing raised and nothing printed at the default
verbosity. That is the ordinary way to use these functions -- edit an
input, run the pipeline again -- and it returned stale audio in silence.
Comparing modification times is what every build system means by "already
done", it needs no provenance sidecar, and it stays silent on a legitimate
cache hit. It is deliberately not a promise that the output *matches* the
inputs: two different inputs written to one destination, the output being
newer than both, is indistinguishable without recording a hash, and
warning on every cache hit to cover it would make the warning worthless.
Parameters
----------
output_audio_filename : str
Path to the output audio file.
overwrite : bool, optional
If True (the default), an existing file is deleted so it can be
rewritten. If False, an existing **valid and up-to-date** audio file is
returned as-is (signalling "skip"); an existing invalid or stale file is
deleted.
inputs : list of str, optional
The paths this operation reads. Used for the staleness comparison only;
omitted means no comparison, which is right for a generator like
:func:`generate_silent_audio` that reads nothing.
Returns
-------
str or None
- The path itself when ``overwrite=False`` and the existing file is
a valid audio file (caller should skip).
- None in every other case (caller should proceed with writing).
"""
if not (overwrite) and osh.file_exists(output_audio_filename):
osh.info(f"Output audio file already exists:\n\t{output_audio_filename}")
if is_valid_audio_file(output_audio_filename):
if not _output_is_stale(output_audio_filename, inputs):
return output_audio_filename
osh.remove_files([output_audio_filename])
osh.info(
f"Output audio file is older than its input, regenerating:"
f"\n\t{output_audio_filename}"
)
return None
else:
osh.remove_files([output_audio_filename])
osh.info(f"Deleting invalid output audio file:\n\t{output_audio_filename}")
elif overwrite and osh.file_exists(output_audio_filename):
osh.remove_files([output_audio_filename])
osh.info(f"Deleting output audio file for overwrite:\n\t{output_audio_filename}")
return None
def _overwrite_audio_list(output_audio_list: list[str], overwrite: bool = True) -> dict | None:
"""
Decide what to do with a list of existing output audio files before (re)writing them.
Parameters
----------
output_audio_list : List[str]
List of paths to the output audio files.
overwrite : bool, optional
If True (the default), any existing files are deleted so they can
be rewritten. If False, when **every** file exists and is a valid
audio file, a ``{stem_name: absolute_path}`` dict is returned
(signalling "skip"); otherwise None is returned (caller proceeds).
Returns
-------
dict or None
``{stem_name: absolute_path}`` when the caller should skip
(overwrite=False and all files already valid), otherwise None.
"""
# Skip-path: only reuse existing outputs when EVERY target already exists
# and is a valid audio file (a generator keeps this short-circuiting).
if not (overwrite) and all(
osh.file_exists(f) and is_valid_audio_file(f) for f in output_audio_list
):
stem_keys = []
stem_files = []
for f in output_audio_list:
_, b, _ = osh.folder_name_ext(f)
o = osh.relative2absolute_path(f)
stem_keys.append(b)
stem_files.append(o)
# Pair stem names with their absolute paths. ``strict=False`` keeps
# the historical zip-truncation behaviour (lists are always equal
# length here since they are built in lock-step above).
d = dict(zip(stem_keys, stem_files, strict=False))
s = "\n\t".join([f"{k}:\t{v}" for k, v in d.items()])
osh.info(f"Sources already separated for at:\n\t{s}")
return d
elif overwrite:
for f in output_audio_list:
if osh.file_exists(f):
osh.remove_files([f])
osh.info(f"Deleting output audio file for overwrite:\n\t{f}")
return None
[docs]
def is_valid_audio_file(file_path: str) -> bool:
"""
Check if the given file is a valid audio file using ffmpeg-python (ffprobe).
Parameters
----------
file_path : str
Path to the audio file.
Returns
-------
bool
True if the file contains a valid audio stream, False otherwise.
Notes
-----
The function uses ffprobe to inspect the file and determine if an audio stream is present.
"""
# ``audio_extensions`` is read-only here (module-level list built once at
# import), so no ``global`` declaration is needed.
# By default, the file is considered invalid
valid = False
try:
probe = ffmpeg.probe(file_path)
audio_stream = next(
(stream for stream in probe["streams"] if stream["codec_type"] == "audio"),
None,
)
valid = audio_stream is not None
except (ffmpeg.Error, KeyError, OSError) as exc:
osh.debug(f"is_valid_audio_file: ffprobe failed for {file_path}: {exc}")
valid = False
_, _, ext = osh.folder_name_ext(file_path)
if ext.lower() not in audio_extensions:
valid = False
osh.info(f"Audio file {file_path} is {'valid' if valid else 'invalid'}")
return valid
[docs]
def get_audio_duration(file_path: str) -> float:
"""
Get the duration of an audio file in seconds using ffmpeg.
Parameters
----------
file_path : str
Path to the audio file.
Returns
-------
float
Duration of the audio file in seconds.
Raises
------
AssertionError
If the file is missing, if it contains no audio stream, or if
neither the stream nor the container reports a duration.
ffmpeg.Error
If ``ffmpeg.probe`` itself fails on the file.
"""
osh.checkfile(file_path, msg=f"Audio file not found at {file_path}")
probe = ffmpeg.probe(file_path)
audio_stream = next(
(stream for stream in probe["streams"] if stream["codec_type"] == "audio"), None
)
assert audio_stream is not None, f"No audio stream found in the file: {file_path}"
# Some containers (raw/headerless formats among audio_extensions, certain
# ADTS/MP3 streams) omit "duration" on the audio stream itself and only
# report it at the container ("format") level — is_valid_audio_file above
# already treats a missing key here as an expected ffprobe outcome (its
# except clause catches KeyError), so fall back the same way here instead
# of letting a raw KeyError escape from the one function that actually
# needs this value.
duration = audio_stream.get("duration") or probe.get("format", {}).get("duration")
assert duration is not None, f"No duration reported for the file: {file_path}"
return float(duration)
[docs]
def load_audio(
file_path: str,
target_sample_rate: int | None = None,
to_mono: bool = True,
to_numpy: bool = False,
two_channels: bool = False,
) -> tuple[torch.Tensor | np.ndarray, int]:
"""
Load ANY audio (or video-with-audio) file, optionally resample, convert to
mono or stereo, and return as a torch.Tensor (or optionally a NumPy array).
Decoding goes through :func:`sound_converter` (ffmpeg), NOT libsndfile /
``soundfile`` — so formats libsndfile cannot open (AAC/``.m4a``, ``.opus``,
``.webm``, and the audio track of video containers) load fine. The
intermediate float32 WAV is read with :mod:`scipy.io.wavfile`.
Parameters
----------
file_path : str
Path to the audio file.
target_sample_rate : int, optional
The target sample rate to resample the audio to. Defaults to the original sample rate.
to_mono : bool, optional
Whether to convert the audio to mono (default is True).
to_numpy : bool, optional
Whether to return the audio as a NumPy array (default is False). Otherwise, returns a torch.Tensor.
two_channels : bool, optional
Whether to force the audio into two channels (stereo).
Returns
-------
torch.Tensor or np.ndarray
The loaded audio signal.
int
Sample rate of the loaded audio.
Raises
------
ValueError
If ffmpeg is missing, the file cannot be read/decoded, or it carries
no audio stream — always with an actionable message.
"""
osh.checkfile(file_path, msg=f"Input audio file not found: {file_path}")
# Probe (no decode) for native rate/channels so target_sample_rate=None
# keeps the original rate and channel count is preserved when asked.
try:
probe = ffmpeg.probe(file_path)
except ffmpeg.Error as exc:
detail = exc.stderr.decode("utf-8", "replace") if exc.stderr else str(exc)
raise ValueError(f"ffmpeg could not read {file_path!r}: {detail}") from exc
except FileNotFoundError as exc: # ffmpeg binary not on PATH
raise ValueError(
"ffmpeg is required to decode audio but was not found on PATH — "
"install it (e.g. `brew install ffmpeg` / `apt install ffmpeg`)."
) from exc
audio_streams = [s for s in probe.get("streams", []) if s.get("codec_type") == "audio"]
if not audio_streams:
raise ValueError(f"no audio stream found in {file_path!r}")
native_sr = int(audio_streams[0]["sample_rate"])
native_ch = int(audio_streams[0].get("channels", 1))
out_rate = int(target_sample_rate) if target_sample_rate else native_sr
out_channels = 1 if to_mono else (2 if two_channels else native_ch)
# Transcode ANY input -> float32 WAV via ffmpeg (sound_converter does the
# decode + resample + down/up-mix), then read it with scipy — no soundfile.
with osh.temporary_filename(suffix=".wav", mode="wb") as tmp_wav:
sound_converter(
file_path,
tmp_wav,
freq=out_rate,
channels=out_channels,
encoding="pcm_f32le",
)
sr, audio = wav.read(tmp_wav)
audio = np.asarray(audio, dtype=np.float32)
sample_rate = int(sr)
# scipy returns (n,) for a single channel, (n, channels) otherwise.
# Honour the historical contract: mono -> 1-D, two_channels -> (n, 2).
if to_mono and audio.ndim == 2:
audio = np.mean(audio, axis=1)
elif two_channels and audio.ndim == 1:
audio = np.vstack([audio, audio]).T
if to_numpy:
return audio, sample_rate
torch = _require_torch()
# torch.Tensor with (channels, time) shape, matching prior behaviour.
audio_t = audio.T if audio.ndim == 2 else audio
return torch.from_numpy(np.ascontiguousarray(audio_t)), sample_rate
#: Codecs whose stream has no compressed bitrate to carry over. Their figure
#: is arithmetic, not a choice: 24 kHz 16-bit mono PCM reads 384 kb/s, and
#: handing that to a lossy encoder asks for the ceiling of the codec rather
#: than for the quality of the source.
LOSSLESS_CODECS: frozenset[str] = frozenset(
{"flac", "alac", "wavpack", "tta", "truehd", "mlp", "ape", "shorten"}
)
def _source_bitrate(input_audio: str) -> str | None:
"""
The input's own bitrate, when it has one worth carrying over.
Parameters
----------
input_audio : str
Path to the input audio file.
Returns
-------
str or None
Bits per second as a string, or None when the input is uncompressed
or when ffprobe reports no usable figure.
Notes
-----
Stream first, container second, the same fallback
:func:`get_audio_duration` makes and for the same reason: some MP3 and
ADTS streams report a bitrate only at the container level.
A failed probe returns None rather than raising. The caller is choosing a
default, and losing a default is not worth failing a conversion that
ffmpeg would otherwise complete.
"""
try:
probe = ffmpeg.probe(input_audio)
except (ffmpeg.Error, OSError): # pragma: no cover - depends on the file
return None
stream = next((s for s in probe.get("streams", []) if s.get("codec_type") == "audio"), None)
if stream is None:
return None
codec = str(stream.get("codec_name", "")).lower()
if codec.startswith("pcm_") or codec in LOSSLESS_CODECS:
return None
value = stream.get("bit_rate") or probe.get("format", {}).get("bit_rate")
if value in (None, "", "N/A"):
return None
try:
measured = int(float(value))
except (TypeError, ValueError): # pragma: no cover - ffprobe oddity
return None
return str(measured) if measured > 0 else None
def _normalise_bitrate(bitrate: str | int) -> str:
"""
Normalise a requested bitrate into the string ffmpeg expects for ``-b:a``.
Parameters
----------
bitrate : str or int
Bits per second as an integer (``64000``), or an ffmpeg-style string
(``"64k"``, ``"128K"``, ``"64000"``).
Returns
-------
str
The value to hand to ffmpeg, e.g. ``"64k"``.
Raises
------
AssertionError
If the value is not a positive integer nor an ffmpeg-style string.
Notes
-----
``bool`` is a subclass of ``int``, so ``True`` would otherwise pass as the
bitrate 1. It is refused by name.
"""
assert not isinstance(bitrate, bool), f"Invalid bitrate: {bitrate!r}"
if isinstance(bitrate, int):
assert bitrate > 0, f"Invalid bitrate: {bitrate!r} (must be greater than zero)"
return str(bitrate)
text = str(bitrate).strip()
assert re.fullmatch(r"[1-9][0-9]*[kKmM]?", text), (
f"Invalid bitrate: {bitrate!r} (expected bits per second such as 64000, "
'or an ffmpeg-style string such as "64k")'
)
return text
def _ffmpeg_quiet() -> bool:
"""Whether ffmpeg's own output should be muted, read from ``osh.verbosity()``.
Muted at ``0`` (the default, WARNING), audible from ``1`` (INFO). Every
helper that runs ffmpeg asks here, so raising the verbosity to see why an
encode failed shows ffmpeg's own diagnostics wherever it runs.
"""
return osh.verbosity() <= 0
[docs]
def sound_converter(
input_audio: str,
output_audio: str,
freq: int = 44100,
channels: int = 1,
encoding: str = "pcm_s16le",
overwrite: bool = True,
bitrate: str | int | None = None,
) -> str:
"""
Convert an audio file to the specified format using ffmpeg-python.
Parameters
----------
input_audio : str
Path to the input audio file.
output_audio : str
Path to the output audio file with the desired format extension.
freq : int, optional
Output sample rate in Hz (default is 44100).
channels : int, optional
Number of audio channels in the output (default is 1 for mono).
encoding : str, optional
Audio codec to use for encoding the output file (default is 'pcm_s16le').
overwrite : bool, optional
If True (the default), an existing output file is deleted and rewritten.
If False, an existing **valid** output file is returned as-is without
re-converting (see :func:`_overwrite_audio_file`).
bitrate : str or int or None, optional
Target bitrate for a lossy output: bits per second as an integer
(``64000``) or an ffmpeg-style string (``"64k"``). None, the default,
**carries the input's own bitrate over**, so a 96 kb/s MP3 re-encoded
to MP3 stays 96 kb/s instead of dropping to whatever ffmpeg would
have picked. An uncompressed input has no bitrate to carry (see
:data:`LOSSLESS_CODECS`), and the choice is left to ffmpeg there.
Ignored with a warning when the *output* is WAV, which is
uncompressed and has no bitrate to set.
Returns
-------
str :
Path to the output audio file.
Raises
------
AssertionError
If the input audio file does not exist or is not a valid audio file.
ffmpeg.Error
If the underlying ffmpeg invocation fails.
Notes
-----
The conversion is handled using a temporary file structure to manage intermediate formats.
Two intermediate WAV files are used before generating the final output audio file.
Which stage each setting lands on matters, because the function encodes
twice. ``freq`` and ``channels`` are applied to the intermediate WAV, so
the final container inherits them. A **lossy** ``encoding`` is applied to
the final encode instead: naming it on the intermediate WAV would write
compressed payload inside a WAV container, which the last step then
decodes and re-encodes, two lossy passes for one requested conversion.
A PCM ``encoding`` stays on the intermediate, where it belongs.
``bitrate`` follows the lossy codec, on the final encode.
Inheriting the input's bitrate is what a conversion is usually asked to
do: the caller named a container, not a quality, and silently halving the
quality is the kind of loss nobody measures until much later. Asked for
nothing on a 24 kHz mono stream, ffmpeg picks 32 kb/s, so "no bitrate"
was never a neutral choice.
Examples
--------
>>> # A forty-minute narration at 64 kb/s, 24 kHz, mono.
>>> sound_converter(
... "narration.wav",
... "narration.mp3",
... freq=24000,
... channels=1,
... bitrate="64k",
... ) # doctest: +SKIP
'narration.mp3'
>>> # Re-containering a podcast episode keeps its 96 kb/s: the caller
>>> # asked to change the container, not the quality.
>>> sound_converter("episode.mp3", "episode.m4a") # doctest: +SKIP
'episode.m4a'
"""
osh.info(f"Converting audio file: {input_audio} into {output_audio}")
# Check if the input audio file exists
osh.checkfile(input_audio, msg=f"Input audio file not found: {input_audio}")
# Check if the input audio file is valid
assert is_valid_audio_file(input_audio), f"Invalid audio file: {input_audio}"
_refuse_output_over_input(output_audio, [input_audio])
o = _overwrite_audio_file(output_audio, overwrite, [input_audio])
if o is not None:
return o
_, _, ext_in = osh.folder_name_ext(input_audio)
_, _, ext_out = osh.folder_name_ext(output_audio)
quiet = _ffmpeg_quiet()
# A lossy codec and its bitrate belong on the FINAL encode; PCM stays on
# the intermediate WAV. See the Notes above for why the distinction is not
# cosmetic.
intermediate_is_pcm = encoding.lower().startswith("pcm_")
intermediate_encoding = encoding if intermediate_is_pcm else "pcm_s16le"
final_kwargs: dict[str, str] = {}
if not intermediate_is_pcm:
final_kwargs["acodec"] = encoding
if bitrate is not None:
if ext_out.lower() == "wav":
osh.warning(
f"bitrate={bitrate!r} ignored for {output_audio}: a WAV output is "
"uncompressed, so it carries no bitrate. Ask for a lossy container "
"(.mp3, .m4a, .ogg) to make the setting meaningful."
)
else:
final_kwargs["audio_bitrate"] = _normalise_bitrate(bitrate)
elif ext_out.lower() != "wav":
# Nothing asked for: carry the input's own bitrate over rather than
# let ffmpeg pick. An uncompressed input has none to carry, and
# inventing one there would be a decision this package has no basis
# to make.
inherited = _source_bitrate(input_audio)
if inherited is not None:
final_kwargs["audio_bitrate"] = inherited
osh.info(f"carrying the input bitrate over: {inherited} b/s")
else:
osh.debug(
f"{input_audio} reports no compressed bitrate to carry over; "
"leaving the choice to ffmpeg"
)
# Use temporary files for intermediate WAV processing (for robustness)
with (
osh.temporary_filename(suffix=".wav", mode="wb") as first_wav,
osh.temporary_filename(suffix=".wav", mode="wb") as second_wav,
):
# Non-WAV inputs are transcoded to WAV first; a WAV input is just
# copied so we never re-encode losslessly-decoded audio needlessly.
if ext_in.lower() != "wav":
# Convert the input audio file to a temporary WAV file
ffmpeg.input(input_audio).output(first_wav, format="wav").run(
overwrite_output=True, quiet=quiet
)
else:
osh.copyfile(input_audio, first_wav)
# Convert the temporary WAV file to another WAV file with specified parameters
ffmpeg.input(first_wav).output(
second_wav, ar=freq, ac=channels, acodec=intermediate_encoding
).run(overwrite_output=True, quiet=quiet)
# Same branch on the output side: encode to the requested container
# unless the target is already WAV, in which case a copy suffices.
if ext_out.lower() != "wav":
# Final conversion to the specified output format
ffmpeg.input(second_wav).output(output_audio, **final_kwargs).run(
overwrite_output=True, quiet=quiet
)
else:
osh.copyfile(second_wav, output_audio)
# Check if the output audio file was successfully created
osh.checkfile(output_audio, msg=f"Failed to convert audio file:\n\t{output_audio}")
assert is_valid_audio_file(output_audio), f"Invalid audio file:\n\t{output_audio}"
osh.info(f"Audio file converted successfully:\n\t{output_audio}")
return output_audio
[docs]
def save_audio(signal: torch.Tensor | np.ndarray, file_path: str, sample_rate: int = 44100) -> None:
"""
Save an audio signal as a file using torchaudio (tensor) or scipy.io.wavfile (numpy).
Parameters
----------
signal : torch.Tensor or np.ndarray
The audio signal to save.
file_path : str
Path to the output audio file.
sample_rate : int
The sample rate of the audio signal, in Hz.
Raises
------
AssertionError
If ``signal`` is neither a torch.Tensor nor a numpy.ndarray.
ImportError
If ``signal`` is a torch.Tensor but the optional ``demucs`` extra
(torch / torchaudio) is not installed.
"""
try:
import torch as _torch
is_tensor = isinstance(signal, _torch.Tensor)
except ImportError:
is_tensor = False
if is_tensor: # (channels, time) convention
signal = signal.detach().cpu().numpy()
if len(signal.shape) == 1:
signal = signal.reshape(1, -1) # (1, time) convention
signal = signal.T # transpose to the (time, channels) convention
save_audio(signal, file_path, sample_rate)
elif isinstance(signal, np.ndarray): # (time, channels) convention
_, _, ext = osh.folder_name_ext(file_path)
if ext.lower() == "wav":
wav.write(file_path, sample_rate, signal)
else:
with osh.temporary_filename(suffix=".wav", mode="wb") as wav_audio_file:
wav.write(wav_audio_file, sample_rate, signal)
channels = 1 if len(signal.shape) == 1 else signal.shape[1]
sound_converter(
input_audio=wav_audio_file,
output_audio=file_path,
freq=sample_rate,
channels=channels,
encoding="pcm_s16le",
overwrite=True,
)
assert is_valid_audio_file(file_path), f"Audio file not saved to {file_path}"
osh.info(f"Audio signal saved to {file_path}")
else:
raise AssertionError(
f"signal must be a torch.Tensor or a numpy.ndarray, got {type(signal)!r}"
)
def _separate_sources(
model: torch.nn.Module,
mix: torch.Tensor,
sample_rate: int,
segment: float = 10.0,
overlap: float = 0.1,
device: str | None = None,
nb_workers: int | None = 2,
) -> torch.Tensor:
"""
Apply a source separation model to a given audio mixture, processing the mixture in segments with overlap and fades,
using multithreading to parallelize segment processing.
Parameters
----------
model : torch.nn.Module
The pre-trained source separation model to apply to the audio mixture.
mix : torch.Tensor
The audio mixture tensor with shape (batch_size, channels, length).
sample_rate : int
The sample rate of the audio mixture.
segment : float, optional
The length of each segment in seconds to process (default is 10.0 seconds).
overlap : float, optional
The overlap duration between consecutive segments in seconds (default is 0.1 seconds).
device : str, optional
The device on which to run the computations.
nb_workers : int, optional
The number of threads to use for parallel processing (default is 2).
Returns
-------
torch.Tensor
Tensor containing the separated sources, with shape (batch_size, num_sources, channels, length).
Notes
-----
The function processes the audio mixture in overlapping segments, applies the source separation model to each segment
in parallel using multithreading, and uses linear fades to smooth transitions between overlapping segments. The
separated sources are then reassembled into the final output tensor.
"""
torch = _require_torch()
# Do not use all cores and leave one for the system!
# Get the number of workers from osh if nb_workers is not provided
if nb_workers is None:
nb_workers = osh.get_nb_workers()
# Adjust workers count if nb_workers is negative (sklearn convention: -1 = all cores, -2 = all but one, ...)
if nb_workers < 0:
nb_workers = osh.get_nb_workers() + nb_workers + 1
# Limit the number of workers to the maximum available minus one for the system
MAX_NB_WORKERS = osh.get_nb_workers()
if nb_workers >= MAX_NB_WORKERS:
nb_workers = MAX_NB_WORKERS - 1
if nb_workers < 1:
nb_workers = 1
# Check if cuda is available
if device is None:
# check pytorch device
device = "cuda" if torch.cuda.is_available() else "cpu"
# If CUDA, limit the number of workers to 1 to avoid CUDA out-of-memory errors
if device == "cuda" or device == torch.device("cuda"):
nb_workers = 1
# Convert the device to a torch.device object if it is a string
if isinstance(device, str):
device = torch.device(device)
# Move the audio mixture and the model to the specified device. Unlike
# nn.Module.to (in-place), Tensor.to returns a NEW tensor — the result
# must be reassigned or `mix` silently stays on its original device,
# which then raises a cross-device error the first time `model.forward`
# (on `device`) touches a `mix` slice still sitting on the CPU.
mix = mix.to(device)
mix = mix.float() # Convert mix to float32
model.to(device)
# Get the batch size, number of channels, and length of the audio mixture
batch, channels, length = mix.shape
# Calculate the length of each chunk (segment) in frames, accounting for overlap
chunk_len = int(sample_rate * segment * (1 + overlap))
# Calculate the number of overlap frames
overlap_frames = int(overlap * sample_rate)
from torchaudio.transforms import Fade
# Initialize a tensor to store the final separated sources
final = torch.zeros(batch, len(model.sources), channels, length, device=device)
# Calculate the total number of chunks to process
total_chunks = (length - overlap_frames) // (chunk_len - overlap_frames) + 1
# Define the function to process each chunk in parallel
def process_chunk(start: int, end: int) -> tuple[int, int, torch.Tensor]:
"""
Function to process a single chunk of the audio mixture.
Parameters
----------
start : int
Start index of the chunk.
end : int
End index of the chunk.
Returns
-------
tuple of (int, int, torch.Tensor)
The clamped start and end indices and the separated output for
this chunk (already faded so overlapping regions blend cleanly).
"""
# A fresh Fade instance per call, never a shared/mutated one: chunks run
# concurrently under ThreadPoolExecutor when nb_workers > 1, and mutating
# a single closed-over Fade's fade_out_len in place (the original code)
# raced across threads — the truncated last chunk's fade_out_len=0 could
# be read by an unrelated, concurrently-running chunk's fade(out) call.
fade_out_len = overlap_frames
if end > length:
end = length
fade_out_len = 0 # Disable fade out for the last (truncated) chunk
fade = Fade(fade_in_len=0, fade_out_len=fade_out_len, fade_shape="linear")
chunk = mix[:, :, start:end]
with torch.no_grad():
out = model.forward(chunk)
out = fade(out)
return start, end, out
# Use ThreadPoolExecutor to process the chunks in parallel
if nb_workers > 1:
with concurrent.futures.ThreadPoolExecutor(max_workers=nb_workers) as executor:
futures = []
start = 0
end = chunk_len
# Submit each chunk processing task to the executor
for _ in range(total_chunks):
futures.append(executor.submit(process_chunk, start, end))
start += chunk_len - overlap_frames
end = start + chunk_len
# Collect results and assemble the final output
for future in tqdm(
concurrent.futures.as_completed(futures),
desc="Processing chunks",
total=total_chunks,
):
start, end, out = future.result()
final[:, :, :, start:end] += out
else:
for i in tqdm(
range(total_chunks), desc="Processing chunks for source separation", total=total_chunks
):
start = i * (chunk_len - overlap_frames)
end = start + chunk_len
start, end, out = process_chunk(start, end)
final[:, :, :, start:end] += out
return final
# Process-wide cache for the Demucs model and its native sample rate. Loading
# the bundle is expensive (network + weights), so we build it once on the
# first ``separate_sources`` call and reuse it for every subsequent call.
separator_engine: torch.nn.Module | None = None
separator_engine_sample_rate: int | None = None
[docs]
def separate_sources(
input_audio_file: str,
output_folder: str | None = None,
device: str | None = None,
overwrite: bool = False,
nb_workers: int = -2,
output_format: str = "mp3",
) -> dict:
"""
Separate an input audio file into different sources (e.g., vocals, bass, drums, other) using a pre-trained model from pytorch called DEMUCS.
Parameters
----------
input_audio_file : str
Path to the input audio file.
output_folder : str, optional
Folder to save the separated sources. If None, the output folder will be created based on the input file's name.
device : str, optional
The device on which to run the computations. If not specified, CUDA will be used if available, otherwise CPU.
overwrite : bool, optional
Whether to overwrite existing files if they already exist (default is False).
nb_workers : int, optional
The number of workers (threads) to use for parallel processing of segments (default is -2 which corresponds to all cores except one).
output_format : str, optional
The format of the output audio files (default is 'mp3').
Returns
-------
dict
A dictionary mapping source names (e.g., 'vocals', 'bass', etc.) to the paths of the separated audio files.
Examples
--------
>>> separated_sources = separate_sources("input_audio.mp3", output_folder="output_folder", overwrite=True)
>>> print(separated_sources)
{'vocals': 'output_folder/vocals.mp3', 'drums': 'output_folder/drums.mp3', 'bass': 'output_folder/bass.mp3', 'other': 'output_folder/other.mp3'}
Notes
-----
The function uses the HDEMUCS_HIGH_MUSDB_PLUS model to separate audio into its constituent sources. It processes
the audio in segments with optional multithreading for parallel processing. The separated sources are saved as
audio files in the specified or generated output folder.
"""
global separator_engine, separator_engine_sample_rate
osh.info(f"Separating sources for:\n\t{input_audio_file}")
# Set up the output folder if not specified
if output_folder is None:
f, _, _ = osh.folder_name_ext(input_audio_file)
output_folder = f
# Check if files already exist and skip if not overwriting
stem_keys = ["vocals", "drums", "bass", "other"]
stem_files = [osh.join([output_folder, f"{stem}.{output_format}"]) for stem in stem_keys]
d = _overwrite_audio_list(stem_files, overwrite)
if d is not None:
return d
_require_torch()
_require_torchaudio()
# Initialize the separator engine if it hasn't been initialized yet
if separator_engine is None:
from torchaudio.pipelines import HDEMUCS_HIGH_MUSDB_PLUS
bundle = HDEMUCS_HIGH_MUSDB_PLUS
separator_engine = bundle.get_model()
separator_engine_sample_rate = bundle.sample_rate
two_channels = True
# Load the audio file and resample it if needed
waveform, sample_rate = load_audio(
input_audio_file,
target_sample_rate=separator_engine_sample_rate,
to_numpy=False,
two_channels=two_channels,
)
waveform = waveform.reshape(1, 2 if two_channels else 1, -1)
# Normalize the audio signal
ref = waveform.mean(0)
waveform = (waveform - ref.mean()) / ref.std()
# Separate the audio into sources using the multithreaded _separate_sources function
sources = _separate_sources(
separator_engine,
waveform,
separator_engine_sample_rate,
device=device,
segment=10,
overlap=0.1,
nb_workers=nb_workers,
)[0]
# Denormalize the separated sources
sources = sources * ref.std() + ref.mean()
# Get the list of sources from the model and process each source
sources_list = separator_engine.sources
sources = list(sources)
# Dictionary to store the output file paths for each source
res = {}
for stem in sources_list:
audio = sources.pop(0) # in (channels, time) shape
# convert it in scipy (time, channels) shape
audio = audio.detach().cpu().numpy().T
# reduce to mono which means channels = 1
audio = np.mean(audio, axis=1)
# check sample rate
if sample_rate != separator_engine_sample_rate:
num_samples = int(len(audio) * sample_rate / separator_engine_sample_rate)
audio = resample(audio, num_samples)
osh.make_directory(output_folder)
output_audio_file = osh.join([output_folder, f"{stem}.{output_format}"])
save_audio(audio, output_audio_file, sample_rate)
res[stem] = output_audio_file
osh.info(f"Saved {stem} to\n\t{output_audio_file}")
return res
[docs]
def generate_silent_audio(
duration: float,
output_audio_filename: str | None = None,
sample_rate: int = 44100,
overwrite: bool = False,
) -> str:
"""
Generate a silent audio file of a specified duration.
Parameters
----------
duration : float
The duration of the silent audio file in seconds.
output_audio_filename : str, optional
The path to save the generated silent audio file. If None, a default file name will be generated.
sample_rate : int, optional
The sample rate of the silent audio file in Hz (default is 44100 Hz).
overwrite : bool, optional
Whether to overwrite the output file if it already exists (default is False).
Returns
-------
str
The path to the generated silent audio file.
Raises
------
AssertionError
If ``duration`` is not strictly positive, or if the resulting file
is missing or empty after generation.
ffmpeg.Error
If the underlying ffmpeg invocation fails.
Notes
-----
Uses ffmpeg's ``anullsrc`` filter to write a silent audio file of the
specified duration and sample rate. When ``output_audio_filename`` is
None, a default name is derived from ``duration``.
"""
assert duration > 0, f"duration must be strictly positive, got {duration}"
# Generate default output file name if not provided
if osh.emptystring(output_audio_filename):
t = round(duration * 1000) # Convert duration to milliseconds for the filename
output_audio_filename = osh.join([f"silent_{t}.wav"])
# Check if the file already exists and handle based on the overwrite flag
if _overwrite_audio_file(output_audio_filename, overwrite) is not None:
return output_audio_filename
# Just make zeros (float32 PCM written via scipy, never soundfile)
zeros = np.zeros(int(duration * sample_rate), dtype=np.float32)
_, _, ext = osh.folder_name_ext(output_audio_filename)
if ext.lower() == "wav":
wav.write(output_audio_filename, sample_rate, zeros)
else:
with osh.temporary_filename(suffix=".wav", mode="wb") as temp_wav:
wav.write(temp_wav, sample_rate, zeros)
sound_converter(temp_wav, output_audio_filename, freq=sample_rate)
# Verify that the file was successfully generated and is valid
osh.checkfile(
output_audio_filename,
msg=f"Failed to generate silent audio file: {output_audio_filename}",
)
assert is_valid_audio_file(output_audio_filename), (
f"Generated silent audio file is invalid: {output_audio_filename}"
)
signal, sample_rate = load_audio(output_audio_filename, to_numpy=True, to_mono=True)
assert np.sum(np.abs(signal)) == 0, (
f"Generated silent audio file is not silent:\n\t{output_audio_filename}"
)
osh.info(f"Generated silent audio file: {output_audio_filename}")
return output_audio_filename
[docs]
def audio_concatenation(
audio_files: list[str],
output_audio_filename: str | None = None,
overwrite: bool = False,
) -> str:
"""
Concatenate multiple audio files into a single audio file.
Parameters
----------
audio_files : list
List of paths to the audio files.
output_audio_filename : str or None, optional
Path to save the concatenated audio file.
If None, the output file will be saved in the same folder as the first audio file.
overwrite : bool, optional
Whether to overwrite the output file if it already exists (default is False).
Returns
-------
str
Path to the concatenated audio file
Notes
-----
The function uses ffmpeg to concatenate multiple audio files into a single audio file.
"""
assert isinstance(audio_files, list) and len(audio_files) > 0, (
f"Invalid audio files list: {audio_files}"
)
s = "\n\t".join(audio_files)
# Two separate guards give the caller a precise reason: a missing path
# versus a present-but-non-audio file. Generators short-circuit on the
# first failure, so we do not probe every file once one is already bad.
assert all(osh.file_exists(f) for f in audio_files), (
f"Invalid audio files (file existence):\n\t{s}"
)
assert all(is_valid_audio_file(f) for f in audio_files), (
f"Invalid audio files (audio type):\n\t{s}"
)
if osh.emptystring(output_audio_filename):
folder, _, ext = osh.folder_name_ext(audio_files[0])
audio_files_basename = []
for f in audio_files:
_, b, _ = osh.folder_name_ext(f)
audio_files_basename.append(b)
b = "-".join(audio_files_basename)
output_audio_filename = osh.join([folder, f"{b}-concatenated.{ext}"])
# Check if the file already exists and handle based on the overwrite flag
_refuse_output_over_input(output_audio_filename, audio_files)
if _overwrite_audio_file(output_audio_filename, overwrite, audio_files) is not None:
return output_audio_filename
input_streams = [ffmpeg.input(f) for f in audio_files]
quiet = _ffmpeg_quiet()
_, _, ext = osh.folder_name_ext(output_audio_filename)
if ext.lower() == "wav":
(
ffmpeg.concat(*input_streams, v=0, a=1)
.output(output_audio_filename)
.run(overwrite_output=True, quiet=quiet)
)
else:
with osh.temporary_filename(suffix=".wav", mode="wb") as temp_wav:
(
ffmpeg.concat(*input_streams, v=0, a=1)
.output(temp_wav)
.run(overwrite_output=True, quiet=quiet)
)
sound_converter(temp_wav, output_audio_filename, freq=44100)
osh.checkfile(output_audio_filename, msg=f"Failed to concatenate audio files: {audio_files}")
assert is_valid_audio_file(output_audio_filename), (
f"Failed to concatenate audio files: {audio_files}"
)
osh.info(f"Concatenated audio files into: {output_audio_filename}")
return output_audio_filename
[docs]
def mix_room_tone(
input_audio: str,
output_audio: str | None = None,
noise_db: float = -42.0,
color: str = "pink",
sample_rate: int = 44100,
overwrite: bool = False,
) -> str:
"""
Mix a constant low-level ambient noise (room tone) on top of an audio track.
Parameters
----------
input_audio : str
Path to the input audio file (the "speech" track).
output_audio : str, optional
Path to the output file. If ``None``, ``<input>-roomtone.<ext>``
is written next to the input.
noise_db : float, optional
Noise level in decibels (default ``-42`` — inaudible but present;
sit between ``-45`` and ``-38`` for typical post-production use).
Amplitude is computed as ``10 ** (noise_db / 20)``.
color : str, optional
Noise color (default ``"pink"``). Accepted values follow ffmpeg's
``anoisesrc`` filter: ``"white"``, ``"pink"``, ``"brown"``
(sometimes ``"red"``), ``"blue"``, ``"violet"``, ``"velvet"``.
Pink is the standard "natural" choice for masking gaps between
speech recordings.
sample_rate : int, optional
Sample rate for the noise source (default 44100). The output
sample rate matches whatever ffmpeg's ``amix`` produces.
overwrite : bool, optional
Whether to overwrite the output file if it already exists
(default ``False``).
Returns
-------
str
Path to the mixed output file.
Notes
-----
Standard post-production trick to homogenize a montage of disparate
speech takes: perfectly silent gaps between cuts contrast unpleasantly
with the speech and make every cut audible. A constant background
ambience that sits ~40 dB below the speech masks the boundary while
staying below the conscious hearing threshold.
The noise length is the input duration + 0.5 s (a small head-room so
``amix=duration=first`` stops exactly at the speech end without
truncating the last few samples).
Examples
--------
>>> mix_room_tone("narration.wav", "narration-rt.wav", noise_db=-42)
>>> mix_room_tone("voice.mp3", color="brown", noise_db=-38)
"""
osh.checkfile(input_audio, msg=f"Input audio file not found: {input_audio}")
assert is_valid_audio_file(input_audio), f"Invalid input audio file: {input_audio}"
assert color in {"white", "pink", "brown", "red", "blue", "violet", "velvet"}, (
f"Unsupported noise color: {color!r}"
)
if osh.emptystring(output_audio):
folder, base, ext = osh.folder_name_ext(input_audio)
output_audio = osh.join([folder, f"{base}-roomtone.{ext}"])
_refuse_output_over_input(output_audio, [input_audio])
if _overwrite_audio_file(output_audio, overwrite, [input_audio]) is not None:
return output_audio
duration = get_audio_duration(input_audio)
amplitude = 10 ** (noise_db / 20.0)
# Small overshoot: anoisesrc duration = speech + 0.5s, then amix's
# `duration=first` clamps back to the speech length. Avoids truncating
# the tail of the speech by sub-sample rounding.
noise_duration = duration + 0.5
quiet = _ffmpeg_quiet()
# Codec is dictated by the output extension (.wav → pcm_s16le, others
# → ffmpeg's default for that container). Pcm preserves the mix
# losslessly; non-pcm containers compress.
_, _, ext = osh.folder_name_ext(output_audio)
out_kwargs = {}
if ext.lower() == "wav":
out_kwargs["acodec"] = "pcm_s16le"
speech = ffmpeg.input(input_audio)
noise = ffmpeg.input(
f"anoisesrc=color={color}:amplitude={amplitude:.6f}:"
f"duration={noise_duration:.3f}:sample_rate={sample_rate}",
f="lavfi",
)
mixed = ffmpeg.filter(
[speech.audio, noise.audio],
"amix",
inputs=2,
duration="first",
dropout_transition=0,
)
ffmpeg.output(mixed, output_audio, **out_kwargs).run(
overwrite_output=True,
quiet=quiet,
)
osh.checkfile(
output_audio,
msg=f"Failed to write room-tone output: {output_audio}",
)
assert is_valid_audio_file(output_audio), f"Generated room-tone file is invalid: {output_audio}"
osh.info(
f"Mixed room tone ({noise_db} dB {color}) into: {output_audio}",
)
return output_audio
[docs]
def split_audio_regularly(
sound_path: str,
chunk_folder: str,
split_time: float,
output_format: str = "mp3",
overwrite: bool = False,
suffix: str = "split",
) -> list[str]:
"""
Split an audio file into chunks of a specified duration.
Parameters
----------
sound_path : str
Path to the audio file.
chunk_folder : str
Path to the folder where the audio chunks will be saved.
split_time : float
Duration of each audio chunk in seconds.
output_format : str, optional
The format of the output audio files (default is 'mp3').
overwrite : bool, optional
Whether to overwrite the output files if they already exist (default is False).
suffix : str, optional
Suffix to add to the output audio files (default is 'split').
Returns
-------
List of audio file paths
Notes
-----
The function uses ffmpeg to split the audio file into chunks of the specified duration.
A trailing remainder shorter than 1 second is dropped rather than saved as its own
chunk (not worth a separate file), so the returned chunks may cover slightly less
than the full ``sound_path`` duration. This "drop if < 1s" rule only ever applies to
a remainder that follows at least one already-produced chunk: the very first chunk is
always emitted, so a ``sound_path`` shorter than 1 second still yields one chunk that
covers it in full instead of silently returning an empty list.
"""
assert is_valid_audio_file(sound_path), f"Invalid audio file: {sound_path}"
output_format = output_format.lower().replace(".", "")
# Ensure the output directory exists
osh.make_directory(chunk_folder)
# Calculate the total duration of the audio file
total_duration = get_audio_duration(sound_path)
# Process the chunks (actual splitting)
time_cursor = 0
counter = 0
output_audio_paths = []
# `counter == 0` keeps the very first chunk even when the whole file is
# shorter than 1 second: the `total_duration - time_cursor >= 1` half of
# this condition alone is meant to drop a tiny *trailing* remainder after
# at least one full chunk already exists, not to discard the entire file
# when it never reaches 1 second in the first place (that used to return
# an empty list for any sub-1s input, silently dropping all the audio).
while time_cursor < total_duration and (counter == 0 or total_duration - time_cursor >= 1):
chunk_path = osh.join([chunk_folder, f"chunk_{counter:04d}_{suffix}.{output_format}"])
s = time_cursor
e = min(time_cursor + split_time, total_duration)
extract_audio_chunk(sound_path, s, e, output_audio_filename=chunk_path, overwrite=overwrite)
added_duration = get_audio_duration(chunk_path)
osh.info(f"Chunk {counter:04d} of duration {added_duration} saved to:\n\t{chunk_path}")
output_audio_paths.append(chunk_path)
time_cursor += added_duration
counter += 1
s = "\n\t".join(output_audio_paths)
osh.info(
f"Audio file {sound_path} split into chunks of {split_time} seconds in {chunk_folder}:\n\t{s}"
)
return output_audio_paths
[docs]
def hz_to_mel(hz: float) -> float:
"""
Convert a frequency in Hertz to the Mel scale.
Parameters
----------
hz : float
Frequency in Hertz.
Returns
-------
float
Frequency in Mels.
"""
return 2595 * np.log10(1 + hz / 700.0)
[docs]
def mel_to_hz(mel: float) -> float:
"""
Convert a frequency in the Mel scale back to Hertz.
Parameters
----------
mel : float
Frequency in Mels.
Returns
-------
float
Frequency in Hertz.
"""
return 700 * (10 ** (mel / 2595.0) - 1)
[docs]
def mel_filter_banks(
num_filters: int, n_fft: int, sample_rate: int, low_freq: int, high_freq: int
) -> np.ndarray:
"""
Compute a Mel-filter bank for given parameters.
Parameters
----------
num_filters : int
Number of Mel filters to generate.
n_fft : int
The size of the FFT (number of FFT points).
sample_rate : int
The sample rate of the audio signal (in Hz).
low_freq : int
The lowest frequency in the Mel filter bank (in Hz).
high_freq : int
The highest frequency in the Mel filter bank (in Hz).
Returns
-------
np.ndarray
A 2D array where each row is a filter in the Mel-filter bank.
"""
# Guard the band before any arithmetic. Without this, a high_freq above the
# Nyquist frequency maps to an FFT bin past the end of ``fbank`` and the
# function died on a raw ``IndexError`` thrown from inside the inner loop
# ("index 257 is out of bounds for axis 1 with size 257") -- a message naming
# neither the offending argument nor the constraint it broke. Every other
# guard in this module states the problem in the caller's terms; this one
# was simply missing.
nyquist = sample_rate / 2
assert num_filters >= 1, f"num_filters must be >= 1, got {num_filters}"
assert low_freq >= 0, f"low_freq must be >= 0, got {low_freq}"
assert low_freq < high_freq, f"low_freq ({low_freq}) must be below high_freq ({high_freq})"
assert high_freq <= nyquist, (
f"high_freq ({high_freq} Hz) exceeds the Nyquist frequency ({nyquist:g} Hz) "
f"for sample_rate={sample_rate}. A filter bank cannot describe energy above "
f"Nyquist because the FFT does not represent it."
)
# Convert frequencies to the Mel scale
low_mel = hz_to_mel(low_freq)
high_mel = hz_to_mel(high_freq)
mel_points = np.linspace(low_mel, high_mel, num_filters + 2) # Equally spaced in Mel scale
# Convert Mel frequencies back to Hz
hz_points = mel_to_hz(mel_points)
# Convert Hz frequencies to FFT bin indices
bin_points = np.floor((n_fft + 1) * hz_points / sample_rate).astype(np.int32)
# Create the Mel filter bank
fbank = np.zeros((num_filters, int(np.floor(n_fft / 2 + 1))))
for i in range(1, num_filters + 1):
f_m_minus = bin_points[i - 1] # Left
f_m = bin_points[i] # Center
f_m_plus = bin_points[i + 1] # Right
# Construct the filters
for j in range(f_m_minus, f_m):
fbank[i - 1, j] = (j - f_m_minus) / (f_m - f_m_minus)
for j in range(f_m, f_m_plus):
fbank[i - 1, j] = (f_m_plus - j) / (f_m_plus - f_m)
# Warn when the requested resolution exceeds what the FFT grid can carry.
# ``bin_points`` is floored to integers, so with many filters over a narrow
# band neighbouring edges collapse onto the same bin, both inner loops run
# over an empty range, and the filter comes out entirely zero. It then
# contributes nothing to the MFCCs -- the caller silently gets fewer
# effective filters than requested, with no error and no clue.
empty = int((fbank.sum(axis=1) == 0).sum())
if empty:
osh.warning(
f"{empty} of {num_filters} mel filters are empty: the {num_filters}-filter "
f"grid is finer than n_fft={n_fft} can resolve between {low_freq} and "
f"{high_freq} Hz. Lower num_filters or raise n_fft."
)
return fbank
[docs]
def mfcc(
signal: np.ndarray,
sample_rate: int,
num_mfcc: int = 13,
n_fft: int = 512,
num_filters: int = 26,
low_freq: int = 0,
high_freq: int | None = None,
) -> np.ndarray:
"""
Compute Mel-frequency Cepstral Coefficients (MFCC) for an audio signal.
Parameters
----------
signal : np.ndarray
The input audio signal as a 1D NumPy array.
sample_rate : int
The sample rate of the audio signal (in Hz).
num_mfcc : int, optional
The number of MFCC features to return, by default 13.
n_fft : int, optional
The FFT size to use, by default 512.
num_filters : int, optional
The number of Mel filters to use, by default 26.
low_freq : int, optional
The lowest frequency to consider in the Mel filter bank, by default 0 Hz.
high_freq : int, optional
The highest frequency to consider in the Mel filter bank, by default None (set to half the sample rate).
Returns
-------
np.ndarray
A 2D NumPy array containing the computed MFCC features for each frame.
"""
# 1. Pre-emphasis filter: Apply a high-pass filter to amplify high frequencies
pre_emphasis = 0.97
emphasized_signal = np.append(signal[0], signal[1:] - pre_emphasis * signal[:-1])
# 2. Framing: Split the signal into overlapping frames of 25ms (default)
frame_size = 0.025 # 25 milliseconds per frame
frame_stride = 0.01 # 10 milliseconds between consecutive frames
frame_length = int(round(frame_size * sample_rate)) # Convert frame length to samples
frame_step = int(round(frame_stride * sample_rate)) # Convert frame step to samples
# Compute the number of frames and pad the signal to fit exact frames
num_frames = int(np.ceil(float(len(emphasized_signal) - frame_length) / frame_step)) + 1
pad_signal_length = num_frames * frame_step + frame_length
pad_signal = np.append(emphasized_signal, np.zeros(pad_signal_length - len(emphasized_signal)))
# Create an index array for all frames and extract frames
indices = (
np.tile(np.arange(0, frame_length), (num_frames, 1))
+ np.tile(np.arange(0, num_frames * frame_step, frame_step), (frame_length, 1)).T
)
frames = pad_signal[indices.astype(np.int32, copy=False)]
# 3. Windowing: Apply a Hamming window to reduce spectral leakage
frames *= get_window("hamming", frame_length)
# 4. FFT and Power Spectrum: Compute the FFT and power spectrum for each frame
mag_frames = np.absolute(np.fft.rfft(frames, n_fft)) # Magnitude of the FFT
pow_frames = (1.0 / n_fft) * (mag_frames**2) # Power spectrum
# 5. Mel Filter Banks: Convert the power spectrum into the Mel scale
high_freq = (
high_freq or sample_rate / 2
) # If not provided, use half the sample rate (Nyquist frequency)
mel_filters = mel_filter_banks(num_filters, n_fft, sample_rate, low_freq, high_freq)
filter_banks = np.dot(pow_frames, mel_filters.T)
filter_banks = np.where(filter_banks == 0, np.finfo(float).eps, filter_banks) # Avoid log(0)
filter_banks = 20 * np.log10(filter_banks) # Convert to decibels (logarithmic scale)
# 6. DCT: Compute the Discrete Cosine Transform (DCT) of the log Mel filter banks
mfccs = dct(filter_banks, type=2, axis=1, norm="ortho")[:, :num_mfcc]
return mfccs
[docs]
def sound_resemblance(audio_file_1: str, audio_file_2: str) -> float:
"""
Compute the resemblance score between two audio files using Mel-frequency Cepstral Coefficients (MFCC).
Measure the resemblance between two audio files using the correlation coefficient.
Score is between 0 and 1.
The closer to 1, the more similar the signals.
The closer to 0, the more different the signals.
Parameters
----------
audio_file_1 : str
Path to the first audio file.
audio_file_2 : str
Path to the second audio file.
Returns
-------
float
The resemblance score between the two audio files based on MFCC.
Notes
-----
The function computes the resemblance score between two audio files using the cosine similarity of their MFCC features.
"""
sample_rate = 24000
# Load audio files, convert to mono and numpy arrays, resample to target sample rate
audio_1, _ = load_audio(
audio_file_1, to_numpy=True, to_mono=True, target_sample_rate=sample_rate
)
audio_2, _ = load_audio(
audio_file_2, to_numpy=True, to_mono=True, target_sample_rate=sample_rate
)
max_len = max(len(audio_1), len(audio_2))
audio_1 = np.pad(audio_1, (0, max_len - len(audio_1)))
audio_2 = np.pad(audio_2, (0, max_len - len(audio_2)))
mfcc1 = mfcc(audio_1, sample_rate)
mfcc2 = mfcc(audio_2, sample_rate)
a = np.abs(np.dot(mfcc1.ravel(), mfcc2.ravel()))
b = np.sqrt(np.dot(mfcc1.ravel(), mfcc1.ravel()) * np.dot(mfcc2.ravel(), mfcc2.ravel()))
if b == 0:
return 0.0 # by convention (one of the signals is made of zeros)
score = a / b
return score