Create app.py

app.py

@@ -0,0 +1,195 @@
+import gradio as gr
+import yt_dlp
+import ffmpeg
+import subprocess
+import numpy as np
+import librosa
+import soundfile
+from __future__ import unicode_literals
+
+class Slicer:
+    def __init__(self, sr, threshold=-40., min_length=5000, min_interval=300, hop_size=20, max_sil_kept=5000):
+        if not min_length >= min_interval >= hop_size:
+            raise ValueError('The following condition must be satisfied: min_length >= min_interval >= hop_size')
+        if not max_sil_kept >= hop_size:
+            raise ValueError('The following condition must be satisfied: max_sil_kept >= hop_size')
+        min_interval = sr * min_interval / 1000
+        self.threshold = 10 ** (threshold / 20.)
+        self.hop_size = round(sr * hop_size / 1000)
+        self.win_size = min(round(min_interval), 4 * self.hop_size)
+        self.min_length = round(sr * min_length / 1000 / self.hop_size)
+        self.min_interval = round(min_interval / self.hop_size)
+        self.max_sil_kept = round(sr * max_sil_kept / 1000 / self.hop_size)
+
+    def _apply_slice(self, waveform, begin, end):
+        if len(waveform.shape) > 1:
+            return waveform[:, begin * self.hop_size: min(waveform.shape[1], end * self.hop_size)]
+        else:
+            return waveform[begin * self.hop_size: min(waveform.shape[0], end * self.hop_size)]
+
+    def slice(self, waveform):
+        if len(waveform.shape) > 1:
+            samples = waveform.mean(axis=0)
+        else:
+            samples = waveform
+        if samples.shape[0] <= self.min_length:
+            return [waveform]
+        rms_list = get_rms(y=samples, frame_length=self.win_size, hop_length=self.hop_size).squeeze(0)
+        sil_tags = []
+        silence_start = None
+        clip_start = 0
+        for i, rms in enumerate(rms_list):
+            if rms < self.threshold:
+                if silence_start is None:
+                    silence_start = i
+                continue
+            if silence_start is None:
+                continue
+            is_leading_silence = silence_start == 0 and i > self.max_sil_kept
+            need_slice_middle = i - silence_start >= self.min_interval and i - clip_start >= self.min_length
+            if not is_leading_silence and not need_slice_middle:
+                silence_start = None
+                continue
+            if i - silence_start <= self.max_sil_kept:
+                pos = rms_list[silence_start: i + 1].argmin() + silence_start
+                if silence_start == 0:
+                    sil_tags.append((0, pos))
+                else:
+                    sil_tags.append((pos, pos))
+                clip_start = pos
+            elif i - silence_start <= self.max_sil_kept * 2:
+                pos = rms_list[i - self.max_sil_kept: silence_start + self.max_sil_kept + 1].argmin()
+                pos += i - self.max_sil_kept
+                pos_l = rms_list[silence_start: silence_start + self.max_sil_kept + 1].argmin() + silence_start
+                pos_r = rms_list[i - self.max_sil_kept: i + 1].argmin() + i - self.max_sil_kept
+                if silence_start == 0:
+                    sil_tags.append((0, pos_r))
+                    clip_start = pos_r
+                else:
+                    sil_tags.append((min(pos_l, pos), max(pos_r, pos)))
+                    clip_start = max(pos_r, pos)
+            else:
+                pos_l = rms_list[silence_start: silence_start + self.max_sil_kept + 1].argmin() + silence_start
+                pos_r = rms_list[i - self.max_sil_kept: i + 1].argmin() + i - self.max_sil_kept
+                if silence_start == 0:
+                    sil_tags.append((0, pos_r))
+                else:
+                    sil_tags.append((pos_l, pos_r))
+                clip_start = pos_r
+            silence_start = None
+        total_frames = rms_list.shape[0]
+        if silence_start is not None and total_frames - silence_start >= self.min_interval:
+            silence_end = min(total_frames, silence_start + self.max_sil_kept)
+            pos = rms_list[silence_start: silence_end + 1].argmin() + silence_start
+            sil_tags.append((pos, total_frames + 1))
+        if len(sil_tags) == 0:
+            return [waveform]
+        else:
+            chunks = []
+            if sil_tags[0][0] > 0:
+                chunks.append(self._apply_slice(waveform, 0, sil_tags[0][0]))
+            for i in range(len(sil_tags) - 1):
+                chunks.append(self._apply_slice(waveform, sil_tags[i][1], sil_tags[i + 1][0]))
+            if sil_tags[-1][1] < total_frames:
+                chunks.append(self._apply_slice(waveform, sil_tags[-1][1], total_frames))
+            return chunks
+
+def get_rms(y, frame_length=2048, hop_length=512, pad_mode="constant"):
+    padding = (int(frame_length // 2), int(frame_length // 2))
+    y = np.pad(y, padding, mode=pad_mode)
+    axis = -1
+    out_strides = y.strides + tuple([y.strides[axis]])
+    x_shape_trimmed = list(y.shape)
+    x_shape_trimmed[axis] -= frame_length - 1
+    out_shape = tuple(x_shape_trimmed) + tuple([frame_length])
+    xw = np.lib.stride_tricks.as_strided(y, shape=out_shape, strides=out_strides)
+    if axis < 0:
+        target_axis = axis - 1
+    else:
+        target_axis = axis + 1
+    xw = np.moveaxis(xw, -1, target_axis)
+    slices = [slice(None)] * xw.ndim
+    slices[axis] = slice(0, None, hop_length)
+    x = xw[tuple(slices)]
+    power = np.mean(np.abs(x) ** 2, axis=-2, keepdims=True)
+    return np.sqrt(power)
+
+def download_audio(dataset, url, drive_path, audio_name):
+    if dataset == "Drive":
+        return "Dataset is set to Drive. Skipping download."
+    elif dataset == "Youtube":
+        ydl_opts = {
+            'format': 'bestaudio/best',
+            'postprocessors': [{
+                'key': 'FFmpegExtractAudio',
+                'preferredcodec': 'wav',
+            }],
+            "outtmpl": f'/content/youtubeaudio/{audio_name}', 
+        }
+        def download_from_url(url):
+            with yt_dlp.YoutubeDL(ydl_opts) as ydl:
+                ydl.download([url])
+
+        download_from_url(url)
+        return f'Audio downloaded and saved as /content/youtubeaudio/{audio_name}.wav'
+
+def separate_audio(dataset, audio_name, drive_path):
+    if dataset == "Drive":
+        audio_input = drive_path
+    elif dataset == "Youtube":
+        audio_input = f"/content/youtubeaudio/{audio_name}.wav"
+    command = f"demucs --two-stems=vocals {audio_input}"
+    result = subprocess.run(command.split(), stdout=subprocess.PIPE)
+    subprocess.run(f"!mkdir -p /content/audio/{audio_name}", shell=True)
+    subprocess.run(f"!cp -r /content/separated/htdemucs/{audio_name}/* /content/audio/{audio_name}", shell=True)
+    if dataset == "Youtube":
+        subprocess.run(f"!cp -r /content/youtubeaudio/{audio_name}.wav /content/audio/{audio_name}", shell=True)
+    return result.stdout.decode()
+
+def split_audio(audio_name):
+    audio, sr = librosa.load(f'/content/separated/htdemucs/{audio_name}/vocals.wav', sr=None, mono=False)
+    slicer = Slicer(
+        sr=sr,
+        threshold=-40,
+        min_length=5000,
+        min_interval=500,
+        hop_size=10,
+        max_sil_kept=500
+    )
+    chunks = slicer.slice(audio)
+    subprocess.run(f"!mkdir -p /content/dataset/{audio_name}", shell=True)
+    for i, chunk in enumerate(chunks):
+        if len(chunk.shape) > 1:
+            chunk = chunk.T
+        soundfile.write(f'/content/dataset/{audio_name}/split_{i}.wav', chunk, sr)
+    subprocess.run(f"!mkdir -p /content/dataset/{audio_name}", shell=True)
+    subprocess.run(f"!cp -r /content/dataset/{audio_name}/* /content/dataset/{audio_name}", shell=True)
+    return "Audio split into chunks and saved."
+
+def process_audio(mode, dataset, url, drive_path, audio_name):
+    download_result = download_audio(dataset, url, drive_path, audio_name)
+    if "Skipping download" not in download_result:
+        separate_result = separate_audio(dataset, audio_name, drive_path)
+        if mode == "Splitting":
+            split_result = split_audio(audio_name)
+            return f"{download_result}\n{separate_result}\n{split_result}"
+        else:
+            return f"{download_result}\n{separate_result}\nMode is set to Separate. Skipping splitting."
+    else:
+        return download_result
+
+demo = gr.Interface(
+    fn=process_audio,
+    inputs=[
+        gr.Dropdown(choices=["Splitting", "Separate"], label="Mode"),
+        gr.Dropdown(choices=["Youtube", "Drive"], label="Dataset"),
+        gr.Textbox(label="URL"),
+        gr.Textbox(label="Drive Path"),
+        gr.Textbox(label="Audio Name"),
+    ],
+    outputs="text",
+    title="Dataset Maker",
+    description="Process audio from Youtube or Drive and split it based on silence detection."
+)
+
+demo.launch()