import spaces import random import argparse import glob import json import os import time import gradio as gr import numpy as np import torch import torch.nn.functional as F import tqdm from huggingface_hub import hf_hub_download import MIDI from midi_model import MIDIModel, MIDIModelConfig from midi_synthesizer import MidiSynthesizer MAX_SEED = np.iinfo(np.int32).max OUTPUT_BATCH_SIZE = 4 in_space = os.getenv("SYSTEM") == "spaces" @torch.inference_mode() def generate(model: MIDIModel, prompt=None, batch_size=1, max_len=512, temp=1.0, top_p=0.98, top_k=20, disable_patch_change=False, disable_control_change=False, disable_channels=None, generator=None): tokenizer = model.tokenizer if disable_channels is not None: disable_channels = [tokenizer.parameter_ids["channel"][c] for c in disable_channels] else: disable_channels = [] max_token_seq = tokenizer.max_token_seq if prompt is None: input_tensor = torch.full((1, max_token_seq), tokenizer.pad_id, dtype=torch.long, device=model.device) input_tensor[0, 0] = tokenizer.bos_id # bos input_tensor = input_tensor.unsqueeze(0) input_tensor = torch.cat([input_tensor] * batch_size, dim=0) else: if len(prompt.shape) == 2: prompt = prompt[None, :] prompt = np.repeat(prompt, repeats=batch_size, axis=0) elif prompt.shape[0] == 1: prompt = np.repeat(prompt, repeats=batch_size, axis=0) elif len(prompt.shape) != 3 or prompt.shape[0] != batch_size: raise ValueError(f"invalid shape for prompt, {prompt.shape}") prompt = prompt[..., :max_token_seq] if prompt.shape[-1] < max_token_seq: prompt = np.pad(prompt, ((0, 0), (0, 0), (0, max_token_seq - prompt.shape[-1])), mode="constant", constant_values=tokenizer.pad_id) input_tensor = torch.from_numpy(prompt).to(dtype=torch.long, device=model.device) cur_len = input_tensor.shape[1] bar = tqdm.tqdm(desc="generating", total=max_len - cur_len) with bar: while cur_len < max_len: end = [False] * batch_size hidden = model.forward(input_tensor)[:, -1] next_token_seq = None event_names = [""] * batch_size for i in range(max_token_seq): mask = torch.zeros((batch_size, tokenizer.vocab_size), dtype=torch.int64, device=model.device) for b in range(batch_size): if end[b]: mask[b, tokenizer.pad_id] = 1 continue if i == 0: mask_ids = list(tokenizer.event_ids.values()) + [tokenizer.eos_id] if disable_patch_change: mask_ids.remove(tokenizer.event_ids["patch_change"]) if disable_control_change: mask_ids.remove(tokenizer.event_ids["control_change"]) mask[b, mask_ids] = 1 else: param_names = tokenizer.events[event_names[b]] if i > len(param_names): mask[b, tokenizer.pad_id] = 1 continue param_name = param_names[i - 1] mask_ids = tokenizer.parameter_ids[param_name] if param_name == "channel": mask_ids = [i for i in mask_ids if i not in disable_channels] mask[b, mask_ids] = 1 mask = mask.unsqueeze(1) logits = model.forward_token(hidden, next_token_seq)[:, -1:] scores = torch.softmax(logits / temp, dim=-1) * mask samples = model.sample_top_p_k(scores, top_p, top_k, generator=generator) if i == 0: next_token_seq = samples for b in range(batch_size): if end[b]: continue eid = samples[b].item() if eid == tokenizer.eos_id: end[b] = True else: event_names[b] = tokenizer.id_events[eid] else: next_token_seq = torch.cat([next_token_seq, samples], dim=1) if all([len(tokenizer.events[event_names[b]]) == i for b in range(batch_size) if not end[b]]): break if next_token_seq.shape[1] < max_token_seq: next_token_seq = F.pad(next_token_seq, (0, max_token_seq - next_token_seq.shape[1]), "constant", value=tokenizer.pad_id) next_token_seq = next_token_seq.unsqueeze(1) input_tensor = torch.cat([input_tensor, next_token_seq], dim=1) cur_len += 1 bar.update(1) yield next_token_seq[:, 0].cpu().numpy() if all(end): break def create_msg(name, data): return {"name": name, "data": data} def send_msgs(msgs): return json.dumps(msgs) def get_duration(model_name, tab, mid_seq, continuation_state, continuation_select, instruments, drum_kit, bpm, time_sig, key_sig, mid, midi_events, reduce_cc_st, remap_track_channel, add_default_instr, remove_empty_channels, seed, seed_rand, gen_events, temp, top_p, top_k, allow_cc): if "large" in model_name: return gen_events // 10 + 15 else: return gen_events // 20 + 15 @spaces.GPU(duration=get_duration) def run(model_name, tab, mid_seq, continuation_state, continuation_select, instruments, drum_kit, bpm, time_sig, key_sig, mid, midi_events, reduce_cc_st, remap_track_channel, add_default_instr, remove_empty_channels, seed, seed_rand, gen_events, temp, top_p, top_k, allow_cc): model = models[model_name] model.to(device=opt.device) tokenizer = model.tokenizer bpm = int(bpm) if time_sig == "auto": time_sig = None time_sig_nn = 4 time_sig_dd = 2 else: time_sig_nn, time_sig_dd = time_sig.split('/') time_sig_nn = int(time_sig_nn) time_sig_dd = {2: 1, 4: 2, 8: 3}[int(time_sig_dd)] if key_sig == 0: key_sig = None key_sig_sf = 0 key_sig_mi = 0 else: key_sig = (key_sig - 1) key_sig_sf = key_sig // 2 - 7 key_sig_mi = key_sig % 2 gen_events = int(gen_events) max_len = gen_events if seed_rand: seed = random.randint(0, MAX_SEED) generator = torch.Generator(opt.device).manual_seed(seed) disable_patch_change = False disable_channels = None if tab == 0: i = 0 mid = [[tokenizer.bos_id] + [tokenizer.pad_id] * (tokenizer.max_token_seq - 1)] if tokenizer.version == "v2": if time_sig is not None: mid.append(tokenizer.event2tokens(["time_signature", 0, 0, 0, time_sig_nn - 1, time_sig_dd - 1])) if key_sig is not None: mid.append(tokenizer.event2tokens(["key_signature", 0, 0, 0, key_sig_sf + 7, key_sig_mi])) if bpm != 0: mid.append(tokenizer.event2tokens(["set_tempo", 0, 0, 0, bpm])) patches = {} if instruments is None: instruments = [] for instr in instruments: patches[i] = patch2number[instr] i = (i + 1) if i != 8 else 10 if drum_kit != "None": patches[9] = drum_kits2number[drum_kit] for i, (c, p) in enumerate(patches.items()): mid.append(tokenizer.event2tokens(["patch_change", 0, 0, i + 1, c, p])) mid = np.asarray([mid] * OUTPUT_BATCH_SIZE, dtype=np.int64) mid_seq = mid.tolist() if len(instruments) > 0: disable_patch_change = True disable_channels = [i for i in range(16) if i not in patches] elif tab == 1 and mid is not None: eps = 4 if reduce_cc_st else 0 mid = tokenizer.tokenize(MIDI.midi2score(mid), cc_eps=eps, tempo_eps=eps, remap_track_channel=remap_track_channel, add_default_instr=add_default_instr, remove_empty_channels=remove_empty_channels) mid = mid[:int(midi_events)] mid = np.asarray([mid] * OUTPUT_BATCH_SIZE, dtype=np.int64) mid_seq = mid.tolist() elif tab == 2 and mid_seq is not None: mid = np.asarray(mid_seq, dtype=np.int64) if continuation_select > 0: continuation_state.append(mid_seq) mid = np.repeat(mid[continuation_select - 1:continuation_select], repeats=OUTPUT_BATCH_SIZE, axis=0) mid_seq = mid.tolist() else: continuation_state.append(mid.shape[1]) else: continuation_state = [0] mid = [[tokenizer.bos_id] + [tokenizer.pad_id] * (tokenizer.max_token_seq - 1)] mid = np.asarray([mid] * OUTPUT_BATCH_SIZE, dtype=np.int64) mid_seq = mid.tolist() if mid is not None: max_len += mid.shape[1] init_msgs = [create_msg("progress", [0, gen_events])] if not (tab == 2 and continuation_select == 0): for i in range(OUTPUT_BATCH_SIZE): events = [tokenizer.tokens2event(tokens) for tokens in mid_seq[i]] init_msgs += [create_msg("visualizer_clear", [i, tokenizer.version]), create_msg("visualizer_append", [i, events])] yield mid_seq, continuation_state, seed, send_msgs(init_msgs) midi_generator = generate(model, mid, batch_size=OUTPUT_BATCH_SIZE, max_len=max_len, temp=temp, top_p=top_p, top_k=top_k, disable_patch_change=disable_patch_change, disable_control_change=not allow_cc, disable_channels=disable_channels, generator=generator) events = [list() for i in range(OUTPUT_BATCH_SIZE)] t = time.time() for i, token_seqs in enumerate(midi_generator): token_seqs = token_seqs.tolist() for j in range(OUTPUT_BATCH_SIZE): token_seq = token_seqs[j] mid_seq[j].append(token_seq) events[j].append(tokenizer.tokens2event(token_seq)) if time.time() - t > 0.2: msgs = [create_msg("progress", [i + 1, gen_events])] for j in range(OUTPUT_BATCH_SIZE): msgs += [create_msg("visualizer_append", [j, events[j]])] events[j] = list() yield mid_seq, continuation_state, seed, send_msgs(msgs) t = time.time() yield mid_seq, continuation_state, seed, send_msgs([]) def finish_run(model_name, mid_seq): if mid_seq is None: return None, None, [] tokenizer = models[model_name].tokenizer outputs = [] end_msgs = [create_msg("progress", [0, 0])] if not os.path.exists("outputs"): os.mkdir("outputs") for i in range(OUTPUT_BATCH_SIZE): events = [tokenizer.tokens2event(tokens) for tokens in mid_seq[i]] mid = tokenizer.detokenize(mid_seq[i]) audio = synthesizer.synthesis(MIDI.score2opus(mid)) with open(f"outputs/output{i + 1}.mid", 'wb') as f: f.write(MIDI.score2midi(mid)) outputs += [(44100, audio), f"outputs/output{i + 1}.mid"] end_msgs += [create_msg("visualizer_clear", [i, tokenizer.version]), create_msg("visualizer_append", [i, events]), create_msg("visualizer_end", i)] return *outputs, send_msgs(end_msgs) def undo_continuation(model_name, mid_seq, continuation_state): if mid_seq is None or len(continuation_state) < 2: return mid_seq, continuation_state, send_msgs([]) tokenizer = models[model_name].tokenizer if isinstance(continuation_state[-1], list): mid_seq = continuation_state[-1] else: mid_seq = [ms[:continuation_state[-1]] for ms in mid_seq] continuation_state = continuation_state[:-1] end_msgs = [create_msg("progress", [0, 0])] for i in range(OUTPUT_BATCH_SIZE): events = [tokenizer.tokens2event(tokens) for tokens in mid_seq[i]] end_msgs += [create_msg("visualizer_clear", [i, tokenizer.version]), create_msg("visualizer_append", [i, events]), create_msg("visualizer_end", i)] return mid_seq, continuation_state, send_msgs(end_msgs) def load_javascript(dir="javascript"): scripts_list = glob.glob(f"{dir}/*.js") javascript = "" for path in scripts_list: with open(path, "r", encoding="utf8") as jsfile: javascript += f"\n" template_response_ori = gr.routes.templates.TemplateResponse def template_response(*args, **kwargs): res = template_response_ori(*args, **kwargs) res.body = res.body.replace( b'', f'{javascript}'.encode("utf8")) res.init_headers() return res gr.routes.templates.TemplateResponse = template_response def hf_hub_download_retry(repo_id, filename): print(f"downloading {repo_id} {filename}") retry = 0 err = None while retry < 30: try: return hf_hub_download(repo_id=repo_id, filename=filename) except Exception as e: err = e retry += 1 if err: raise err number2drum_kits = {-1: "None", 0: "Standard", 8: "Room", 16: "Power", 24: "Electric", 25: "TR-808", 32: "Jazz", 40: "Blush", 48: "Orchestra"} patch2number = {v: k for k, v in MIDI.Number2patch.items()} drum_kits2number = {v: k for k, v in number2drum_kits.items()} key_signatures = ['C♭', 'A♭m', 'G♭', 'E♭m', 'D♭', 'B♭m', 'A♭', 'Fm', 'E♭', 'Cm', 'B♭', 'Gm', 'F', 'Dm', 'C', 'Am', 'G', 'Em', 'D', 'Bm', 'A', 'F♯m', 'E', 'C♯m', 'B', 'G♯m', 'F♯', 'D♯m', 'C♯', 'A♯m'] if __name__ == "__main__": parser = argparse.ArgumentParser() parser.add_argument("--share", action="store_true", default=False, help="share gradio app") parser.add_argument("--port", type=int, default=7860, help="gradio server port") parser.add_argument("--device", type=str, default="cuda", help="device to run model") parser.add_argument("--max-gen", type=int, default=1024, help="max") opt = parser.parse_args() soundfont_path = hf_hub_download_retry(repo_id="skytnt/midi-model", filename="soundfont.sf2") synthesizer = MidiSynthesizer(soundfont_path) models_info = { "generic pretrain model (tv2o-medium) by skytnt": ["skytnt/midi-model-tv2o-medium", "", "tv2o-medium"], "generic pretrain model (tv2o-large) by asigalov61": ["asigalov61/Music-Llama", "", "tv2o-large"], "generic pretrain model (tv2o-medium) by asigalov61": ["asigalov61/Music-Llama-Medium", "", "tv2o-medium"], "generic pretrain model (tv1-medium) by skytnt": ["skytnt/midi-model", "", "tv1-medium"], "j-pop finetune model (tv2o-medium) by skytnt": ["skytnt/midi-model-ft", "jpop-tv2o-medium/", "tv2o-medium"], "touhou finetune model (tv2o-medium) by skytnt": ["skytnt/midi-model-ft", "touhou-tv2o-medium/", "tv2o-medium"], } models = {} if opt.device == "cuda": torch.backends.cudnn.deterministic = True torch.backends.cudnn.benchmark = False torch.backends.cuda.matmul.allow_tf32 = True torch.backends.cudnn.allow_tf32 = True torch.backends.cuda.enable_mem_efficient_sdp(True) torch.backends.cuda.enable_flash_sdp(True) for name, (repo_id, path, config) in models_info.items(): model_path = hf_hub_download_retry(repo_id=repo_id, filename=f"{path}model.ckpt") model = MIDIModel(config=MIDIModelConfig.from_name(config)) ckpt = torch.load(model_path, map_location="cpu", weights_only=True) state_dict = ckpt.get("state_dict", ckpt) model.load_state_dict(state_dict, strict=False) model.to(device="cpu", dtype=torch.float32) models[name] = model load_javascript() app = gr.Blocks() with app: gr.Markdown("