Create utils.py

lib/vc/utils.py

@@ -0,0 +1,84 @@
+import os
+import wave
+import subprocess
+import yt_dlp
+import ffmpeg
+import logging
+from fairseq import checkpoint_utils
+logger = logging.getLogger(__name__)
+
+def load_hubert(config):
+    path_check = os.path.exists("assets/hubert/hubert_base.pt")
+    if path_check is False:
+        logger.warn("hubert_base.pt is missing. Please check the documentation for to get it.")
+    else:
+        logger.info("hubert_base.pt found.")
+    models, _, _ = checkpoint_utils.load_model_ensemble_and_task(
+        [os.path.join("assets", "hubert", "hubert_base.pt")],
+        suffix="",
+    )
+    hubert_model = models[0]
+    hubert_model = hubert_model.to(config.device)
+    if config.is_half:
+        hubert_model = hubert_model.half()
+    else:
+        hubert_model = hubert_model.float()
+    hubert_model.eval()
+    return hubert_model
+
+def download_audio(url, audio_provider):
+    logs = []
+    if url == "":
+        logs.append("URL required!")
+        yield None, "\n".join(logs)
+        return None, "\n".join(logs)
+    if not os.path.exists("yt"):
+        os.mkdir("yt")
+    if audio_provider == "Youtube":
+        logs.append("Downloading the audio...")
+        yield None, "\n".join(logs)
+        ydl_opts = {
+            'noplaylist': True,
+            'format': 'bestaudio/best',
+            'postprocessors': [{
+                'key': 'FFmpegExtractAudio',
+                'preferredcodec': 'wav',
+            }],
+            "outtmpl": 'yt/audio',
+        }
+        audio_path = "yt/audio.wav"
+        with yt_dlp.YoutubeDL(ydl_opts) as ydl:
+            ydl.download([url])
+        logs.append("Download Complete.")
+        yield audio_path, "\n".join(logs)
+
+def cut_vocal_and_inst(split_model):
+    logs = []
+    logs.append("Starting the audio splitting process...")
+    yield "\n".join(logs), None, None, None
+    command = f"demucs --two-stems=vocals -n {split_model} yt/audio.wav -o output"
+    result = subprocess.Popen(command.split(), stdout=subprocess.PIPE, text=True)
+    for line in result.stdout:
+        logs.append(line)
+        yield "\n".join(logs), None, None, None
+    logger.info(result.stdout)
+    vocal = f"output/{split_model}/audio/vocals.wav"
+    inst = f"output/{split_model}/audio/no_vocals.wav"
+    logs.append("Audio splitting complete.")
+    yield "\n".join(logs), vocal, inst, vocal
+
+def combine_vocal_and_inst(audio_data, vocal_volume, inst_volume, split_model):
+    if not os.path.exists("output/result"):
+        os.mkdir("output/result")
+    vocal_path = "output/result/output.wav"
+    output_path = "output/result/combine.mp3"
+    inst_path = f"output/{split_model}/audio/no_vocals.wav"
+    with wave.open(vocal_path, "w") as wave_file:
+        wave_file.setnchannels(1) 
+        wave_file.setsampwidth(2)
+        wave_file.setframerate(audio_data[0])
+        wave_file.writeframes(audio_data[1].tobytes())
+    command =  f'ffmpeg -y -i {inst_path} -i {vocal_path} -filter_complex [0:a]volume={inst_volume}[i];[1:a]volume={vocal_volume}[v];[i][v]amix=inputs=2:duration=longest[a] -map [a] -b:a 320k -c:a libmp3lame {output_path}'
+    result = subprocess.run(command.split(), stdout=subprocess.PIPE)
+    logger.info(result.stdout.decode())
+    return output_path