Audio files

kudio.core.io

Audio file I/O: path resolution, loading and saving of wave files.

Classes:

Name	Description
AudioInfo	What a file is, before you decide whether to load it.
ConvertResult	What convert_folder did. failed is (path, reason) pairs.
LoadAudio	Iterate over the .wav files found (recursively) in a directory,

Functions:

Name	Description
check_path	Return path if it is free (or exist_ok), otherwise an incremented
check_file	Normalize an output wave path: force the .wav suffix, create parent
check_input	Recursively collect .wav files from input.
file_load	Load a wave file; returns (waveform float32, sample_rate).
audio_info	Describe an audio file (or an in-memory waveform) without interpreting it.
load_wave	Load one wave file or a list of them into (waveform, sr) tuples.
load_waves	Convert every .wav in wave (see check_input) to waveform data.
save_wave	Write wave to an audio file via soundfile.
copy_waves	Copy wave files into directory path and return where they went.
convert_folder	Resample / re-encode / trim / normalise every .wav file under src.

AudioInfo dataclass

AudioInfo(
    sr: int,
    channels: int,
    frames: int,
    duration: float,
    peak: Optional[float],
    path: Optional[Path] = None,
    subtype: Optional[str] = None,
    format: Optional[str] = None,
)

What a file is, before you decide whether to load it.

duration is in seconds and peak is the largest absolute sample: nan when some samples are not finite, None when it was not measured (peak=False). subtype / format are soundfile's ('PCM_16', 'WAV') and are None when the info was taken from an array rather than a file.

ConvertResult dataclass

ConvertResult(
    written: int,
    total: int,
    outputs: List[Path],
    failed: List[Tuple[Path, str]],
)

What convert_folder did. failed is (path, reason) pairs.

LoadAudio

LoadAudio(waves_dirs: PathLike)

Iterate over the .wav files found (recursively) in a directory, collected the way check_input collects them.

Source code in kudio/core/io.py

def __init__(self, waves_dirs: PathLike):
    self.sr_pth = Path(waves_dirs)
    if not self.sr_pth.is_dir():
        raise NotADirectoryError(f"Folder not found: {self.sr_pth.absolute()}")
    self.files = check_input(self.sr_pth)[0]
    self.nf = len(self.files)

check_path

check_path(
    path: PathLike, exist_ok: bool = False, sep: str = ""
) -> str

Return path if it is free (or exist_ok), otherwise an incremented variant, e.g. runs/exp -> runs/exp2, runs/exp3 ...

Source code in kudio/core/io.py

def check_path(path: PathLike, exist_ok: bool = False, sep: str = '') -> str:
    """Return ``path`` if it is free (or ``exist_ok``), otherwise an incremented
    variant, e.g. ``runs/exp`` -> ``runs/exp2``, ``runs/exp3`` ...
    """
    path = Path(path)
    if exist_ok or not path.exists():
        return str(path)
    # listed rather than globbed: a name holding [ or * is a glob pattern,
    # and runs/exp[1] missed its own runs/exp[1]2 and handed it back
    pattern = re.compile(rf"{re.escape(path.name)}{re.escape(sep)}(\d+)")
    siblings = path.parent.iterdir() if path.parent.exists() else []
    indices = [int(m.group(1)) for m in (pattern.fullmatch(p.name) for p in siblings) if m]
    n = max(indices) + 1 if indices else 2
    while Path(f"{path}{sep}{n}").exists():
        n += 1
    return f"{path}{sep}{n}"

check_file

check_file(
    file_dir: PathLike, rename: bool = True, sep: str = ""
) -> str

Normalize an output wave path: force the .wav suffix, create parent directories, and (optionally) rename to avoid overwriting an existing file.

A folder becomes <folder>/unknown.wav. A path ending in a separator is taken as a folder even before it exists: Path drops the trailing slash, so "newdir/" would otherwise come back as newdir.wav.

Source code in kudio/core/io.py

def check_file(file_dir: PathLike, rename: bool = True, sep: str = '') -> str:
    """Normalize an output wave path: force the ``.wav`` suffix, create parent
    directories, and (optionally) rename to avoid overwriting an existing file.

    A folder becomes ``<folder>/unknown.wav``. A path ending in a separator is
    taken as a folder even before it exists: ``Path`` drops the trailing
    slash, so ``"newdir/"`` would otherwise come back as ``newdir.wav``.
    """
    spelled = str(file_dir)
    file_dir = Path(file_dir)
    if file_dir.is_dir() or spelled.endswith(('/', '\\')):
        file_dir = file_dir / 'unknown'
    file_dir = _as_wave_path(file_dir)
    if file_dir.exists():
        log.info("File already exists: %s", file_dir)
        if rename:
            file_dir = _free_wave_path(file_dir, sep)
            log.info("Renamed to: %s", file_dir)
        else:
            log.warning("Existing file will be overwritten: %s", file_dir)
    file_dir.parent.mkdir(exist_ok=True, parents=True)
    return str(file_dir)

check_input

check_input(input: Any) -> Tuple[List[Path], List[Any]]

Recursively collect .wav files from input.

input may be a file path, a directory, a .txt file containing one path per line, or a (nested) list/tuple of any of those.

Returns (right_list, wrong_list) where right_list holds resolved .wav paths and wrong_list everything that could not be resolved.

The suffix test ignores case, so a recorder's ZOOM0001.WAV counts whether it is named directly, listed, or found in a folder -- and a folder collects the same files on Windows as on Linux, where a plain rglob("*.wav") would not. A .txt list is read as UTF-8 (a BOM is fine), and relative paths in it resolve against the working directory.

Source code in kudio/core/io.py

def check_input(input: Any) -> Tuple[List[Path], List[Any]]:
    """Recursively collect ``.wav`` files from *input*.

    ``input`` may be a file path, a directory, a ``.txt`` file containing one
    path per line, or a (nested) list/tuple of any of those.

    Returns ``(right_list, wrong_list)`` where ``right_list`` holds resolved
    ``.wav`` paths and ``wrong_list`` everything that could not be resolved.

    The suffix test ignores case, so a recorder's ``ZOOM0001.WAV`` counts
    whether it is named directly, listed, or found in a folder -- and a folder
    collects the same files on Windows as on Linux, where a plain
    ``rglob("*.wav")`` would not. A ``.txt`` list is read as UTF-8 (a BOM is
    fine), and relative paths in it resolve against the working directory.
    """
    right_list: List[Path] = []
    wrong_list: List[Any] = []
    if isinstance(input, (list, tuple)):
        for item in input:
            r, w = check_input(item)
            right_list.extend(r)
            wrong_list.extend(w)
    elif input and isinstance(input, (str, PurePath)):
        input = Path(input)
        if input.is_file():
            if input.suffix.lower() == '.txt':
                text = input.read_text(encoding='utf-8-sig')
                lines = [x.strip() for x in text.strip().splitlines() if x.strip()]
                for item in lines:
                    r, w = check_input(item)
                    right_list.extend(r)
                    wrong_list.extend(w)
            elif input.suffix.lower() == WAVE_SUFFIX:
                right_list.append(input)
            else:
                wrong_list.append(str(input))
        elif input.is_dir():
            right_list.extend(sorted(p for p in input.rglob('*')
                                     if p.suffix.lower() == WAVE_SUFFIX and p.is_file()))
        else:
            wrong_list.append(str(input))
    else:
        wrong_list.append(input)
    return right_list, wrong_list

file_load

file_load(
    wav_dir: PathLike,
    sr: Optional[int] = None,
    mono: bool = True,
) -> Tuple[ndarray, int]

Load a wave file; returns (waveform float32, sample_rate).

sr=None keeps the file's native sample rate and reads through soundfile alone. A requested sr resamples through librosa.

Multi-channel audio (mono=False) comes back channels last, shaped (frames, channels) -- the layout soundfile uses and the one kudio.loudness expects. A mono file stays 1-D with mono=False; it does not become (frames, 1).

Raises:

Type	Description
ValueError	for an sr that is not a positive rate. That is a mistake in the call, so it is not dressed up as a bad file.
AudioIOError	when the file cannot be read.

Source code in kudio/core/io.py

def file_load(wav_dir: PathLike, sr: Optional[int] = None, mono: bool = True
              ) -> Tuple[np.ndarray, int]:
    """Load a wave file; returns ``(waveform float32, sample_rate)``.

    ``sr=None`` keeps the file's native sample rate and reads through
    soundfile alone. A requested ``sr`` resamples through librosa.

    Multi-channel audio (``mono=False``) comes back **channels last**, shaped
    ``(frames, channels)`` -- the layout soundfile uses and the one
    :func:`kudio.loudness` expects. A mono file stays 1-D with ``mono=False``;
    it does not become ``(frames, 1)``.

    :raises ValueError: for an *sr* that is not a positive rate. That is a
        mistake in the call, so it is not dressed up as a bad file.
    :raises AudioIOError: when the file cannot be read.
    """
    if sr is not None and sr <= 0:
        raise ValueError(f"sr must be a positive rate or None, got {sr!r}")
    try:
        if sr is None:
            y, rate = sf.read(str(wav_dir), dtype='float32', always_2d=False)
            if mono and y.ndim > 1:
                y = y.mean(axis=1).astype(np.float32)
            y = np.asarray(y, dtype=np.float32)
            _warn_non_finite(y, wav_dir)
            return y, int(rate)
        # resampling path (heavier)
        import librosa
        y, rate = librosa.load(str(wav_dir), sr=sr, mono=mono)
        if y.ndim > 1:
            # librosa hands back (channels, frames); the rest of kudio -- and
            # the sr=None branch above -- speak (frames, channels). Two layouts
            # from one function, chosen by an unrelated argument, is exactly
            # the kind of thing nobody notices until a model is training on
            # transposed audio.
            y = y.T
        y = np.ascontiguousarray(y, dtype=np.float32)
        _warn_non_finite(y, wav_dir)
        return y, int(rate)
    except Exception as e:
        raise AudioIOError(f"Could not load audio file: {wav_dir} ({e})") from e

resample

resample(
    y: ndarray,
    orig_sr: int,
    target_sr: int,
    res_type: str = "soxr_hq",
) -> ndarray

Resample a waveform already in memory.

The in-memory counterpart of file_load(..., sr=target_sr), which can only resample on the way in from disk. Returns y untouched when the rates already match, so it is safe to call unconditionally.

>>> y16 = kudio.resample(y8, orig_sr=8000, target_sr=16000)

A (frames, channels) array is resampled along its frames, the layout file_load returns. Integer input comes back in its own dtype, rounded and clipped to its range rather than truncated.

Source code in kudio/core/io.py

def resample(y: np.ndarray, orig_sr: int, target_sr: int,
             res_type: str = 'soxr_hq') -> np.ndarray:
    """Resample a waveform already in memory.

    The in-memory counterpart of ``file_load(..., sr=target_sr)``, which can
    only resample on the way in from disk. Returns *y* untouched when the rates
    already match, so it is safe to call unconditionally.

    >>> y16 = kudio.resample(y8, orig_sr=8000, target_sr=16000)

    A ``(frames, channels)`` array is resampled along its frames, the layout
    :func:`file_load` returns. Integer input comes back in its own dtype,
    rounded and clipped to its range rather than truncated.
    """
    if orig_sr <= 0 or target_sr <= 0:
        raise ValueError(f"sample rates must be > 0, got {orig_sr} -> {target_sr}")
    y = np.asarray(y)
    if orig_sr == target_sr or y.size == 0:
        return y
    if y.ndim > 2:
        raise ValueError(f"expected (frames,) or (frames, channels), got shape {y.shape}")
    import librosa
    # librosa resamples its last axis by default, which on a channels-last
    # array is the channel axis: a (16000, 2) clip came back (16000, 1)
    out = librosa.resample(y.astype(np.float32), orig_sr=orig_sr,
                           target_sr=target_sr, res_type=res_type, axis=0)
    if np.issubdtype(y.dtype, np.integer):
        limits = np.iinfo(y.dtype)
        return np.clip(np.round(out), limits.min, limits.max).astype(y.dtype)
    return out.astype(y.dtype, copy=False)

audio_info

audio_info(
    source: Union[PathLike, ndarray],
    sr: Optional[int] = None,
    peak: bool = True,
) -> AudioInfo

Describe an audio file (or an in-memory waveform) without interpreting it.

>>> info = kudio.audio_info("clip.wav")
>>> info.duration, info.sr, info.channels
(3.5, 16000, 1)

The header alone answers rate, channels, frames and duration, so peak=False never touches the samples and stays instant on a large file. The default reads them, because every caller so far wanted the peak too.

An array needs its sr passed in — a waveform does not carry one — and is read channels last, matching file_load.

Source code in kudio/core/io.py

def audio_info(source: Union[PathLike, np.ndarray], sr: Optional[int] = None,
               peak: bool = True) -> AudioInfo:
    """Describe an audio file (or an in-memory waveform) without interpreting it.

    >>> info = kudio.audio_info("clip.wav")
    >>> info.duration, info.sr, info.channels
    (3.5, 16000, 1)

    The header alone answers rate, channels, frames and duration, so
    ``peak=False`` never touches the samples and stays instant on a large file.
    The default reads them, because every caller so far wanted the peak too.

    An array needs its *sr* passed in — a waveform does not carry one — and is
    read **channels last**, matching :func:`file_load`.
    """
    if isinstance(source, np.ndarray):
        if sr is None:
            raise ValueError("audio_info(array) needs sr=")
        y = source
        frames = int(y.shape[0]) if y.size else 0
        channels = 1 if y.ndim == 1 else int(y.shape[1])
        top: Optional[float] = None
        if peak:
            # on the full scale a file's peak is read on; abs(-32768) is
            # -32768 in int16 besides
            top = _peak(as_float(y), "<array>")
        return AudioInfo(sr=int(sr), channels=channels, frames=frames,
                         duration=frames / float(sr) if sr else 0.0, peak=top)

    path = Path(source)
    try:
        meta = sf.info(str(path))
    except Exception as e:
        raise AudioIOError(f"Could not read audio header: {path} ({e})") from e

    top: Optional[float] = None
    if peak:
        top, bad = 0.0, 0
        try:
            with sf.SoundFile(str(path)) as fh:
                for block in fh.blocks(blocksize=1 << 20, dtype='float32'):
                    if block.size:
                        bad += _count_non_finite(block)
                        top = max(top, float(np.max(np.abs(block))))
        except Exception as e:
            raise AudioIOError(f"Could not read audio file: {path} ({e})") from e
        if bad:
            top = _undefined_peak(path, bad)

    return AudioInfo(sr=int(meta.samplerate), channels=int(meta.channels),
                     frames=int(meta.frames),
                     duration=float(meta.frames) / float(meta.samplerate)
                     if meta.samplerate else 0.0,
                     peak=top, path=path,
                     subtype=meta.subtype, format=meta.format)

load_wave

load_wave(
    wave: Any, num: Optional[int] = None
) -> Union[Tuple[ndarray, int], List[Tuple[ndarray, int]]]

Load one wave file or a list of them into (waveform, sr) tuples.

Source code in kudio/core/io.py

def load_wave(wave: Any, num: Optional[int] = None
              ) -> Union[Tuple[np.ndarray, int], List[Tuple[np.ndarray, int]]]:
    """Load one wave file or a list of them into ``(waveform, sr)`` tuples."""
    info = wave if num is None else wave[num]
    if isinstance(info, (list, tuple)):
        return [load_wave(w) for w in tqdm(info, desc="[kudio] Loading audios")]
    if isinstance(info, (str, PurePath)) and Path(info).is_file() \
            and Path(info).suffix.lower() == WAVE_SUFFIX:
        return file_load(info)
    raise ValueError(f"Unexpected wave input: {info!r}")

load_waves

load_waves(
    wave: Any,
    use_pool: bool = False,
    std_len: bool = False,
    core_: int = 8,
) -> Union[
    List[Tuple[ndarray, int]],
    Tuple[List[ndarray], int, int],
]

Convert every .wav in wave (see check_input) to waveform data.

Returns a list of (np.ndarray, int). With std_len=True returns (waves_2d, sr, min_len) where every waveform is truncated to the shortest length (all files must share one sample rate).

Source code in kudio/core/io.py

def load_waves(wave: Any, use_pool: bool = False, std_len: bool = False, core_: int = 8
               ) -> Union[List[Tuple[np.ndarray, int]], Tuple[List[np.ndarray], int, int]]:
    """Convert every ``.wav`` in *wave* (see :func:`check_input`) to waveform data.

    Returns a list of ``(np.ndarray, int)``. With ``std_len=True`` returns
    ``(waves_2d, sr, min_len)`` where every waveform is truncated to the
    shortest length (all files must share one sample rate).
    """
    waves, skipped = check_input(wave)
    for item in skipped:
        log.warning("Skipped, not an existing .wav file: %s", item)
    if std_len and not waves:
        raise AudioIOError(f"no audio files found in {wave!r}")
    if use_pool and len(waves) > 300:
        with Pool(core_) as pool:
            result = list(tqdm(pool.imap(file_load, map(str, waves)),
                               total=len(waves), desc="[kudio] Loading audios (pool)"))
    else:
        result = load_wave(waves)

    if std_len:
        w2d = [y for y, _ in result]
        srs = {sr for _, sr in result}
        if len(srs) != 1:
            raise ValueError(f"Different audio sample rates found: {sorted(srs)}")
        min_len = min(len(y) for y in w2d)
        w2d = [y[:min_len] for y in w2d]
        return w2d, srs.pop(), min_len

    return result

save_wave

save_wave(
    path: PathLike,
    wave: ndarray,
    sr: int,
    d_sample: Optional[int] = None,
    subtype: str = "PCM_16",
) -> None

Write wave to an audio file via soundfile.

The container follows the extension, the way soundfile picks it: a .wav path writes WAV, .flac writes FLAC.

Parameters:

Name	Type	Description	Default
subtype	str	soundfile subtype — 'PCM_16' (default), 'PCM_24', 'FLOAT' (32-bit float), ... see soundfile.available_subtypes().	'PCM_16'
d_sample	Optional[int]	optional resample rate before writing. Integer input is scaled by its own type's full scale, the way soundfile reads such a file back: int16 by 1 / 32768, int32 by 1 / 2**31, and unsigned types around their midpoint, so a 16-bit array is written unchanged. Float input is written as it is; DOUBLE keeps float64. A fixed-point subtype clips anything beyond ±1.0 to full scale, and every PCM_* subtype is rounded to the nearest step. Both are done here rather than left to libsndfile, whose conversion floors -- every sample half a step low -- and wraps around unless clipping has been switched on: 1.2 comes back as -0.8. Whether it has been is up to the soundfile release.	None

Raises:

Type	Description
AudioIOError	if any sample is NaN or inf, before anything is written or any folder created. Neither subtype family survives one: FLOAT keeps it, and an integer subtype writes NaN as a zero that reads back as silence, and inf as a full-scale sample.

Source code in kudio/core/io.py

def save_wave(path: PathLike, wave: np.ndarray, sr: int,
              d_sample: Optional[int] = None, subtype: str = 'PCM_16') -> None:
    """Write *wave* to an audio file via soundfile.

    The container follows the extension, the way soundfile picks it: a
    ``.wav`` path writes WAV, ``.flac`` writes FLAC.

    :param subtype: soundfile subtype — ``'PCM_16'`` (default), ``'PCM_24'``,
        ``'FLOAT'`` (32-bit float), ... see ``soundfile.available_subtypes()``.
    :param d_sample: optional resample rate before writing.

    Integer input is scaled by its own type's full scale, the way soundfile
    reads such a file back: int16 by ``1 / 32768``, int32 by ``1 / 2**31``,
    and unsigned types around their midpoint, so a 16-bit array is written
    unchanged. Float input is written as it is; ``DOUBLE`` keeps float64.

    A fixed-point *subtype* clips anything beyond ±1.0 to full scale, and every
    ``PCM_*`` subtype is rounded to the nearest step. Both are done here rather
    than left to libsndfile, whose conversion floors -- every sample half a
    step low -- and wraps around unless clipping has been switched on: 1.2
    comes back as -0.8. Whether it has been is up to the soundfile release.

    :raises AudioIOError: if any sample is NaN or inf, before anything is
        written or any folder created. Neither subtype family survives one:
        ``FLOAT`` keeps it, and an integer subtype writes NaN as a zero that
        reads back as silence, and inf as a full-scale sample.
    """
    wave = np.asarray(wave)
    precision = np.float64 if str(subtype).upper() == 'DOUBLE' else np.float32

    if np.issubdtype(wave.dtype, np.integer):
        limits = np.iinfo(wave.dtype)
        if limits.min == 0:                      # unsigned: silence is the middle
            middle = (int(limits.max) + 1) / 2.0
            wave = ((wave.astype(np.float64) - middle) / middle).astype(precision)
        else:
            wave = (wave.astype(np.float64) / -float(limits.min)).astype(precision)
    else:
        wave = wave.astype(precision)

    bad = _count_non_finite(wave)
    if bad:
        # Neither subtype family survives this. `FLOAT` writes the NaN out,
        # and every reader of that file gets nan back for the peak, the RMS
        # and the loudness; an integer subtype converts it to a zero, which is
        # worse, because the clip then reads as plain silence and nothing ever
        # says otherwise. One truncated file or one divide by a silent frame
        # upstream is enough, so it stops here with the name of the file.
        raise AudioIOError(
            f"refusing to write {path}: {bad} of {wave.size} sample(s) are "
            f"not finite (NaN or inf). Written as FLOAT they stay, and as an "
            f"integer subtype they become zeros that read back as silence. "
            f"Find what produced them -- a gain computed from a silent frame "
            f"is the usual one -- or drop the clip.")

    if d_sample is not None and d_sample != sr:
        wave = resample(wave, sr, d_sample)        # along the frames, not the channels
        sr = d_sample

    if _is_fixed_point(subtype):
        wave = np.clip(wave, -1.0, 1.0)
        bits = _PCM_BITS.get(str(subtype).upper())
        if bits:
            # Rounded here to the subtype's own step and handed over as the
            # integer type that holds it, shifted to the top bits. libsndfile
            # keeps the top bits of an integer, so nothing is floored on the
            # way out -- PCM_24 and PCM_U8 used to average a third and a half
            # of a step low.
            container = 16 if bits <= 16 else 32
            full = 2.0 ** (bits - 1)
            steps = np.clip(np.round(wave.astype(np.float64) * full), -full, full - 1)
            wave = (steps * 2.0 ** (container - bits)).astype(
                np.int16 if container == 16 else np.int32)

    Path(path).parent.mkdir(exist_ok=True, parents=True)
    try:
        sf.write(str(path), wave, sr, subtype=subtype)
    except Exception as e:
        raise AudioIOError(f"Could not write audio file: {path} ({e})") from e

copy_waves

copy_waves(path: PathLike, wave_: Any) -> List[Path]

Copy wave files into directory path and return where they went.

Note the order: the destination comes first. wave_ is anything check_input accepts -- a file, a folder, a .txt list, or a list of those. Files land flat in path; a name already taken there gets a free one (check_file), so two talkers' sx133.wav both arrive.

Source code in kudio/core/io.py

def copy_waves(path: PathLike, wave_: Any) -> List[Path]:
    """Copy wave files into directory *path* and return where they went.

    Note the order: the destination comes first. *wave_* is anything
    :func:`check_input` accepts -- a file, a folder, a ``.txt`` list, or a list
    of those. Files land flat in *path*; a name already taken there gets a free
    one (:func:`check_file`), so two talkers' ``sx133.wav`` both arrive.
    """
    path = Path(path)
    if path.suffix:
        log.warning("Destination is not a directory: %s", path)
        return []
    path.mkdir(exist_ok=True, parents=True)

    files, skipped = check_input(wave_)
    for item in skipped:
        log.warning("Not copied, not an existing .wav file: %s", item)
    copied = []
    for w in tqdm(files, desc=f"[kudio] Copying waves to {path.resolve()}",
                  disable=len(files) < 2):
        target = Path(check_file(path / w.name, rename=True))
        shutil.copy(str(w), target)
        copied.append(target)
    return copied

convert_folder

convert_folder(
    src: Any,
    dst: PathLike,
    *,
    sr: Optional[int] = None,
    mono: bool = True,
    subtype: str = "PCM_16",
    trim_db: Optional[float] = None,
    peak: Optional[float] = None,
    lufs: Optional[float] = None,
    overwrite: bool = False,
    progress: Optional[
        Callable[[int, int, Path], None]
    ] = None,
    on_error: str = "collect",
    root: Optional[PathLike] = None,
) -> ConvertResult

Resample / re-encode / trim / normalise every .wav file under src.

src is anything check_input accepts — a folder, a file, a .txt manifest, or a list of those. The output mirrors the input's directory structure below dst, so two files with the same name in different subfolders do not collide. A folder is mirrored from itself; a list or a manifest from the deepest folder its files share, or from root.

>>> kudio.convert_folder("raw/", "16k/", sr=16000, lufs=-23.0)
ConvertResult(written=412, total=412, ...)

Parameters:

Name	Type	Description	Default
trim_db	Optional[float]	top_db for kudio.trim_silence; None skips it.	None
peak	Optional[float]	peak-normalise target (e.g. 0.99).	None
lufs	Optional[float]	loudness-normalise target in LUFS (e.g. -23.0). Mutually exclusive with peak — normalising twice would undo the first one.	None
overwrite	bool	by default an existing output is renamed rather than replaced (check_file).	False
progress	Optional[Callable[[int, int, Path], None]]	optional callback(done, total, path).	None
on_error	str	'collect' (default) records the failure and carries on; 'raise' stops at the first one.	'collect'
root	Optional[PathLike]	the folder the output layout is measured from. Every input has to be inside it.	None

Source code in kudio/core/io.py

def convert_folder(src: Any, dst: PathLike, *,
                   sr: Optional[int] = None,
                   mono: bool = True,
                   subtype: str = 'PCM_16',
                   trim_db: Optional[float] = None,
                   peak: Optional[float] = None,
                   lufs: Optional[float] = None,
                   overwrite: bool = False,
                   progress: Optional[Callable[[int, int, Path], None]] = None,
                   on_error: str = 'collect',
                   root: Optional[PathLike] = None) -> ConvertResult:
    """Resample / re-encode / trim / normalise every ``.wav`` file under *src*.

    *src* is anything :func:`check_input` accepts — a folder, a file, a ``.txt``
    manifest, or a list of those. The output mirrors the input's directory
    structure below *dst*, so two files with the same name in different
    subfolders do not collide. A folder is mirrored from itself; a list or a
    manifest from the deepest folder its files share, or from *root*.

    >>> kudio.convert_folder("raw/", "16k/", sr=16000, lufs=-23.0)
    ConvertResult(written=412, total=412, ...)

    :param trim_db: ``top_db`` for :func:`kudio.trim_silence`; ``None`` skips it.
    :param peak: peak-normalise target (e.g. ``0.99``).
    :param lufs: loudness-normalise target in LUFS (e.g. ``-23.0``). Mutually
        exclusive with *peak* — normalising twice would undo the first one.
    :param overwrite: by default an existing output is renamed rather than
        replaced (:func:`check_file`).
    :param progress: optional ``callback(done, total, path)``.
    :param on_error: ``'collect'`` (default) records the failure and carries on;
        ``'raise'`` stops at the first one.
    :param root: the folder the output layout is measured from. Every input
        has to be inside it.
    """
    if peak is not None and lufs is not None:
        raise ValueError("pass peak= or lufs=, not both")
    if on_error not in ('collect', 'raise'):
        raise ValueError(f"on_error must be 'collect' or 'raise', got {on_error!r}")

    files, skipped = check_input(src)
    # a dst inside src would be collected on the next run and converted again,
    # one folder deeper each time
    out_abs = _abspath(dst)
    files = [f for f in files if out_abs not in _abspath(f).parents]
    if not files:
        raise AudioIOError(f"no audio files found in {src!r}")

    root = _source_root(src, files, root)
    out_root = Path(dst)
    out_root.mkdir(parents=True, exist_ok=True)

    outputs: List[Path] = []
    # an entry check_input could not use is a failure like any other, not
    # something to drop: a typo in a list used to leave "1/1 written"
    failed: List[Tuple[Path, str]] = [
        (Path(str(item)), "not an existing .wav file") for item in skipped]

    for n, f in enumerate(files, 1):
        try:
            y, rate = file_load(f, sr=sr, mono=mono)
            if trim_db is not None:
                from kudio.effects.silence import trim_silence
                y = trim_silence(y, top_db=trim_db)[0]
            if y.size == 0:
                raise AudioIOError("empty after trimming" if trim_db is not None
                                   else "the file holds no samples")
            if peak is not None:
                from kudio.effects.augment import normalize
                y = normalize(y, peak=peak)
            elif lufs is not None:
                from kudio.core.loudness import normalize_lufs
                y = normalize_lufs(y, rate, lufs=lufs)

            target = (out_root / _relative(f, root)).with_suffix(WAVE_SUFFIX)
            target = Path(check_file(target, rename=not overwrite))
            save_wave(target, y, rate, subtype=subtype)
            outputs.append(target)
        except Exception as e:
            if on_error == 'raise':
                raise
            failed.append((Path(f), f"{type(e).__name__}: {e}"))
        if progress is not None:
            progress(n, len(files), Path(f))

    return ConvertResult(written=len(outputs), total=len(files) + len(skipped),
                         outputs=outputs, failed=failed)

kudio.core.manager

In-memory registry of loaded/recorded audio clips used by the GUI.

Classes:

Name	Description
AudioDataManager	Keyed store of audio clips: {name: {dir, wave, sr, ch}}.

AudioDataManager

AudioDataManager()

Keyed store of audio clips: {name: {dir, wave, sr, ch}}.

wave is a list with one waveform per channel.

Methods:

Name	Description
add_local_file	Load dir into the manager as a mono downmix (ch 1), whatever

Source code in kudio/core/manager.py

def __init__(self):
    self.suffix = '.wav'
    self.rate_: Optional[int] = None
    self.ch_ = 1
    self.default_dict = {"dir": None, "wave": [], "sr": None, "ch": self.ch_}
    self.reset_defaults()

add_local_file

add_local_file(
    dir: PathLike, name: Optional[str] = None
) -> None

Load dir into the manager as a mono downmix (ch 1), whatever set_params says, so save_all writes it back as mono. Any format soundfile reads is fine.

Source code in kudio/core/manager.py

def add_local_file(self, dir: PathLike, name: Optional[str] = None) -> None:
    """Load *dir* into the manager as a mono downmix (``ch`` 1), whatever
    :meth:`set_params` says, so :meth:`save_all` writes it back as mono.
    Any format soundfile reads is fine."""
    dir = Path(dir)
    if not dir.is_file():
        raise FileNotFoundError(f"File not found: {dir}")
    name = dir.stem if name is None else name
    if name in self.get_names():
        log.warning("Manager: name %r already exists", name)
        return
    self.audios_[name] = deepcopy(self.default_dict)
    self.audios_[name]["dir"] = str(dir)
    wave_, rate_ = file_load(dir, mono=True)
    self.audios_[name]["wave"].append(wave_)
    self.audios_[name]["sr"] = rate_
    self.audios_[name]["ch"] = 1