File size: 12,729 Bytes
5238467
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
23fe483
 
5238467
 
 
 
 
 
 
23fe483
5238467
 
 
 
 
 
 
 
 
 
 
 
 
23fe483
 
5238467
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
23fe483
 
5238467
 
 
 
 
 
 
 
 
 
 
 
 
 
 
23fe483
5238467
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
23fe483
5238467
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
d7ef5a5
1028cad
b76a81b
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1028cad
b76a81b
 
 
1028cad
 
b76a81b
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1028cad
 
 
 
b76a81b
1028cad
b76a81b
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
d7ef5a5
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1028cad
 
790c9e0
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
# Copyright (c) Meta Platforms, Inc. and affiliates.
# All rights reserved.
#
# This source code is licensed under the license found in the
# LICENSE file in the root directory of this source tree.

import sys
import typing as tp

import julius
import torch
import torchaudio


def convert_audio_channels(wav: torch.Tensor, channels: int = 2) -> torch.Tensor:
    """Convert audio to the given number of channels.

    Args:
        wav (torch.Tensor): Audio wave of shape [B, C, T].
        channels (int): Expected number of channels as output.
    Returns:
        torch.Tensor: Downmixed or unchanged audio wave [B, C, T].
    """
    *shape, src_channels, length = wav.shape
    if src_channels == channels:
        pass
    elif channels == 1:
        # Case 1:
        # The caller asked 1-channel audio, and the stream has multiple
        # channels, downmix all channels.
        wav = wav.mean(dim=-2, keepdim=True)
    elif src_channels == 1:
        # Case 2:
        # The caller asked for multiple channels, but the input file has
        # a single channel, replicate the audio over all channels.
        wav = wav.expand(*shape, channels, length)
    elif src_channels >= channels:
        # Case 3:
        # The caller asked for multiple channels, and the input file has
        # more channels than requested. In that case return the first channels.
        wav = wav[..., :channels, :]
    else:
        # Case 4: What is a reasonable choice here?
        raise ValueError('The audio file has less channels than requested but is not mono.')
    return wav


def convert_audio(wav: torch.Tensor, from_rate: float,
                  to_rate: float, to_channels: int) -> torch.Tensor:
    """Convert audio to new sample rate and number of audio channels.
    """
    wav = julius.resample_frac(wav, int(from_rate), int(to_rate))
    wav = convert_audio_channels(wav, to_channels)
    return wav


def normalize_loudness(wav: torch.Tensor, sample_rate: int, loudness_headroom_db: float = 14,
                       loudness_compressor: bool = False, energy_floor: float = 2e-3):
    """Normalize an input signal to a user loudness in dB LKFS.
    Audio loudness is defined according to the ITU-R BS.1770-4 recommendation.

    Args:
        wav (torch.Tensor): Input multichannel audio data.
        sample_rate (int): Sample rate.
        loudness_headroom_db (float): Target loudness of the output in dB LUFS.
        loudness_compressor (bool): Uses tanh for soft clipping.
        energy_floor (float): anything below that RMS level will not be rescaled.
    Returns:
        output (torch.Tensor): Loudness normalized output data.
    """
    energy = wav.pow(2).mean().sqrt().item()
    if energy < energy_floor:
        return wav
    transform = torchaudio.transforms.Loudness(sample_rate)
    input_loudness_db = transform(wav).item()
    # calculate the gain needed to scale to the desired loudness level
    delta_loudness = -loudness_headroom_db - input_loudness_db
    gain = 10.0 ** (delta_loudness / 20.0)
    output = gain * wav
    if loudness_compressor:
        output = torch.tanh(output)
    assert output.isfinite().all(), (input_loudness_db, wav.pow(2).mean().sqrt())
    return output


def _clip_wav(wav: torch.Tensor, log_clipping: bool = False, stem_name: tp.Optional[str] = None) -> None:
    """Utility function to clip the audio with logging if specified."""
    max_scale = wav.abs().max()
    if log_clipping and max_scale > 1:
        clamp_prob = (wav.abs() > 1).float().mean().item()
        print(f"CLIPPING {stem_name or ''} happening with proba (a bit of clipping is okay):",
              clamp_prob, "maximum scale: ", max_scale.item(), file=sys.stderr)
    wav.clamp_(-1, 1)


def normalize_audio(wav: torch.Tensor, normalize: bool = True,
                    strategy: str = 'peak', peak_clip_headroom_db: float = 1,
                    rms_headroom_db: float = 18, loudness_headroom_db: float = 14,
                    loudness_compressor: bool = False, log_clipping: bool = False,
                    sample_rate: tp.Optional[int] = None,
                    stem_name: tp.Optional[str] = None) -> torch.Tensor:
    """Normalize the audio according to the prescribed strategy (see after).

    Args:
        wav (torch.Tensor): Audio data.
        normalize (bool): if `True` (default), normalizes according to the prescribed
            strategy (see after). If `False`, the strategy is only used in case clipping
            would happen.
        strategy (str): Can be either 'clip', 'peak', or 'rms'. Default is 'peak',
            i.e. audio is normalized by its largest value. RMS normalizes by root-mean-square
            with extra headroom to avoid clipping. 'clip' just clips.
        peak_clip_headroom_db (float): Headroom in dB when doing 'peak' or 'clip' strategy.
        rms_headroom_db (float): Headroom in dB when doing 'rms' strategy. This must be much larger
            than the `peak_clip` one to avoid further clipping.
        loudness_headroom_db (float): Target loudness for loudness normalization.
        loudness_compressor (bool): If True, uses tanh based soft clipping.
        log_clipping (bool): If True, basic logging on stderr when clipping still
            occurs despite strategy (only for 'rms').
        sample_rate (int): Sample rate for the audio data (required for loudness).
        stem_name (Optional[str]): Stem name for clipping logging.
    Returns:
        torch.Tensor: Normalized audio.
    """
    scale_peak = 10 ** (-peak_clip_headroom_db / 20)
    scale_rms = 10 ** (-rms_headroom_db / 20)
    if strategy == 'peak':
        rescaling = (scale_peak / wav.abs().max())
        if normalize or rescaling < 1:
            wav = wav * rescaling
    elif strategy == 'clip':
        wav = wav.clamp(-scale_peak, scale_peak)
    elif strategy == 'rms':
        mono = wav.mean(dim=0)
        rescaling = scale_rms / mono.pow(2).mean().sqrt()
        if normalize or rescaling < 1:
            wav = wav * rescaling
        _clip_wav(wav, log_clipping=log_clipping, stem_name=stem_name)
    elif strategy == 'loudness':
        assert sample_rate is not None, "Loudness normalization requires sample rate."
        wav = normalize_loudness(wav, sample_rate, loudness_headroom_db, loudness_compressor)
        _clip_wav(wav, log_clipping=log_clipping, stem_name=stem_name)
    else:
        assert wav.abs().max() < 1
        assert strategy == '' or strategy == 'none', f"Unexpected strategy: '{strategy}'"
    return wav


def f32_pcm(wav: torch.Tensor) -> torch.Tensor:
    """Convert audio to float 32 bits PCM format.
    """
    if wav.dtype.is_floating_point:
        return wav
    else:
        assert wav.dtype == torch.int16
        return wav.float() / 2**15


def i16_pcm(wav: torch.Tensor) -> torch.Tensor:
    """Convert audio to int 16 bits PCM format.

    ..Warning:: There exist many formula for doing this convertion. None are perfect
    due to the asymetry of the int16 range. One either have possible clipping, DC offset,
    or inconsistancies with f32_pcm. If the given wav doesn't have enough headroom,
    it is possible that `i16_pcm(f32_pcm)) != Identity`.
    """
    if wav.dtype.is_floating_point:
        assert wav.abs().max() <= 1
        candidate = (wav * 2 ** 15).round()
        if candidate.max() >= 2 ** 15:  # clipping would occur
            candidate = (wav * (2 ** 15 - 1)).round()
        return candidate.short()
    else:
        assert wav.dtype == torch.int16
        return wav

def apply_tafade(audio: torch.Tensor, sample_rate, duration=3.0, out=True, start=True, shape: str = "linear", stem_name: tp.Optional[str] = None) -> torch.Tensor:
    """
    Apply fade-in and/or fade-out effects to the audio tensor.

    Args:
        audio (torch.Tensor): The input audio tensor of shape (C, L).
        sample_rate (int): The sample rate of the audio.
        duration (float, optional): The duration of the fade in seconds. Defaults to 3.0.
        out (bool, optional): Determines whether to apply fade-in (False) or fade-out (True) effect. Defaults to True.
        start (bool, optional): Determines whether the fade is applied to the beginning (True) or end (False) of the audio. Defaults to True.
        shape (str, optional): The shape of the fade. Must be one of: "quarter_sine", "half_sine", "linear", "logarithmic", "exponential". Defaults to "linear".

    Returns:
        torch.Tensor: The audio tensor with the fade effect applied.

    """
    fade_samples = int(sample_rate * duration)  # Number of samples for the fade duration

    # Create the fade transform
    fade_transform = torchaudio.transforms.Fade(fade_in_len=0, fade_out_len=0, fade_shape=shape)

    if out:
        fade_transform.fade_out_len = fade_samples
    else:
        fade_transform.fade_in_len = fade_samples        

    # Select the portion of the audio to apply the fade
    if start:
        audio_fade_section = audio[:, :fade_samples]
    else:
        audio_fade_section = audio[:, -fade_samples:]

    # Apply the fade transform to the audio section
    audio_faded = fade_transform(audio)

    # Replace the selected portion of the audio with the faded section
    if start:
        audio_faded[:, :fade_samples] = audio_fade_section
    else:
        audio_faded[:, -fade_samples:] = audio_fade_section

    wav = normalize_loudness(audio_faded,sample_rate, loudness_headroom_db=18, loudness_compressor=True)
    _clip_wav(wav, log_clipping=False, stem_name=stem_name)
    return wav


def apply_fade(audio: torch.Tensor, sample_rate, duration=3.0, out=True, start=True, curve_start:float=0.0, curve_end:float=1.0, current_device:str="cpu", stem_name: tp.Optional[str] = None) -> torch.Tensor:
    """
    Apply fade-in and/or fade-out effects to the audio tensor.

    Args:
        audio (torch.Tensor): The input audio tensor of shape (C, L).
        sample_rate (int): The sample rate of the audio.
        duration (float, optional): The duration of the fade in seconds. Defaults to 3.0.
        out (bool, optional): Determines whether to apply fade-in (False) or fade-out (True) effect. Defaults to True.
        start (bool, optional): Determines whether the fade is applied to the beginning (True) or end (False) of the audio. Defaults to True.
        curve_start (float, optional): The starting amplitude of the fade curve. Defaults to 0.0.
        curve_end (float, optional): The ending amplitude of the fade curve. Defaults to 1.0.
        current_device (str, optional): The device on which the fade curve tensor should be created. Defaults to "cpu".

    Returns:
        torch.Tensor: The audio tensor with the fade effect applied.

    """
    fade_samples = int(sample_rate * duration)  # Number of samples for the fade duration
    fade_curve = torch.linspace(curve_start, curve_end, fade_samples, device=current_device)  # Generate linear fade curve

    if out:
        fade_curve = fade_curve.flip(0)  # Reverse the fade curve for fade out

    # Select the portion of the audio to apply the fade
    if start:
        audio_fade_section = audio[:, :fade_samples]
    else:
        audio_fade_section = audio[:, -fade_samples:]

    # Apply the fade curve to the audio section
    audio_faded = audio.clone()
    audio_faded[:, :fade_samples] *= fade_curve.unsqueeze(0)
    audio_faded[:, -fade_samples:] *= fade_curve.unsqueeze(0)

    # Replace the selected portion of the audio with the faded section
    if start:
        audio_faded[:, :fade_samples] = audio_fade_section
    else:
        audio_faded[:, -fade_samples:] = audio_fade_section

    wav = normalize_loudness(audio_faded,sample_rate, loudness_headroom_db=18, loudness_compressor=True)
    _clip_wav(wav, log_clipping=False, stem_name=stem_name)
    return wav

def apply_splice_effect(waveform1, sample_rate1, waveform2, sample_rate2, overlap):
    # Convert sample rates to integers
    sample_rate1 = int(sample_rate1)
    sample_rate2 = int(sample_rate2)

    # Convert tensors to mono-channel if needed
    if waveform1.ndim > 2:
        waveform1 = waveform1.mean(dim=1)
    if waveform2.ndim > 2:
        waveform2 = waveform2.mean(dim=1)

    ## Convert tensors to numpy arrays
    #waveform1_np = waveform1.numpy()
    #waveform2_np = waveform2.numpy()

    # Apply splice effect using torchaudio.sox_effects.apply_effects_tensor
    effects = [
        ["splice", f"-q {waveform1},{overlap}"],
    ]
    output_waveform, output_sample_rate = torchaudio.sox_effects.apply_effects_tensor(
        torch.cat([waveform1.unsqueeze(0), waveform2.unsqueeze(0)], dim=2),
        sample_rate1,
        effects
    )

    return output_waveform.squeeze(0), output_sample_rate