Spaces:
Runtime error
Runtime error
| import argparse | |
| from concurrent.futures import ProcessPoolExecutor | |
| import os | |
| from pathlib import Path | |
| import subprocess as sp | |
| from tempfile import NamedTemporaryFile | |
| import time | |
| import typing as tp | |
| import warnings | |
| from concurrent.futures import Future | |
| import torch | |
| import gradio as gr | |
| import pydub | |
| from scipy.io.wavfile import write | |
| from audiocraft.data.audio_utils import convert_audio | |
| from audiocraft.data.audio import audio_write | |
| from audiocraft.models import MusicGen | |
| MODEL = None # Last used model | |
| IS_BATCHED = "facebook/MusicGen" in os.environ.get('SPACE_ID', '') | |
| MAX_BATCH_SIZE = 6 | |
| BATCHED_DURATION = 15 | |
| INTERRUPTING = False | |
| # We have to wrap subprocess call to clean a bit the log when using gr.make_waveform | |
| _old_call = sp.call | |
| files = ["./out/mdx_extra_q/test/full.wav", | |
| "./out/mdx_extra_q/test/vocals.wav", | |
| "./out/mdx_extra_q/test/bass.wav", | |
| "./out/mdx_extra_q/test/drums.wav", | |
| "./out/mdx_extra_q/test/other.wav" | |
| ] | |
| def _call_nostderr(*args, **kwargs): | |
| # Avoid ffmpeg vomitting on the logs. | |
| kwargs['stderr'] = sp.DEVNULL | |
| kwargs['stdout'] = sp.DEVNULL | |
| _old_call(*args, **kwargs) | |
| sp.call = _call_nostderr | |
| # Preallocating the pool of processes. | |
| pool = ProcessPoolExecutor(3) | |
| pool.__enter__() | |
| def interrupt(): | |
| global INTERRUPTING | |
| INTERRUPTING = True | |
| class FileCleaner: | |
| def __init__(self, file_lifetime: float = 3600): | |
| self.file_lifetime = file_lifetime | |
| self.files = [] | |
| def add(self, path: tp.Union[str, Path]): | |
| self._cleanup() | |
| self.files.append((time.time(), Path(path))) | |
| def _cleanup(self): | |
| now = time.time() | |
| for time_added, path in list(self.files): | |
| if now - time_added > self.file_lifetime: | |
| if path.exists(): | |
| path.unlink() | |
| self.files.pop(0) | |
| else: | |
| break | |
| file_cleaner = FileCleaner() | |
| def make_waveform(*args, **kwargs): | |
| # Further remove some warnings. | |
| be = time.time() | |
| with warnings.catch_warnings(): | |
| warnings.simplefilter('ignore') | |
| out = gr.make_waveform(*args, **kwargs) | |
| print("Make a video took", time.time() - be) | |
| return out | |
| def load_model(version='melody'): | |
| global MODEL | |
| print("Loading model", version) | |
| if MODEL is None or MODEL.name != version: | |
| MODEL = MusicGen.get_pretrained(version) | |
| def _do_predictions(texts, melodies, duration, progress=False, **gen_kwargs): | |
| MODEL.set_generation_params(duration=duration, **gen_kwargs) | |
| print("new batch", len(texts), texts, [None if m is None else (m[0], m[1].shape) for m in melodies]) | |
| be = time.time() | |
| processed_melodies = [] | |
| target_sr = 32000 | |
| target_ac = 1 | |
| for melody in melodies: | |
| if melody is None: | |
| processed_melodies.append(None) | |
| else: | |
| sr, melody = melody[0], torch.from_numpy(melody[1]).to(MODEL.device).float().t() | |
| if melody.dim() == 1: | |
| melody = melody[None] | |
| melody = melody[..., :int(sr * duration)] | |
| melody = convert_audio(melody, sr, target_sr, target_ac) | |
| processed_melodies.append(melody) | |
| if any(m is not None for m in processed_melodies): | |
| outputs = MODEL.generate_with_chroma( | |
| descriptions=texts, | |
| melody_wavs=processed_melodies, | |
| melody_sample_rate=target_sr, | |
| progress=progress, | |
| ) | |
| else: | |
| outputs = MODEL.generate(texts, progress=progress) | |
| outputs = outputs.detach().cpu().float() | |
| out_files = [] | |
| for output in outputs: | |
| with NamedTemporaryFile("wb", suffix=".wav", delete=False) as file: | |
| audio_write( | |
| file.name, output, MODEL.sample_rate, strategy="loudness", | |
| loudness_headroom_db=16, loudness_compressor=True, add_suffix=False) | |
| out_files.append(file.name) # Store the filename as a string | |
| file_cleaner.add(file.name) | |
| res = [out_file for out_file in out_files] | |
| for file in res: | |
| if isinstance(file, Future): # Check if it's a Future object | |
| file = file.result() # Extract the filename from the Future object | |
| file_cleaner.add(file) | |
| print("batch finished", len(texts), time.time() - be) | |
| print("Tempfiles currently stored: ", len(file_cleaner.files)) | |
| return res | |
| def predict_batched(texts, melodies): | |
| max_text_length = 512 | |
| texts = [text[:max_text_length] for text in texts] | |
| load_model('melody') | |
| res = _do_predictions(texts, melodies, BATCHED_DURATION) | |
| return [res] | |
| def predict_full(model, text, melody, duration, topk, topp, temperature, cfg_coef, progress=gr.Progress()): | |
| global INTERRUPTING | |
| INTERRUPTING = False | |
| if temperature < 0: | |
| raise gr.Error("Temperature must be >= 0.") | |
| if topk < 0: | |
| raise gr.Error("Topk must be non-negative.") | |
| if topp < 0: | |
| raise gr.Error("Topp must be non-negative.") | |
| topk = int(topk) | |
| load_model(model) | |
| def _progress(generated, to_generate): | |
| progress((generated, to_generate)) | |
| if INTERRUPTING: | |
| raise gr.Error("Interrupted.") | |
| MODEL.set_custom_progress_callback(_progress) | |
| outs = _do_predictions( | |
| [text], [melody], duration, progress=True, | |
| top_k=topk, top_p=topp, temperature=temperature, cfg_coef=cfg_coef) | |
| return inference(outs[0]) | |
| def inference(audio): | |
| print (audio) | |
| os.makedirs("out", exist_ok=True) | |
| command = "python3 -m demucs.separate -n mdx_extra_q -d cpu test.wav -o out" | |
| # Find the index of "test.wav" in the command | |
| index = command.find("test.wav") | |
| # Replace "test.wav" with the file path | |
| new_command = command[:index] + audio + command[index + len("test.wav"):] | |
| process = sp.run(new_command, shell=True, stdout=sp.PIPE, stderr=sp.PIPE) | |
| print("Demucs script output:", process.stdout.decode()) | |
| os.makedirs("out", exist_ok=True) | |
| result = os.system(new_command) | |
| print(f"Demucs script result: {result}") | |
| directory_name = os.path.splitext(os.path.basename(audio))[0] | |
| files = [audio, | |
| f"./out/mdx_extra_q/{directory_name}/vocals.wav", | |
| f"./out/mdx_extra_q/{directory_name}/bass.wav", | |
| f"./out/mdx_extra_q/{directory_name}/drums.wav", | |
| f"./out/mdx_extra_q/{directory_name}/other.wav" | |
| ] | |
| for file in files: | |
| if not os.path.isfile(file): | |
| print(f"File not found: {file}") | |
| else: | |
| print(f"File exists: {file}") | |
| return files | |
| def toggle_audio_src(choice): | |
| if choice == "mic": | |
| return gr.update(source="microphone", value=None, label="Microphone") | |
| else: | |
| return gr.update(source="upload", value=None, label="File") | |
| def ui_full(launch_kwargs): | |
| interface = gr.Interface( | |
| fn=predict_full, | |
| inputs=[ | |
| gr.Radio(["melody", "medium", "small", "large"], label="Model", default="melody"), | |
| gr.Text(label="Input Text"), | |
| gr.Audio(source="upload", type="numpy", label="File", interactive=True, elem_id="melody-input"), | |
| gr.Slider(minimum=1, maximum=120, default=10, label="Duration", step=1), | |
| gr.Number(label="Top-k", default=250), | |
| gr.Number(label="Top-p", default=0), | |
| gr.Number(label="Temperature", default=1.0), | |
| gr.Number(label="Classifier Free Guidance", default=3.0), | |
| ], | |
| outputs=[ | |
| gr.outputs.Audio(type='filepath'), | |
| gr.outputs.Audio(type='filepath'), | |
| gr.outputs.Audio(type='filepath'), | |
| gr.outputs.Audio(type='filepath'), | |
| gr.outputs.Audio(type='filepath') | |
| ], | |
| title="MusicGen", | |
| description="This is your private demo for MusicGen, a simple and controllable model for music generation.", | |
| allow_flagging="never", | |
| layout="vertical", | |
| **launch_kwargs | |
| ) | |
| interface.launch() | |
| if __name__ == "__main__": | |
| parser = argparse.ArgumentParser() | |
| parser.add_argument( | |
| '--listen', | |
| type=str, | |
| default='0.0.0.0' if 'SPACE_ID' in os.environ else '127.0.0.1', | |
| help='IP to listen on for connections to Gradio', | |
| ) | |
| parser.add_argument( | |
| '--username', type=str, default='', help='Username for authentication' | |
| ) | |
| parser.add_argument( | |
| '--password', type=str, default='', help='Password for authentication' | |
| ) | |
| parser.add_argument( | |
| '--server_port', | |
| type=int, | |
| default=0, | |
| help='Port to run the server listener on', | |
| ) | |
| parser.add_argument( | |
| '--inbrowser', action='store_true', help='Open in browser' | |
| ) | |
| parser.add_argument( | |
| '--share', action='store_true', help='Share the gradio UI' | |
| ) | |
| args = parser.parse_args() | |
| launch_kwargs = {} | |
| launch_kwargs['server_name'] = args.listen | |
| if args.username and args.password: | |
| launch_kwargs['auth'] = (args.username, args.password) | |
| if args.server_port: | |
| launch_kwargs['server_port'] = args.server_port | |
| if args.inbrowser: | |
| launch_kwargs['inbrowser'] = args.inbrowser | |
| if args.share: | |
| launch_kwargs['share'] = args.share | |
| # Show the interface | |
| ui_full(launch_kwargs) | |
| def ui_batched(launch_kwargs): | |
| with gr.Blocks() as demo: | |
| gr.Markdown( | |
| """ | |
| # MusicGen | |
| This is the demo for [MusicGen](https://github.com/facebookresearch/audiocraft), | |
| a simple and controllable model for music generation | |
| presented at: ["Simple and Controllable Music Generation"](https://huggingface.co/papers/2306.05284). | |
| <br/> | |
| <a href="https://huggingface.co/spaces/facebook/MusicGen?duplicate=true" | |
| style="display: inline-block;margin-top: .5em;margin-right: .25em;" target="_blank"> | |
| <img style="margin-bottom: 0em;display: inline;margin-top: -.25em;" | |
| src="https://bit.ly/3gLdBN6" alt="Duplicate Space"></a> | |
| for longer sequences, more control and no queue.</p> | |
| """ | |
| ) | |
| with gr.Row(): | |
| with gr.Column(): | |
| with gr.Row(): | |
| text = gr.Text(label="Describe your music", lines=2, interactive=True) | |
| with gr.Column(): | |
| radio = gr.Radio(["file", "mic"], value="file", | |
| label="Condition on a melody (optional) File or Mic") | |
| melody = gr.Audio(source="upload", type="numpy", label="File", | |
| interactive=True, elem_id="melody-input") | |
| with gr.Row(): | |
| submit = gr.Button("Generate") | |
| with gr.Column(): | |
| output = [gr.Audio(file, label=f"Generated Music {i+1}") for i, file in enumerate(files)] | |
| submit.click(predict_batched, inputs=[text, melody], | |
| outputs=[output], batch=True, max_batch_size=MAX_BATCH_SIZE) | |
| radio.change(toggle_audio_src, radio, [melody], queue=False, show_progress=False) | |
| gr.Examples( | |
| fn=predict_batched, | |
| examples=[ | |
| [ | |
| "An 80s driving pop song with heavy drums and synth pads in the background", | |
| "./assets/bach.mp3", | |
| ], | |
| [ | |
| "A cheerful country song with acoustic guitars", | |
| "./assets/bolero_ravel.mp3", | |
| ], | |
| [ | |
| "90s rock song with electric guitar and heavy drums", | |
| None, | |
| ], | |
| [ | |
| "a light and cheerly EDM track, with syncopated drums, aery pads, and strong emotions bpm: 130", | |
| "./assets/bach.mp3", | |
| ], | |
| [ | |
| "lofi slow bpm electro chill with organic samples", | |
| None, | |
| ], | |
| ], | |
| inputs=[text, melody], | |
| outputs=[output] | |
| ) | |
| gr.Markdown(""" | |
| ### More details | |
| The model will generate 12 seconds of audio based on the description you provided. | |
| You can optionaly provide a reference audio from which a broad melody will be extracted. | |
| The model will then try to follow both the description and melody provided. | |
| All samples are generated with the `melody` model. | |
| You can also use your own GPU or a Google Colab by following the instructions on our repo. | |
| See [github.com/facebookresearch/audiocraft](https://github.com/facebookresearch/audiocraft) | |
| for more details. | |
| """) | |
| demo.queue(max_size=8 * 4).launch(**launch_kwargs) | |
| if __name__ == "__main__": | |
| parser = argparse.ArgumentParser() | |
| parser.add_argument( | |
| '--listen', | |
| type=str, | |
| default='0.0.0.0' if 'SPACE_ID' in os.environ else '127.0.0.1', | |
| help='IP to listen on for connections to Gradio', | |
| ) | |
| parser.add_argument( | |
| '--username', type=str, default='', help='Username for authentication' | |
| ) | |
| parser.add_argument( | |
| '--password', type=str, default='', help='Password for authentication' | |
| ) | |
| parser.add_argument( | |
| '--server_port', | |
| type=int, | |
| default=0, | |
| help='Port to run the server listener on', | |
| ) | |
| parser.add_argument( | |
| '--inbrowser', action='store_true', help='Open in browser' | |
| ) | |
| parser.add_argument( | |
| '--share', action='store_true', help='Share the gradio UI' | |
| ) | |
| args = parser.parse_args() | |
| launch_kwargs = {} | |
| launch_kwargs['server_name'] = args.listen | |
| if args.username and args.password: | |
| launch_kwargs['auth'] = (args.username, args.password) | |
| if args.server_port: | |
| launch_kwargs['server_port'] = args.server_port | |
| if args.inbrowser: | |
| launch_kwargs['inbrowser'] = args.inbrowser | |
| if args.share: | |
| launch_kwargs['share'] = args.share | |
| # Show the interface | |
| if IS_BATCHED: | |
| ui_batched(launch_kwargs) | |
| else: | |
| ui_full(launch_kwargs) |