optimize(infer): move onnx into rvc

Author: fumiama

infer/lib/infer_pack/models_onnx.py

@@ -1 +1,2 @@
 from .infer import RVC
+from .exporter import export_onnx

rvc/onnx/__init__.py

@@ -1,10 +1,10 @@
 import torch
 
-from infer.lib.infer_pack.models_onnx import SynthesizerTrnMsNSFsidM
+from .synthesizer import SynthesizerTrnMsNSFsid
 
 
-def export_onnx(ModelPath, ExportedPath):
-    cpt = torch.load(ModelPath, map_location="cpu")
+def export_onnx(from_cpkt_pth: str, to_onnx_pth: str) -> str:
+    cpt = torch.load(from_cpkt_pth, map_location="cpu")
     cpt["config"][-3] = cpt["weight"]["emb_g.weight"].shape[0]
     vec_channels = 256 if cpt.get("version", "v1") == "v1" else 768
 
@@ -17,8 +17,8 @@ def export_onnx(ModelPath, ExportedPath):
 
     device = "cpu"  # 导出时设备（不影响使用模型）
 
-    net_g = SynthesizerTrnMsNSFsidM(
-        *cpt["config"], is_half=False, encoder_dim=vec_channels
+    net_g = SynthesizerTrnMsNSFsid(
+        *cpt["config"], encoder_dim=vec_channels
     )  # fp32导出（C++要支持fp16必须手动将内存重新排列所以暂时不用fp16）
     net_g.load_state_dict(cpt["weight"], strict=False)
     input_names = ["phone", "phone_lengths", "pitch", "pitchf", "ds", "rnd"]
@@ -36,7 +36,7 @@ def export_onnx(ModelPath, ExportedPath):
             test_ds.to(device),
             test_rnd.to(device),
         ),
-        ExportedPath,
+        to_onnx_pth,
         dynamic_axes={
             "phone": [1],
             "pitch": [1],

infer/modules/onnx/export.py renamed to rvc/onnx/exporter.py

@@ -23,7 +23,7 @@ def compute_f0(self, wav: np.ndarray[Any, np.dtype], p_len: Optional[int] = None
         f0 = pyworld.stonemask(wav.astype(np.double), f0, t, self.sampling_rate)
         for index, pitch in enumerate(f0):
             f0[index] = round(pitch, 1)
-        return self.__interpolate_f0(self.__resize_f0(f0, p_len))[0]
+        return self.interpolate_f0(self.resize_f0(f0, p_len))[0]
 
     def compute_f0_uv(
         self, wav: np.ndarray[Any, np.dtype], p_len: Optional[int] = None
@@ -40,4 +40,4 @@ def compute_f0_uv(
         f0 = pyworld.stonemask(wav.astype(np.double), f0, t, self.sampling_rate)
         for index, pitch in enumerate(f0):
             f0[index] = round(pitch, 1)
-        return self.__interpolate_f0(self.__resize_f0(f0, p_len))
+        return self.interpolate_f0(self.resize_f0(f0, p_len))

rvc/onnx/f0predictors/dio.py

@@ -18,7 +18,7 @@ def compute_f0_uv(
         self, wav: np.ndarray[Any, np.dtype], p_len: Optional[int] = None
     ): ...
 
-    def __interpolate_f0(self, f0: np.ndarray[Any, np.dtype]):
+    def interpolate_f0(self, f0: np.ndarray[Any, np.dtype]):
         """
         对F0进行插值处理
         """
@@ -56,7 +56,7 @@ def __interpolate_f0(self, f0: np.ndarray[Any, np.dtype]):
 
         return ip_data[:, 0], vuv_vector[:, 0]
 
-    def __resize_f0(self, x: np.ndarray[Any, np.dtype], target_len: int):
+    def resize_f0(self, x: np.ndarray[Any, np.dtype], target_len: int):
         source = np.array(x)
         source[source < 0.001] = np.nan
         target = np.interp(

rvc/onnx/f0predictors/f0.py

@@ -21,7 +21,7 @@ def compute_f0(self, wav: np.ndarray[Any, np.dtype], p_len: Optional[int] = None
             frame_period=1000 * self.hop_length / self.sampling_rate,
         )
         f0 = pyworld.stonemask(wav.astype(np.double), f0, t, self.fs)
-        return self.__interpolate_f0(self.__resize_f0(f0, p_len))[0]
+        return self.interpolate_f0(self.resize_f0(f0, p_len))[0]
 
     def compute_f0_uv(
         self, wav: np.ndarray[Any, np.dtype], p_len: Optional[int] = None
@@ -36,4 +36,4 @@ def compute_f0_uv(
             frame_period=1000 * self.hop_length / self.sampling_rate,
         )
         f0 = pyworld.stonemask(wav.astype(np.double), f0, t, self.sampling_rate)
-        return self.__interpolate_f0(self.__resize_f0(f0, p_len))
+        return self.interpolate_f0(self.resize_f0(f0, p_len))

rvc/onnx/f0predictors/harvest.py

@@ -31,7 +31,7 @@ def compute_f0(self, wav: np.ndarray[Any, np.dtype], p_len: Optional[int] = None
         pad_size = (p_len - len(f0) + 1) // 2
         if pad_size > 0 or p_len - len(f0) - pad_size > 0:
             f0 = np.pad(f0, [[pad_size, p_len - len(f0) - pad_size]], mode="constant")
-        f0, uv = self.__interpolate_f0(f0)
+        f0, uv = self.interpolate_f0(f0)
         return f0
 
     def compute_f0_uv(
@@ -57,5 +57,5 @@ def compute_f0_uv(
         pad_size = (p_len - len(f0) + 1) // 2
         if pad_size > 0 or p_len - len(f0) - pad_size > 0:
             f0 = np.pad(f0, [[pad_size, p_len - len(f0) - pad_size]], mode="constant")
-        f0, uv = self.__interpolate_f0(f0)
+        f0, uv = self.interpolate_f0(f0)
         return f0, uv

rvc/onnx/f0predictors/pm.py

@@ -1,10 +1,11 @@
+import typing
+import os
+
 import librosa
 import numpy as np
 import onnxruntime
-import typing
-import os
 
-from onnx.f0predictors import (
+from .f0predictors import (
     PMF0Predictor,
     HarvestF0Predictor,
     DioF0Predictor,
@@ -15,7 +16,7 @@
 class Model:
     def __init__(
         self,
-        path: str | bytes | os.PathLike,
+        path: typing.Union[str, bytes, os.PathLike],
         device: typing.Literal["cpu", "cuda", "dml"] = "cpu",
     ):
         if device == "cpu":
@@ -32,7 +33,7 @@ def __init__(
 class ContentVec(Model):
     def __init__(
         self,
-        vec_path: str | bytes | os.PathLike,
+        vec_path: typing.Union[str, bytes, os.PathLike],
         device: typing.Literal["cpu", "cuda", "dml"] = "cpu",
     ):
         super().__init__(vec_path, device)
@@ -66,9 +67,9 @@ def get_f0_predictor(
 class RVC(Model):
     def __init__(
         self,
-        model_path: str | bytes | os.PathLike,
+        model_path: typing.Union[str, bytes, os.PathLike],
         hop_len=512,
-        vec_path: str | bytes | os.PathLike = "vec-768-layer-12.onnx",
+        vec_path: typing.Union[str, bytes, os.PathLike] = "vec-768-layer-12.onnx",
         device: typing.Literal["cpu", "cuda", "dml"] = "cpu",
     ):
         super().__init__(model_path, device)

rvc/onnx/infer.py

@@ -0,0 +1,80 @@
+from typing import List, Optional, Union
+
+import torch
+
+from rvc.layers.synthesizers import SynthesizerTrnMsNSFsid as SynthesizerBase
+
+
+class SynthesizerTrnMsNSFsid(SynthesizerBase):
+    def __init__(
+        self,
+        spec_channels: int,
+        segment_size: int,
+        inter_channels: int,
+        hidden_channels: int,
+        filter_channels: int,
+        n_heads: int,
+        n_layers: int,
+        kernel_size: int,
+        p_dropout: int,
+        resblock: str,
+        resblock_kernel_sizes: List[int],
+        resblock_dilation_sizes: List[List[int]],
+        upsample_rates: List[int],
+        upsample_initial_channel: int,
+        upsample_kernel_sizes: List[int],
+        spk_embed_dim: int,
+        gin_channels: int,
+        sr: Optional[Union[str, int]],
+        encoder_dim: int,
+    ):
+        super().__init__(
+            spec_channels,
+            segment_size,
+            inter_channels,
+            hidden_channels,
+            filter_channels,
+            n_heads,
+            n_layers,
+            kernel_size,
+            p_dropout,
+            resblock,
+            resblock_kernel_sizes,
+            resblock_dilation_sizes,
+            upsample_rates,
+            upsample_initial_channel,
+            upsample_kernel_sizes,
+            spk_embed_dim,
+            gin_channels,
+            sr,
+            encoder_dim,
+            True,
+        )
+        self.speaker_map = None
+
+    def remove_weight_norm(self):
+        self.dec.remove_weight_norm()
+        self.flow.remove_weight_norm()
+        self.enc_q.remove_weight_norm()
+
+    def construct_spkmixmap(self):
+        self.speaker_map = torch.zeros((self.n_speaker, 1, 1, self.gin_channels))
+        for i in range(self.n_speaker):
+            self.speaker_map[i] = self.emb_g(torch.LongTensor([[i]]))
+        self.speaker_map = self.speaker_map.unsqueeze(0)
+
+    def forward(self, phone, phone_lengths, pitch, nsff0, g, rnd, max_len=None):
+        if self.speaker_map is not None:  # [N, S]  *  [S, B, 1, H]
+            g = g.reshape((g.shape[0], g.shape[1], 1, 1, 1))  # [N, S, B, 1, 1]
+            g = g * self.speaker_map  # [N, S, B, 1, H]
+            g = torch.sum(g, dim=1)  # [N, 1, B, 1, H]
+            g = g.transpose(0, -1).transpose(0, -2).squeeze(0)  # [B, H, N]
+        else:
+            g = g.unsqueeze(0)
+            g = self.emb_g(g).transpose(1, 2)
+
+        m_p, logs_p, x_mask = self.enc_p(phone, pitch, phone_lengths)
+        z_p = (m_p + torch.exp(logs_p) * rnd) * x_mask
+        z = self.flow(z_p, x_mask, g=g, reverse=True)
+        o = self.dec((z * x_mask)[:, :, :max_len], nsff0, g=g)
+        return o

rvc/onnx/synthesizer.py

@@ -1,3 +1,3 @@
-from infer.modules.onnx.export import export_onnx
+from rvc.onnx import export_onnx
 
 export_onnx("pt/Justin Bieber.pth", "pt/TestRvc_Rvc.onnx")