Merge branch 'openvpi:master' into master

Author: oxygen-dioxide

acoustic/dfs_models.py

@@ -0,0 +1,19 @@
+from usr.diff.shallow_diffusion_tts import GaussianDiffusion
+
+import torch
+
+device = 'cpu'
+
+
+class GaussianDiffusionFS(GaussianDiffusion):
+    def forward(self, txt_tokens, mel2ph=None, spk_embed=None,
+                ref_mels=None, f0=None, uv=None, energy=None, infer=False, **kwargs):
+        ret = self.fs2(txt_tokens, mel2ph, spk_embed, ref_mels, f0, uv, energy,
+                       skip_decoder=True, infer=infer, **kwargs)
+        return ret['decoder_inp']
+
+
+class GaussianDiffusionDenoise(GaussianDiffusion):
+    def forward(self, x, t, cond):
+        x = self.p_sample(x, t, cond)
+        return [x, cond]

onnx_export_singer.py

@@ -9,7 +9,10 @@
 from utils.audio import save_wav
 from utils.hparams import set_hparams, hparams
 
+import acoustic.dfs_models as adm
+
 import torch
+import numpy as np
 
 from utils.text_encoder import TokenTextEncoder
 from usr.diffsinger_task import DIFF_DECODERS
@@ -26,19 +29,30 @@
 ]
 
 
-class GaussianDiffusionWrap(GaussianDiffusion):
-    def forward(self, txt_tokens, mel2ph,
+class GaussianDiffusionWrap(adm.GaussianDiffusionFS):
+    def forward(self, txt_tokens,
                 # Wrapped Arguments
                 spk_id,
                 pitch_midi,
                 midi_dur,
                 is_slur,
+                mel2ph,
                 ):
 
+        print(f"txt_tokens: {txt_tokens}")
+        print(f"spk_id: {spk_id}")
+        print(f"pitch_midi: {pitch_midi}")
+        print(f"midi_dur: {midi_dur}")
+        print(f"is_slur: {is_slur}")
+        print(f"mel2ph: {mel2ph}")
+
+        if (mel2ph[0].item() == 0):
+            mel2ph = None
+        else:
+            mel2ph = mel2ph[1].item()
+        
         if (torch.numel(txt_tokens) == 0):
             txt_tokens = None
-        if (torch.numel(mel2ph) == 0):
-            mel2ph = None
         if (torch.numel(spk_id) == 0):
             spk_id = None
         if (torch.numel(pitch_midi) == 0):
@@ -57,7 +71,8 @@ class DFSInferWrapped(e2e.DiffSingerE2EInfer):
     def build_model(self):
         model = GaussianDiffusionWrap(
             phone_encoder=self.ph_encoder,
-            out_dims=hparams['audio_num_mel_bins'], denoise_fn=DIFF_DECODERS[hparams['diff_decoder_type']](hparams),
+            out_dims=hparams['audio_num_mel_bins'], denoise_fn=DIFF_DECODERS[hparams['diff_decoder_type']](
+                hparams),
             timesteps=hparams['timesteps'],
             K_step=hparams['K_step'],
             loss_type=hparams['diff_loss_type'],
@@ -71,9 +86,33 @@ def build_model(self):
             self.pe = PitchExtractor().to(self.device)
             load_ckpt(self.pe, hparams['pe_ckpt'], 'model', strict=True)
             self.pe.eval()
-        
+
         return model
 
+
+class DFSInferWrapped2(e2e.DiffSingerE2EInfer):
+    def build_model(self):
+        model = adm.GaussianDiffusionDenoise(
+            phone_encoder=self.ph_encoder,
+            out_dims=hparams['audio_num_mel_bins'], denoise_fn=DIFF_DECODERS[hparams['diff_decoder_type']](
+                hparams),
+            timesteps=hparams['timesteps'],
+            K_step=hparams['K_step'],
+            loss_type=hparams['diff_loss_type'],
+            spec_min=hparams['spec_min'], spec_max=hparams['spec_max'],
+        )
+
+        model.eval()
+        load_ckpt(model, hparams['work_dir'], 'model')
+
+        if hparams.get('pe_enable') is not None and hparams['pe_enable']:
+            self.pe = PitchExtractor().to(self.device)
+            load_ckpt(self.pe, hparams['pe_ckpt'], 'model', strict=True)
+            self.pe.eval()
+
+        return model
+
+
 if __name__ == '__main__':
 
     inp = {
@@ -90,25 +129,43 @@ def build_model(self):
     infer_ins = DFSInferWrapped(hparams)
     infer_ins.model.to(dev)
 
+    infer_ins2 = DFSInferWrapped2(hparams)
+    infer_ins2.model.to(dev)
+
+    adm.device = dev
+
     with torch.no_grad():
-        inp = infer_ins.preprocess_input(inp, input_type=inp['input_type'] if inp.get('input_type') else 'word')
+        inp = infer_ins.preprocess_input(
+            inp, input_type=inp['input_type'] if inp.get('input_type') else 'word')
         sample = infer_ins.input_to_batch(inp)
         txt_tokens = sample['txt_tokens']  # [B, T_t]
         spk_id = sample.get('spk_ids')
 
+        print(txt_tokens)
+        print(spk_id)
+        print(sample['pitch_midi'])
+        print(sample['midi_dur'])
+        print(sample['is_slur'])
+        print(sample['mel2ph'])
+
         torch.onnx.export(
             infer_ins.model,
             (
                 txt_tokens.to(dev),
-                {
-                    'spk_id': spk_id.to(dev),
-                    'pitch_midi': sample['pitch_midi'].to(dev),
-                    'midi_dur': sample['midi_dur'].to(dev),
-                    'is_slur': spk_id.to(dev),
-                    'mel2ph': spk_id.to(dev)
-                }
+                # {
+                #     'spk_id': spk_id.to(dev),
+                #     'pitch_midi': sample['pitch_midi'].to(dev),
+                #     'midi_dur': sample['midi_dur'].to(dev),
+                #     'is_slur': spk_id.to(dev),
+                #     'mel2ph': spk_id.to(dev)
+                # }
+                spk_id.to(dev),
+                sample['pitch_midi'].to(dev),
+                sample['midi_dur'].to(dev),
+                sample['is_slur'].to(dev),
+                torch.from_numpy(np.array([0, 0]).astype(np.int64)).to(dev),
             ),
-            "singer.onnx",
+            "singer_fs.onnx",
             # verbose=True,
             input_names=["txt_tokens", "spk_id",
                          "pitch_midi", "midi_dur", "is_slur", "mel2ph"],
@@ -132,10 +189,41 @@ def build_model(self):
                 "is_slur": {
                     0: "a",
                     1: "b",
+                }
+            },
+            opset_version=11
+        )
+
+        # fs_res = infer_ins.model(txt_tokens, spk_id=spk_id, ref_mels=None, infer=True,
+        #                          pitch_midi=sample['pitch_midi'], midi_dur=sample['midi_dur'],
+        #                          is_slur=sample['is_slur'], mel2ph=sample['mel2ph'])
+        # cond = fs_res.transpose(1, 2)
+        # shape = (cond.shape[0], 1, infer_ins.model.mel_bins, cond.shape[2])
+        # x = torch.randn(shape, device=dev)
+
+        torch.onnx.export(
+            infer_ins2.model,
+            (
+                torch.rand(1, 1, 80, 967).to(dev),
+                torch.full((1,), 1, dtype=torch.long).to(dev),
+                torch.rand(1, 256, 967).to(dev),
+            ),
+            "singer_denoise.onnx",
+            input_names=[
+                "x",
+                "t",
+                "cond",
+            ],
+            dynamic_axes={
+                "x": {
+                    0: "batch_size",
+                    2: "num_mel_bin",
+                    3: "frames",
                 },
-                "mel2ph": {
-                    0: "a",
-                    1: "b",
+                "cond": {
+                    0: "batch_size",
+                    1: "what",
+                    2: "frames",
                 }
             },
             opset_version=11

onnx_test_pe_gpu.py

@@ -0,0 +1,117 @@
+# coding=utf8
+
+import os
+import sys
+import inference.svs.ds_e2e as e2e
+from utils.audio import save_wav
+from utils.hparams import set_hparams, hparams
+
+import numpy as np
+
+import torch
+import onnxruntime as ort
+
+root_dir = os.path.dirname(os.path.abspath(__file__))
+os.environ['PYTHONPATH'] = f'"{root_dir}"'
+
+sys.argv = [
+    f'{root_dir}/inference/svs/ds_e2e.py',
+    '--config',
+    f'{root_dir}/usr/configs/midi/e2e/opencpop/ds100_adj_rel.yaml',
+    '--exp_name',
+    '0228_opencpop_ds100_rel'
+]
+
+
+def to_numpy(tensor):
+    return tensor.detach().cpu().numpy() if tensor.requires_grad else tensor.cpu().numpy()
+
+
+class TestHifiganInfer(e2e.DiffSingerE2EInfer):
+    def __init__(self, hparams, device=None):
+        super().__init__(hparams, device)
+
+        self.pe2 = ort.InferenceSession("xiaoma_pe.onnx", providers=["CUDAExecutionProvider"])
+        self.vocoder2 = ort.InferenceSession("hifigan.onnx", providers=["CUDAExecutionProvider"])
+
+    def run_vocoder(self, c, **kwargs):
+        c = c.transpose(2, 1)  # [B, 80, T]
+        f0 = kwargs.get('f0')  # [B, T]
+
+        if f0 is not None and hparams.get('use_nsf'):
+            ort_inputs = {
+                'x': to_numpy(c),
+                'f0': to_numpy(f0)
+            }
+        else:
+            ort_inputs = {
+                'x': to_numpy(c),
+                'f0': {}
+            }
+            # [T]
+
+        ort_out = self.vocoder2.run(None, ort_inputs)
+        y = torch.from_numpy(ort_out[0]).to(self.device)
+
+        return y[None]
+
+    def forward_model(self, inp):
+        sample = self.input_to_batch(inp)
+        txt_tokens = sample['txt_tokens']  # [B, T_t]
+        spk_id = sample.get('spk_ids')
+
+        print(txt_tokens.shape)
+        print(spk_id.shape)
+        print(sample['pitch_midi'].shape)
+        print(sample['midi_dur'].shape)
+        if (sample['is_slur'] is not None):
+            print(sample['is_slur'].shape)
+        if (sample['mel2ph'] is not None):
+            print(sample['mel2ph'].shape)
+
+        with torch.no_grad():
+            output = self.model(txt_tokens, spk_id=spk_id, ref_mels=None, infer=True,
+                                pitch_midi=sample['pitch_midi'], midi_dur=sample['midi_dur'],
+                                is_slur=sample['is_slur'], mel2ph=sample['mel2ph'])
+
+            mel_out = output['mel_out']  # [B, T,80]
+
+            if hparams.get('pe_enable') is not None and hparams['pe_enable']:
+                pe2_res = self.pe2.run(None, 
+                    {
+                        'mel_input': to_numpy(mel_out)
+                    }
+                )
+                
+                # pe predict from Pred mel
+                f0_pred = torch.from_numpy(pe2_res[1])
+
+            else:
+                f0_pred = output['f0_denorm']
+
+            # Run Vocoder
+            wav_out = self.run_vocoder(mel_out, f0=f0_pred)
+        wav_out = wav_out.cpu().numpy()
+        return wav_out[0]
+
+
+if __name__ == '__main__':
+    c = {
+        'text': '小酒窝长睫毛AP是你最美的记号',
+        'notes': 'C#4/Db4 | F#4/Gb4 | G#4/Ab4 | A#4/Bb4 F#4/Gb4 | F#4/Gb4 C#4/Db4 | C#4/Db4 | rest | C#4/Db4 | A#4/Bb4 | G#4/Ab4 | A#4/Bb4 | G#4/Ab4 | F4 | C#4/Db4',
+        'notes_duration': '0.407140 | 0.376190 | 0.242180 | 0.509550 0.183420 | 0.315400 0.235020 | 0.361660 | 0.223070 | 0.377270 | 0.340550 | 0.299620 | 0.344510 | 0.283770 | 0.323390 | 0.360340',
+        'input_type': 'word'
+    }  # user input: Chinese characters
+
+    target = "./infer_out/onnx_test_res.wav"
+
+    set_hparams(print_hparams=False)
+    infer_ins = TestHifiganInfer(hparams)
+
+    out = infer_ins.infer_once(c)
+    os.makedirs(os.path.dirname(target), exist_ok=True)
+    print(f'| save audio: {target}')
+    save_wav(out, target, hparams['audio_sample_rate'])
+
+    print(infer_ins.pe)
+    print("OK")