Update app.py

app.py

@@ -12,35 +12,61 @@ from huggingface_hub import hf_hub_download
 import MIDI
 from midi_synthesizer import synthesis
 from midi_tokenizer import MIDITokenizer
+
 in_space = os.getenv("SYSTEM") == "spaces"
-@torch.inference_mode()
-def generate(prompt=None, max_len=512, temp=1.0, top_p=0.98, top_k=20,
-             disable_patch_change=False, disable_control_change=False, disable_channels=None, amp=True):
+
+
+def softmax(x, axis):
+    x_max = np.amax(x, axis=axis, keepdims=True)
+    exp_x_shifted = np.exp(x - x_max)
+    return exp_x_shifted / np.sum(exp_x_shifted, axis=axis, keepdims=True)
+
+
+def sample_top_p_k(probs, p, k):
+    probs_idx = np.argsort(-probs, axis=-1)
+    probs_sort = np.take_along_axis(probs, probs_idx, -1)
+    probs_sum = np.cumsum(probs_sort, axis=-1)
+    mask = probs_sum - probs_sort > p
+    probs_sort[mask] = 0.0
+    mask = np.zeros(probs_sort.shape[-1])
+    mask[:k] = 1
+    probs_sort = probs_sort * mask
+    probs_sort /= np.sum(probs_sort, axis=-1, keepdims=True)
+    shape = probs_sort.shape
+    probs_sort_flat = probs_sort.reshape(-1, shape[-1])
+    probs_idx_flat = probs_idx.reshape(-1, shape[-1])
+    next_token = np.stack([np.random.choice(idxs, p=pvals) for pvals, idxs in zip(probs_sort_flat, probs_idx_flat)])
+    next_token = next_token.reshape(*shape[:-1])
+    return next_token
+
+
+def generate(model, prompt=None, max_len=512, temp=1.0, top_p=0.98, top_k=20,
+             disable_patch_change=False, disable_control_change=False, disable_channels=None):
     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 = np.full((1, max_token_seq), tokenizer.pad_id, dtype=np.int64)
         input_tensor[0, 0] = tokenizer.bos_id  # bos
     else:
         prompt = prompt[:, :max_token_seq]
         if prompt.shape[-1] < max_token_seq:
             prompt = np.pad(prompt, ((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)
-    input_tensor = input_tensor.unsqueeze(0)
+        input_tensor = prompt
+    input_tensor = input_tensor[None, :, :]
     cur_len = input_tensor.shape[1]
-    bar = tqdm.tqdm(desc="generating", total=max_len - cur_len)
-    with bar, torch.cuda.amp.autocast(enabled=amp):
+    bar = tqdm.tqdm(desc="generating", total=max_len - cur_len, disable=in_space)
+    with bar:
         while cur_len < max_len:
             end = False
-            hidden = model.forward(input_tensor)[0, -1].unsqueeze(0)
-            next_token_seq = None
+            hidden = model[0].run(None, {'x': input_tensor})[0][:, -1]
+            next_token_seq = np.empty((1, 0), dtype=np.int64)
             event_name = ""
             for i in range(max_token_seq):
-                mask = torch.zeros(tokenizer.vocab_size, dtype=torch.int64, device=model.device)
+                mask = np.zeros(tokenizer.vocab_size, dtype=np.int64)
                 if i == 0:
                     mask_ids = list(tokenizer.event_ids.values()) + [tokenizer.eos_id]
                     if disable_patch_change:
@@ -54,9 +80,9 @@ def generate(prompt=None, max_len=512, temp=1.0, top_p=0.98, top_k=20,
                     if param_name == "channel":
                         mask_ids = [i for i in mask_ids if i not in disable_channels]
                     mask[mask_ids] = 1
-                logits = model.forward_token(hidden, next_token_seq)[:, -1:]
-                scores = torch.softmax(logits / temp, dim=-1) * mask
-                sample = model.sample_top_p_k(scores, top_p, top_k)
+                logits = model[1].run(None, {'x': next_token_seq, "hidden": hidden})[0][:, -1:]
+                scores = softmax(logits / temp, -1) * mask
+                sample = sample_top_p_k(scores, top_p, top_k)
                 if i == 0:
                     next_token_seq = sample
                     eid = sample.item()
@@ -65,58 +91,29 @@ def generate(prompt=None, max_len=512, temp=1.0, top_p=0.98, top_k=20,
                         break
                     event_name = tokenizer.id_events[eid]
                 else:
-                    next_token_seq = torch.cat([next_token_seq, sample], dim=1)
+                    next_token_seq = np.concatenate([next_token_seq, sample], axis=1)
                     if len(tokenizer.events[event_name]) == i:
                         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)
+                next_token_seq = np.pad(next_token_seq, ((0, 0), (0, max_token_seq - next_token_seq.shape[-1])),
+                                        mode="constant", constant_values=tokenizer.pad_id)
+            next_token_seq = next_token_seq[None, :, :]
+            input_tensor = np.concatenate([input_tensor, next_token_seq], axis=1)
             cur_len += 1
             bar.update(1)
-            yield next_token_seq.reshape(-1).cpu().numpy()
+            yield next_token_seq.reshape(-1)
             if end:
                 break
 
 
-def run(tab, instruments, drum_kit, mid, midi_events, gen_events, temp, top_p, top_k, allow_cc, amp):
+def create_msg(name, data):
+    return {"name": name, "data": data}
+
+
+def run(model_name, tab, instruments, drum_kit, mid, midi_events, gen_events, temp, top_p, top_k, allow_cc):
     mid_seq = []
-    max_len = int(gen_events)
-    img_len = 1024
-    img = np.full((128 * 2, img_len, 3), 255, dtype=np.uint8)
-    state = {"t1": 0, "t": 0, "cur_pos": 0}
-    rand = np.random.RandomState(0)
-    colors = {(i, j): rand.randint(0, 200, 3) for i in range(128) for j in range(16)}
-
-    def draw_event(tokens):
-        if tokens[0] in tokenizer.id_events:
-            name = tokenizer.id_events[tokens[0]]
-            if len(tokens) <= len(tokenizer.events[name]):
-                return
-            params = tokens[1:]
-            params = [params[i] - tokenizer.parameter_ids[p][0] for i, p in enumerate(tokenizer.events[name])]
-            if not all([0 <= params[i] < tokenizer.event_parameters[p] for i, p in enumerate(tokenizer.events[name])]):
-                return
-            event = [name] + params
-            state["t1"] += event[1]
-            t = state["t1"] * 16 + event[2]
-            state["t"] = t
-            if name == "note":
-                tr, d, c, p = event[3:7]
-                shift = t + d - (state["cur_pos"] + img_len)
-                if shift > 0:
-                    img[:, :-shift] = img[:, shift:]
-                    img[:, -shift:] = 255
-                    state["cur_pos"] += shift
-                t = t - state["cur_pos"]
-                img[p * 2:(p + 1) * 2, t: t + d] = colors[(tr, c)]
-
-    def get_img():
-        t = state["t"] - state["cur_pos"]
-        img_new = img.copy()
-        img_new[:, t: t + 2] = 0
-        return PIL.Image.fromarray(np.flip(img_new, 0))
+    gen_events = int(gen_events)
+    max_len = gen_events
 
     disable_patch_change = False
     disable_channels = None
@@ -133,7 +130,7 @@ def run(tab, instruments, drum_kit, mid, midi_events, gen_events, temp, top_p, t
             mid.append(tokenizer.event2tokens(["patch_change", 0, 0, i, c, p]))
         mid_seq = mid
         mid = np.asarray(mid, dtype=np.int64)
-        if len(instruments) > 0 or drum_kit != "None":
+        if len(instruments) > 0:
             disable_patch_change = True
             disable_channels = [i for i in range(16) if i not in patches]
     elif mid is not None:
@@ -142,41 +139,67 @@ def run(tab, instruments, drum_kit, mid, midi_events, gen_events, temp, top_p, t
         mid = mid[:int(midi_events)]
         max_len += len(mid)
         for token_seq in mid:
-            mid_seq.append(token_seq)
-            draw_event(token_seq)
-    generator = generate(mid, max_len=max_len, temp=temp, top_p=top_p, top_k=top_k,
+            mid_seq.append(token_seq.tolist())
+    init_msgs = [create_msg("visualizer_clear", None)]
+    for tokens in mid_seq:
+        init_msgs.append(create_msg("visualizer_append", tokenizer.tokens2event(tokens)))
+    yield mid_seq, None, None, init_msgs
+    model = models[model_name]
+    generator = generate(model, mid, 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, amp=amp)
-    for token_seq in generator:
+                         disable_channels=disable_channels)
+    for i, token_seq in enumerate(generator):
+        token_seq = token_seq.tolist()
         mid_seq.append(token_seq)
-        draw_event(token_seq)
-        yield mid_seq, get_img(), None, None
+        event = tokenizer.tokens2event(token_seq)
+        yield mid_seq, None, None, [create_msg("visualizer_append", event), create_msg("progress", [i + 1, gen_events])]
     mid = tokenizer.detokenize(mid_seq)
     with open(f"output.mid", 'wb') as f:
         f.write(MIDI.score2midi(mid))
     audio = synthesis(MIDI.score2opus(mid), soundfont_path)
-    yield mid_seq, get_img(), "output.mid", (44100, audio)
+    yield mid_seq, "output.mid", (44100, audio), [create_msg("visualizer_end", None)]
 
 
 def cancel_run(mid_seq):
+    if mid_seq is None:
+        return None, None
     mid = tokenizer.detokenize(mid_seq)
     with open(f"output.mid", 'wb') as f:
         f.write(MIDI.score2midi(mid))
     audio = synthesis(MIDI.score2opus(mid), soundfont_path)
-    return "output.mid", (44100, audio)
+    return "output.mid", (44100, audio), [create_msg("visualizer_end", None)]
+
+
+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<!-- {path} --><script>{jsfile.read()}</script>"
+    template_response_ori = gr.routes.templates.TemplateResponse
+
+    def template_response(*args, **kwargs):
+        res = template_response_ori(*args, **kwargs)
+        res.body = res.body.replace(
+            b'</head>', f'{javascript}</head>'.encode("utf8"))
+        res.init_headers()
+        return res
 
+    gr.routes.templates.TemplateResponse = template_response
 
-def load_model(path):
-    ckpt = torch.load(path, map_location="cpu")
-    state_dict = ckpt.get("state_dict", ckpt)
-    model.load_state_dict(state_dict, strict=False)
-    model.eval()
-    return "success"
 
+class JSMsgReceiver(gr.HTML):
 
-def get_model_path():
-    model_paths = sorted(glob.glob("**/*.ckpt", recursive=True))
-    return model_path_input.update(choices=model_paths)
+    def __init__(self, **kwargs):
+        super().__init__(elem_id="msg_receiver", visible=False, **kwargs)
+
+    def postprocess(self, y):
+        if y:
+            y = f"<p>{json.dumps(y)}</p>"
+        return super().postprocess(y)
+
+    def get_block_name(self) -> str:
+        return "html"
 
 
 number2drum_kits = {-1: "None", 0: "Standard", 8: "Room", 16: "Power", 24: "Electric", 25: "TR-808", 32: "Jazz",
@@ -186,24 +209,38 @@ drum_kits2number = {v: k for k, v in number2drum_kits.items()}
 
 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")
-    soundfont_path = hf_hub_download(repo_id="skytnt/midi-model", filename="soundfont.sf2")
+    parser.add_argument("--max-gen", type=int, default=1024, help="max")
     opt = parser.parse_args()
+    soundfont_path = hf_hub_download(repo_id="skytnt/midi-model", filename="soundfont.sf2")
+    models_info = {"generic pretrain model": ["skytnt/midi-model", ""],
+                   "j-pop finetune model": ["skytnt/midi-model-ft", "jpop/"],
+                   "touhou finetune model": ["fspecii/ambp", "ambsd/"]}
+    models = {}
     tokenizer = MIDITokenizer()
-    model = MIDIModel(tokenizer).to(device=opt.device)
+    providers = ['CUDAExecutionProvider', 'CPUExecutionProvider']
+    for name, (repo_id, path) in models_info.items():
+        model_base_path = hf_hub_download(repo_id=repo_id, filename=f"{path}onnx/model_base.onnx")
+        model_token_path = hf_hub_download(repo_id=repo_id, filename=f"{path}onnx/model_token.onnx")
+        model_base = rt.InferenceSession(model_base_path, providers=providers)
+        model_token = rt.InferenceSession(model_token_path, providers=providers)
+        models[name] = [model_base, model_token]
 
+    load_javascript()
     app = gr.Blocks()
     with app:
-        with gr.Accordion(label="Model option", open=False):
-            load_model_path_btn = gr.Button("Get Models")
-            model_path_input = gr.Dropdown(label="model")
-            load_model_path_btn.click(get_model_path, [], model_path_input)
-            load_model_btn = gr.Button("Load")
-            model_msg = gr.Textbox()
-            load_model_btn.click(
-                load_model, model_path_input, model_msg
-            )
+        gr.Markdown("<h1 style='text-align: center; margin-bottom: 1rem'>Midi Composer</h1>")
+        gr.Markdown("![Visitors](https://api.visitorbadge.io/api/visitors?path=skytnt.midi-composer&style=flat)\n\n"
+                    "Midi event transformer for music generation\n\n"
+                    "Demo for [SkyTNT/midi-model](https://github.com/SkyTNT/midi-model)\n\n"
+                    "[Open In Colab]"
+                    "(https://colab.research.google.com/github/SkyTNT/midi-model/blob/main/demo.ipynb)"
+                    " for faster running and longer generation"
+                    )
+        js_msg = JSMsgReceiver()
+        input_model = gr.Dropdown(label="select model", choices=list(models.keys()),
+                                  type="value", value=list(models.keys())[0])
         tab_select = gr.Variable(value=0)
         with gr.Tabs():
             with gr.TabItem("instrument prompt") as tab1:
@@ -227,26 +264,28 @@ if __name__ == "__main__":
                 input_midi_events = gr.Slider(label="use first n midi events as prompt", minimum=1, maximum=512,
                                               step=1,
                                               value=128)
+                example2 = gr.Examples([[file, 128] for file in glob.glob("example/*.mid")],
+                                       [input_midi, input_midi_events])
 
         tab1.select(lambda: 0, None, tab_select, queue=False)
         tab2.select(lambda: 1, None, tab_select, queue=False)
-        input_gen_events = gr.Slider(label="generate n midi events", minimum=1, maximum=4096, step=1, value=512)
+        input_gen_events = gr.Slider(label="generate n midi events", minimum=1, maximum=opt.max_gen,
+                                     step=1, value=opt.max_gen // 2)
         with gr.Accordion("options", open=False):
             input_temp = gr.Slider(label="temperature", minimum=0.1, maximum=1.2, step=0.01, value=1)
             input_top_p = gr.Slider(label="top p", minimum=0.1, maximum=1, step=0.01, value=0.98)
             input_top_k = gr.Slider(label="top k", minimum=1, maximum=20, step=1, value=12)
             input_allow_cc = gr.Checkbox(label="allow midi cc event", value=True)
-            input_amp = gr.Checkbox(label="enable amp", value=True)
             example3 = gr.Examples([[1, 0.98, 12], [1.2, 0.95, 8]], [input_temp, input_top_p, input_top_k])
         run_btn = gr.Button("generate", variant="primary")
         stop_btn = gr.Button("stop and output")
         output_midi_seq = gr.Variable()
-        output_midi_img = gr.Image(label="output image")
+        output_midi_visualizer = gr.HTML(elem_id="midi_visualizer_container")
+        output_audio = gr.Audio(label="output audio", format="mp3", elem_id="midi_audio")
         output_midi = gr.File(label="output midi", file_types=[".mid"])
-        output_audio = gr.Audio(label="output audio", format="mp3")
-        run_event = run_btn.click(run, [tab_select, input_instruments, input_drum_kit, input_midi, input_midi_events,
-                                        input_gen_events, input_temp, input_top_p, input_top_k,
-                                        input_allow_cc, input_amp],
-                                  [output_midi_seq, output_midi_img, output_midi, output_audio])
-        stop_btn.click(cancel_run, output_midi_seq, [output_midi, output_audio], cancels=run_event, queue=False)
-    app.queue(1).launch(server_port=opt.port)
+        run_event = run_btn.click(run, [input_model, tab_select, input_instruments, input_drum_kit, input_midi,
+                                        input_midi_events, input_gen_events, input_temp, input_top_p, input_top_k,
+                                        input_allow_cc],
+                                  [output_midi_seq, output_midi, output_audio, js_msg])
+        stop_btn.click(cancel_run, output_midi_seq, [output_midi, output_audio, js_msg], cancels=run_event, queue=False)
+    app.queue(2).launch(server_port=opt.port, share=opt.share, inbrowser=True)