| | import argparse |
| | import gc |
| | import hashlib |
| | import json |
| | import os |
| | import shlex |
| | import subprocess |
| | from contextlib import suppress |
| | from urllib.parse import urlparse, parse_qs |
| |
|
| | import gradio as gr |
| | import librosa |
| | import numpy as np |
| | import soundfile as sf |
| | import sox |
| | import yt_dlp |
| | from pedalboard import Pedalboard, Reverb, Compressor, HighpassFilter |
| | from pedalboard.io import AudioFile |
| | from pydub import AudioSegment |
| | from audio_separator.separator import Separator |
| | from rvc import Config, load_hubert, get_vc, rvc_infer |
| |
|
| | |
| | BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) |
| | mdxnet_models_dir = os.path.join(BASE_DIR, 'mdxnet_models') |
| | rvc_models_dir = os.path.join(BASE_DIR, 'rvc_models') |
| | output_dir = os.path.join(BASE_DIR, 'song_output') |
| |
|
| |
|
| | def get_youtube_video_id(url, ignore_playlist=True): |
| | """ |
| | Extract the YouTube video ID from various URL formats. |
| | |
| | Examples: |
| | http://youtu.be/SA2iWivDJiE |
| | http://www.youtube.com/watch?v=_oPAwA_Udwc&feature=feedu |
| | http://www.youtube.com/embed/SA2iWivDJiE |
| | http://www.youtube.com/v/SA2iWivDJiE?version=3&hl=en_US |
| | """ |
| | parsed_url = urlparse(url) |
| | hostname = parsed_url.hostname or '' |
| | path = parsed_url.path |
| |
|
| | if hostname.lower() == 'youtu.be': |
| | return path.lstrip('/') |
| |
|
| | if hostname.lower() in {'www.youtube.com', 'youtube.com', 'music.youtube.com'}: |
| | if not ignore_playlist: |
| | with suppress(KeyError): |
| | return parse_qs(parsed_url.query)['list'][0] |
| | if parsed_url.path == '/watch': |
| | return parse_qs(parsed_url.query).get('v', [None])[0] |
| | if parsed_url.path.startswith('/watch/'): |
| | return parsed_url.path.split('/')[1] |
| | if parsed_url.path.startswith('/embed/'): |
| | return parsed_url.path.split('/')[2] |
| | if parsed_url.path.startswith('/v/'): |
| | return parsed_url.path.split('/')[2] |
| |
|
| | return None |
| |
|
| |
|
| | def yt_download(link): |
| | """ |
| | Download the audio from a YouTube link as an mp3 file. |
| | """ |
| | ydl_opts = { |
| | 'format': 'bestaudio', |
| | 'outtmpl': '%(title)s', |
| | 'nocheckcertificate': True, |
| | 'ignoreerrors': True, |
| | 'no_warnings': True, |
| | 'quiet': True, |
| | 'extractaudio': True, |
| | 'postprocessors': [{ |
| | 'key': 'FFmpegExtractAudio', |
| | 'preferredcodec': 'mp3' |
| | }], |
| | } |
| | with yt_dlp.YoutubeDL(ydl_opts) as ydl: |
| | result = ydl.extract_info(link, download=True) |
| | download_path = ydl.prepare_filename(result, outtmpl='%(title)s.mp3') |
| | return download_path |
| |
|
| |
|
| | def display_progress(message, percent, is_webui, progress=None): |
| | """ |
| | Display progress either via the provided progress callback or by printing. |
| | """ |
| | if is_webui and progress is not None: |
| | progress(percent, desc=message) |
| | else: |
| | print(message) |
| |
|
| |
|
| | def raise_exception(error_msg, is_webui): |
| | """ |
| | Raise an exception. If running in a web UI, use gr.Error. |
| | """ |
| | if is_webui: |
| | raise gr.Error(error_msg) |
| | else: |
| | raise Exception(error_msg) |
| |
|
| |
|
| | def get_rvc_model(voice_model, is_webui): |
| | """ |
| | Search the specified RVC model directory for the model (.pth) and index (.index) files. |
| | """ |
| | rvc_model_filename, rvc_index_filename = None, None |
| | model_dir = os.path.join(rvc_models_dir, voice_model) |
| | if not os.path.exists(model_dir): |
| | raise_exception(f'Model directory {model_dir} does not exist.', is_webui) |
| | for file in os.listdir(model_dir): |
| | ext = os.path.splitext(file)[1] |
| | if ext == '.pth': |
| | rvc_model_filename = file |
| | if ext == '.index': |
| | rvc_index_filename = file |
| |
|
| | if rvc_model_filename is None: |
| | error_msg = f'No model file exists in {model_dir}.' |
| | raise_exception(error_msg, is_webui) |
| |
|
| | model_path = os.path.join(model_dir, rvc_model_filename) |
| | index_path = os.path.join(model_dir, rvc_index_filename) if rvc_index_filename else '' |
| | return model_path, index_path |
| |
|
| |
|
| | def separation_uvr(filename, output): |
| | """ |
| | Run the separation steps using different pre-trained models. |
| | Returns a tuple of four file paths: |
| | - vocals_no_reverb: The vocals after initial de-echo/de-reverb (used as intermediate vocals) |
| | - instrumental_path: The separated instrumental audio |
| | - main_vocals_dereverb: The lead vocals after final de-reverb processing |
| | - backup_vocals: The backup vocals extracted in the final stage |
| | """ |
| | separator = Separator(output_dir=output) |
| | base_name = os.path.splitext(os.path.basename(filename))[0] |
| |
|
| | instrumental_path = os.path.join(output, f'{base_name}_Instrumental.wav') |
| | initial_vocals = os.path.join(output, f'{base_name}_Vocals.wav') |
| | vocals_no_reverb = os.path.join(output, f'{base_name}_Vocals (No Reverb).wav') |
| | vocals_reverb = os.path.join(output, f'{base_name}_Vocals (Reverb).wav') |
| | main_vocals_dereverb = os.path.join(output, f'{base_name}_Vocals_Main_DeReverb.wav') |
| | backup_vocals = os.path.join(output, f'{base_name}_Vocals_Backup.wav') |
| |
|
| | separator.load_model(model_filename='model_bs_roformer_ep_317_sdr_12.9755.ckpt') |
| | voc_inst = separator.separate(filename) |
| | os.rename(os.path.join(output, voc_inst[0]), instrumental_path) |
| | os.rename(os.path.join(output, voc_inst[1]), initial_vocals) |
| |
|
| | separator.load_model(model_filename='UVR-DeEcho-DeReverb.pth') |
| | voc_no_reverb = separator.separate(initial_vocals) |
| | os.rename(os.path.join(output, voc_no_reverb[0]), vocals_no_reverb) |
| | os.rename(os.path.join(output, voc_no_reverb[1]), vocals_reverb) |
| |
|
| | separator.load_model(model_filename='mel_band_roformer_karaoke_aufr33_viperx_sdr_10.1956.ckpt') |
| | voc_split = separator.separate(vocals_no_reverb) |
| | os.rename(os.path.join(output, voc_split[0]), backup_vocals) |
| | os.rename(os.path.join(output, voc_split[1]), main_vocals_dereverb) |
| |
|
| | if os.path.exists(vocals_reverb): |
| | os.remove(vocals_reverb) |
| |
|
| | return vocals_no_reverb, instrumental_path, main_vocals_dereverb, backup_vocals |
| |
|
| |
|
| | def get_audio_paths(song_dir): |
| | """ |
| | Search the given directory for expected audio files. |
| | Returns: |
| | orig_song_path, instrumentals_path, main_vocals_dereverb_path, backup_vocals_path |
| | """ |
| | orig_song_path = None |
| | instrumentals_path = None |
| | main_vocals_dereverb_path = None |
| | backup_vocals_path = None |
| |
|
| | for file in os.listdir(song_dir): |
| | if file.endswith('_Instrumental.wav'): |
| | instrumentals_path = os.path.join(song_dir, file) |
| | orig_song_path = instrumentals_path.replace('_Instrumental', '') |
| | elif file.endswith('_Vocals_Main_DeReverb.wav'): |
| | main_vocals_dereverb_path = os.path.join(song_dir, file) |
| | elif file.endswith('_Vocals_Backup.wav'): |
| | backup_vocals_path = os.path.join(song_dir, file) |
| |
|
| | return orig_song_path, instrumentals_path, main_vocals_dereverb_path, backup_vocals_path |
| |
|
| |
|
| | def convert_to_stereo(audio_path): |
| | """ |
| | Convert the given audio file to stereo (2 channels) if it is mono. |
| | """ |
| | wave, sr = librosa.load(audio_path, mono=False, sr=44100) |
| | if wave.ndim == 1: |
| | stereo_path = f'{os.path.splitext(audio_path)[0]}_stereo.wav' |
| | command = shlex.split(f'ffmpeg -y -loglevel error -i "{audio_path}" -ac 2 -f wav "{stereo_path}"') |
| | subprocess.run(command, check=True) |
| | return stereo_path |
| | return audio_path |
| |
|
| |
|
| | def pitch_shift(audio_path, pitch_change): |
| | """ |
| | Shift the pitch of the audio by the specified amount. |
| | """ |
| | output_path = f'{os.path.splitext(audio_path)[0]}_p{pitch_change}.wav' |
| | if not os.path.exists(output_path): |
| | y, sr = sf.read(audio_path) |
| | tfm = sox.Transformer() |
| | tfm.pitch(pitch_change) |
| | y_shifted = tfm.build_array(input_array=y, sample_rate_in=sr) |
| | sf.write(output_path, y_shifted, sr) |
| | return output_path |
| |
|
| |
|
| | def get_hash(filepath): |
| | """ |
| | Calculate a short BLAKE2b hash for the given file. |
| | """ |
| | with open(filepath, 'rb') as f: |
| | file_hash = hashlib.blake2b() |
| | while chunk := f.read(8192): |
| | file_hash.update(chunk) |
| | return file_hash.hexdigest()[:11] |
| |
|
| |
|
| | def preprocess_song(song_input, song_id, is_webui, input_type, progress): |
| | """ |
| | Preprocess the input song: |
| | - Download if YouTube URL. |
| | - Convert to stereo. |
| | - Separate vocals and instrumentals. |
| | Returns a tuple with six values matching the expected unpacking in the pipeline. |
| | """ |
| | if input_type == 'yt': |
| | display_progress('[~] Downloading song...', 0, is_webui, progress) |
| | song_link = song_input.split('&')[0] |
| | orig_song_path = yt_download(song_link) |
| | elif input_type == 'local': |
| | orig_song_path = song_input |
| | else: |
| | orig_song_path = None |
| |
|
| | song_output_dir = os.path.join(output_dir, song_id) |
| | if not os.path.exists(song_output_dir): |
| | os.makedirs(song_output_dir) |
| |
|
| | orig_song_path = convert_to_stereo(orig_song_path) |
| |
|
| | display_progress('[~] Separating Vocals from Instrumental...', 0.1, is_webui, progress) |
| | vocals_no_reverb, instrumental_path, main_vocals_dereverb, backup_vocals = separation_uvr(orig_song_path, song_output_dir) |
| | return orig_song_path, vocals_no_reverb, instrumental_path, main_vocals_dereverb, backup_vocals, main_vocals_dereverb |
| |
|
| |
|
| | def voice_change(voice_model, vocals_path, output_path, pitch_change, f0_method, |
| | index_rate, filter_radius, rms_mix_rate, protect, crepe_hop_length, is_webui): |
| | """ |
| | Convert the input vocals using the specified RVC model. |
| | """ |
| | rvc_model_path, rvc_index_path = get_rvc_model(voice_model, is_webui) |
| | device = 'cuda:0' |
| | config = Config(device, True) |
| | hubert_model = load_hubert(embedder_model="contentvec", embedder_model_custom=None) |
| | cpt, version, net_g, tgt_sr, vc = get_vc(device, config.is_half, config, rvc_model_path) |
| |
|
| | rvc_infer( |
| | rvc_index_path, index_rate, vocals_path, output_path, pitch_change, f0_method, |
| | cpt, version, net_g, filter_radius, tgt_sr, rms_mix_rate, protect, |
| | crepe_hop_length, vc, hubert_model |
| | ) |
| | del hubert_model, cpt |
| | gc.collect() |
| |
|
| |
|
| | def add_audio_effects(audio_path, reverb_rm_size, reverb_wet, reverb_dry, reverb_damping): |
| | """ |
| | Apply a chain of audio effects (highpass, compression, reverb) to the input audio. |
| | """ |
| | output_path = f'{os.path.splitext(audio_path)[0]}_mixed.wav' |
| | board = Pedalboard([ |
| | HighpassFilter(), |
| | Compressor(ratio=4, threshold_db=-15), |
| | Reverb(room_size=reverb_rm_size, dry_level=reverb_dry, wet_level=reverb_wet, damping=reverb_damping) |
| | ]) |
| |
|
| | with AudioFile(audio_path) as f: |
| | with AudioFile(output_path, 'w', f.samplerate, f.num_channels) as o: |
| | while f.tell() < f.frames: |
| | chunk = f.read(int(f.samplerate)) |
| | effected = board(chunk, f.samplerate, reset=False) |
| | o.write(effected) |
| | return output_path |
| |
|
| |
|
| | def combine_audio(audio_paths, output_path, main_gain, backup_gain, inst_gain, output_format): |
| | """ |
| | Combine main vocals, backup vocals, and instrumental audio into a final mix. |
| | """ |
| | main_vocal_audio = AudioSegment.from_wav(audio_paths[0]) - 4 + main_gain |
| | backup_vocal_audio = AudioSegment.from_wav(audio_paths[1]) - 6 + backup_gain |
| | instrumental_audio = AudioSegment.from_wav(audio_paths[2]) - 7 + inst_gain |
| | final_audio = main_vocal_audio.overlay(backup_vocal_audio).overlay(instrumental_audio) |
| | final_audio.export(output_path, format=output_format) |
| |
|
| |
|
| | def song_cover_pipeline(song_input, voice_model, pitch_change, keep_files, |
| | is_webui=0, main_gain=0, backup_gain=0, inst_gain=0, index_rate=0.5, filter_radius=3, |
| | rms_mix_rate=0.25, f0_method='rmvpe', crepe_hop_length=128, protect=0.33, pitch_change_all=0, |
| | reverb_rm_size=0.15, reverb_wet=0.2, reverb_dry=0.8, reverb_damping=0.7, output_format='mp3', |
| | progress=gr.Progress()): |
| | """ |
| | Main pipeline that orchestrates the AI cover song generation. |
| | """ |
| | try: |
| | if not song_input or not voice_model: |
| | raise_exception('Ensure that the song input field and voice model field is filled.', is_webui) |
| |
|
| | display_progress('[~] Starting AI Cover Generation Pipeline...', 0, is_webui, progress) |
| |
|
| | if urlparse(song_input).scheme == 'https': |
| | input_type = 'yt' |
| | song_id = get_youtube_video_id(song_input) |
| | if song_id is None: |
| | raise_exception('Invalid YouTube url.', is_webui) |
| | else: |
| | input_type = 'local' |
| | song_input = song_input.strip('\"') |
| | if os.path.exists(song_input): |
| | song_id = get_hash(song_input) |
| | else: |
| | raise_exception(f'{song_input} does not exist.', is_webui) |
| |
|
| | song_dir = os.path.join(output_dir, song_id) |
| |
|
| | if not os.path.exists(song_dir): |
| | os.makedirs(song_dir) |
| | (orig_song_path, vocals_path, instrumentals_path, |
| | main_vocals_path, backup_vocals_path, main_vocals_dereverb_path) = preprocess_song( |
| | song_input, song_id, is_webui, input_type, progress |
| | ) |
| | else: |
| | vocals_path, main_vocals_path = None, None |
| | paths = get_audio_paths(song_dir) |
| | if any(path is None for path in paths) or keep_files: |
| | (orig_song_path, vocals_path, instrumentals_path, |
| | main_vocals_path, backup_vocals_path, main_vocals_dereverb_path) = preprocess_song( |
| | song_input, song_id, is_webui, input_type, progress |
| | ) |
| | else: |
| | orig_song_path, instrumentals_path, main_vocals_dereverb_path, backup_vocals_path = paths |
| | main_vocals_path = main_vocals_dereverb_path |
| |
|
| | pitch_change += pitch_change_all |
| |
|
| | base_song_name = os.path.splitext(os.path.basename(orig_song_path))[0] |
| | algo_suffix = f"_{crepe_hop_length}" if f0_method == "mangio-crepe" else "" |
| | ai_vocals_path = os.path.join( |
| | song_dir, |
| | f'{base_song_name}_lead_{voice_model}_p{pitch_change}_i{index_rate}_fr{filter_radius}_' |
| | f'rms{rms_mix_rate}_pro{protect}_{f0_method}{algo_suffix}.wav' |
| | ) |
| | ai_backing_path = os.path.join( |
| | song_dir, |
| | f'{base_song_name}_backing_{voice_model}_p{pitch_change}_i{index_rate}_fr{filter_radius}_' |
| | f'rms{rms_mix_rate}_pro{protect}_{f0_method}{algo_suffix}.wav' |
| | ) |
| | ai_cover_path = os.path.join(song_dir, f'{base_song_name} ({voice_model} Ver).{output_format}') |
| | ai_cover_backing_path = os.path.join(song_dir, f'{base_song_name} ({voice_model} Ver With Backing).{output_format}') |
| |
|
| | if not os.path.exists(ai_vocals_path): |
| | display_progress('[~] Converting lead voice using RVC...', 0.5, is_webui, progress) |
| | voice_change(voice_model, main_vocals_dereverb_path, ai_vocals_path, pitch_change, |
| | f0_method, index_rate, filter_radius, rms_mix_rate, protect, crepe_hop_length, is_webui) |
| |
|
| | display_progress('[~] Converting backing voice using RVC...', 0.65, is_webui, progress) |
| | voice_change(voice_model, backup_vocals_path, ai_backing_path, pitch_change, |
| | f0_method, index_rate, filter_radius, rms_mix_rate, protect, crepe_hop_length, is_webui) |
| |
|
| | display_progress('[~] Applying audio effects to Vocals...', 0.8, is_webui, progress) |
| | ai_vocals_mixed_path = add_audio_effects(ai_vocals_path, reverb_rm_size, reverb_wet, reverb_dry, reverb_damping) |
| | ai_backing_mixed_path = add_audio_effects(ai_backing_path, reverb_rm_size, reverb_wet, reverb_dry, reverb_damping) |
| |
|
| | if pitch_change_all != 0: |
| | display_progress('[~] Applying overall pitch change', 0.85, is_webui, progress) |
| | instrumentals_path = pitch_shift(instrumentals_path, pitch_change_all) |
| | backup_vocals_path = pitch_shift(backup_vocals_path, pitch_change_all) |
| |
|
| | display_progress('[~] Combining AI Vocals and Instrumentals...', 0.9, is_webui, progress) |
| | combine_audio([ai_vocals_mixed_path, backup_vocals_path, instrumentals_path], |
| | ai_cover_path, main_gain, backup_gain, inst_gain, output_format) |
| | combine_audio([ai_vocals_mixed_path, ai_backing_mixed_path, instrumentals_path], |
| | ai_cover_backing_path, main_gain, backup_gain, inst_gain, output_format) |
| |
|
| | if not keep_files: |
| | display_progress('[~] Removing intermediate audio files...', 0.95, is_webui, progress) |
| | intermediate_files = [vocals_path, main_vocals_path, ai_vocals_mixed_path, ai_backing_mixed_path] |
| | if pitch_change_all != 0: |
| | intermediate_files += [instrumentals_path, backup_vocals_path] |
| | for file in intermediate_files: |
| | if file and os.path.exists(file): |
| | os.remove(file) |
| |
|
| | return ai_cover_path, ai_cover_backing_path |
| |
|
| | except Exception as e: |
| | raise_exception(str(e), is_webui) |
| |
|
| |
|
| | if __name__ == '__main__': |
| | parser = argparse.ArgumentParser( |
| | description='AICoverGen: Mod.', |
| | add_help=True |
| | ) |
| | parser.add_argument('-i', '--song-input', type=str, required=True, |
| | help='Link to a YouTube video or the filepath to a local mp3/wav file to create an AI cover of') |
| | parser.add_argument('-dir', '--rvc-dirname', type=str, required=True, |
| | help='Name of the folder in the rvc_models directory containing the RVC model file and optional index file to use') |
| | parser.add_argument('-p', '--pitch-change', type=int, required=True, |
| | help='Change the pitch of AI Vocals only. Generally, use 1 for male to female and -1 for vice-versa. (Octaves)') |
| | parser.add_argument('-k', '--keep-files', action=argparse.BooleanOptionalAction, |
| | help='Whether to keep all intermediate audio files generated in the song_output/id directory, e.g. Isolated Vocals/Instrumentals') |
| | parser.add_argument('-ir', '--index-rate', type=float, default=0.5, |
| | help='A decimal number e.g. 0.5, used to reduce/resolve the timbre leakage problem. If set to 1, more biased towards the timbre quality of the training dataset') |
| | parser.add_argument('-fr', '--filter-radius', type=int, default=3, |
| | help='A number between 0 and 7. If >=3: apply median filtering to the harvested pitch results. The value represents the filter radius and can reduce breathiness.') |
| | parser.add_argument('-rms', '--rms-mix-rate', type=float, default=0.25, |
| | help="A decimal number e.g. 0.25. Control how much to use the original vocal's loudness (0) or a fixed loudness (1).") |
| | parser.add_argument('-palgo', '--pitch-detection-algo', type=str, default='rmvpe', |
| | help='Best option is rmvpe (clarity in vocals), then mangio-crepe (smoother vocals).') |
| | parser.add_argument('-hop', '--crepe-hop-length', type=int, default=128, |
| | help='If pitch detection algo is mangio-crepe, controls how often it checks for pitch changes in milliseconds. Recommended: 128.') |
| | parser.add_argument('-pro', '--protect', type=float, default=0.33, |
| | help='A decimal number e.g. 0.33. Protect voiceless consonants and breath sounds to prevent artifacts such as tearing in electronic music.') |
| | parser.add_argument('-mv', '--main-vol', type=int, default=0, |
| | help='Volume change for AI main vocals in decibels. Use -3 to decrease by 3 dB and 3 to increase by 3 dB') |
| | parser.add_argument('-bv', '--backup-vol', type=int, default=0, |
| | help='Volume change for backup vocals in decibels') |
| | parser.add_argument('-iv', '--inst-vol', type=int, default=0, |
| | help='Volume change for instrumentals in decibels') |
| | parser.add_argument('-pall', '--pitch-change-all', type=int, default=0, |
| | help='Change the pitch/key of vocals and instrumentals. Changing this slightly reduces sound quality') |
| | parser.add_argument('-rsize', '--reverb-size', type=float, default=0.15, |
| | help='Reverb room size between 0 and 1') |
| | parser.add_argument('-rwet', '--reverb-wetness', type=float, default=0.2, |
| | help='Reverb wet level between 0 and 1') |
| | parser.add_argument('-rdry', '--reverb-dryness', type=float, default=0.8, |
| | help='Reverb dry level between 0 and 1') |
| | parser.add_argument('-rdamp', '--reverb-damping', type=float, default=0.7, |
| | help='Reverb damping between 0 and 1') |
| | parser.add_argument('-oformat', '--output-format', type=str, default='mp3', |
| | help='Output format of audio file. mp3 for smaller file size, wav for best quality') |
| | args = parser.parse_args() |
| |
|
| | rvc_dir = os.path.join(rvc_models_dir, args.rvc_dirname) |
| | if not os.path.exists(rvc_dir): |
| | raise Exception(f'The folder {rvc_dir} does not exist.') |
| |
|
| | cover_path, cover_with_backing = song_cover_pipeline( |
| | args.song_input, args.rvc_dirname, args.pitch_change, args.keep_files, |
| | main_gain=args.main_vol, backup_gain=args.backup_vol, inst_gain=args.inst_vol, |
| | index_rate=args.index_rate, filter_radius=args.filter_radius, |
| | rms_mix_rate=args.rms_mix_rate, f0_method=args.pitch_detection_algo, |
| | crepe_hop_length=args.crepe_hop_length, protect=args.protect, |
| | pitch_change_all=args.pitch_change_all, |
| | reverb_rm_size=args.reverb_size, reverb_wet=args.reverb_wetness, |
| | reverb_dry=args.reverb_dryness, reverb_damping=args.reverb_damping, |
| | output_format=args.output_format |
| | ) |
| | print(f'[+] Cover generated at {cover_path}') |
