WIP:common sample

server/restapi/MMVC_Rest_Fileuploader.py

@@ -24,61 +24,58 @@ class MMVC_Rest_Fileuploader:
         self.router = APIRouter()
         self.router.add_api_route("/info", self.get_info, methods=["GET"])
         self.router.add_api_route("/performance", self.get_performance, methods=["GET"])
-        self.router.add_api_route(
+        self.router.add_api_route("/upload_file", self.post_upload_file, methods=["POST"])
-            "/upload_file", self.post_upload_file, methods=["POST"]
+        self.router.add_api_route("/concat_uploaded_file", self.post_concat_uploaded_file, methods=["POST"])
-        )
+        self.router.add_api_route("/update_settings", self.post_update_settings, methods=["POST"])
-        self.router.add_api_route(
-            "/concat_uploaded_file", self.post_concat_uploaded_file, methods=["POST"]
-        )
-        self.router.add_api_route(
-            "/update_settings", self.post_update_settings, methods=["POST"]
-        )
         self.router.add_api_route("/load_model", self.post_load_model, methods=["POST"])
         self.router.add_api_route("/model_type", self.post_model_type, methods=["POST"])
         self.router.add_api_route("/model_type", self.get_model_type, methods=["GET"])
         self.router.add_api_route("/onnx", self.get_onnx, methods=["GET"])
-        self.router.add_api_route(
+        self.router.add_api_route("/merge_model", self.post_merge_models, methods=["POST"])
-            "/merge_model", self.post_merge_models, methods=["POST"]
+        self.router.add_api_route("/update_model_default", self.post_update_model_default, methods=["POST"])
-        )
+        self.router.add_api_route("/update_model_info", self.post_update_model_info, methods=["POST"])
-        self.router.add_api_route(
+        self.router.add_api_route("/upload_model_assets", self.post_upload_model_assets, methods=["POST"])
-            "/update_model_default", self.post_update_model_default, methods=["POST"]
-        )
-        self.router.add_api_route(
-            "/update_model_info", self.post_update_model_info, methods=["POST"]
-        )
-        self.router.add_api_route(
-            "/upload_model_assets", self.post_upload_model_assets, methods=["POST"]
-        )
 
     def post_upload_file(self, file: UploadFile = File(...), filename: str = Form(...)):
-        res = upload_file(UPLOAD_DIR, file, filename)
+        try:
-        json_compatible_item_data = jsonable_encoder(res)
+            res = upload_file(UPLOAD_DIR, file, filename)
-        return JSONResponse(content=json_compatible_item_data)
+            json_compatible_item_data = jsonable_encoder(res)
+            return JSONResponse(content=json_compatible_item_data)
+        except Exception as e:
+            print("[Voice Changer] ", e)
 
-    def post_concat_uploaded_file(
+    def post_concat_uploaded_file(self, filename: str = Form(...), filenameChunkNum: int = Form(...)):
-        self, filename: str = Form(...), filenameChunkNum: int = Form(...)
+        try:
-    ):
+            res = concat_file_chunks(UPLOAD_DIR, filename, filenameChunkNum, UPLOAD_DIR)
-        res = concat_file_chunks(UPLOAD_DIR, filename, filenameChunkNum, UPLOAD_DIR)
+            json_compatible_item_data = jsonable_encoder(res)
-        json_compatible_item_data = jsonable_encoder(res)
+            return JSONResponse(content=json_compatible_item_data)
-        return JSONResponse(content=json_compatible_item_data)
+        except Exception as e:
+            print("[Voice Changer] ", e)
 
     def get_info(self):
-        info = self.voiceChangerManager.get_info()
+        try:
-        json_compatible_item_data = jsonable_encoder(info)
+            info = self.voiceChangerManager.get_info()
-        return JSONResponse(content=json_compatible_item_data)
+            json_compatible_item_data = jsonable_encoder(info)
+            return JSONResponse(content=json_compatible_item_data)
+        except Exception as e:
+            print("[Voice Changer] ", e)
 
     def get_performance(self):
-        info = self.voiceChangerManager.get_performance()
+        try:
-        json_compatible_item_data = jsonable_encoder(info)
+            info = self.voiceChangerManager.get_performance()
-        return JSONResponse(content=json_compatible_item_data)
+            json_compatible_item_data = jsonable_encoder(info)
+            return JSONResponse(content=json_compatible_item_data)
+        except Exception as e:
+            print("[Voice Changer] ", e)
 
-    def post_update_settings(
+    def post_update_settings(self, key: str = Form(...), val: Union[int, str, float] = Form(...)):
-        self, key: str = Form(...), val: Union[int, str, float] = Form(...)
+        try:
-    ):
+            print("[Voice Changer] update configuration:", key, val)
-        print("[Voice Changer] update configuration:", key, val)
+            info = self.voiceChangerManager.update_settings(key, val)
-        info = self.voiceChangerManager.update_settings(key, val)
+            json_compatible_item_data = jsonable_encoder(info)
-        json_compatible_item_data = jsonable_encoder(info)
+            return JSONResponse(content=json_compatible_item_data)
-        return JSONResponse(content=json_compatible_item_data)
+        except Exception as e:
+            print("[Voice Changer] ", e)
 
     def post_load_model(
         self,
@@ -86,61 +83,83 @@ class MMVC_Rest_Fileuploader:
         isHalf: bool = Form(...),
         params: str = Form(...),
     ):
-        paramDict = json.loads(params)
+        try:
-        # print("paramDict", paramDict)
+            paramDict = json.loads(params)
+            # print("paramDict", paramDict)
 
-        # Change Filepath
+            # Change Filepath
-        newFilesDict = {}
+            newFilesDict = {}
-        for key, val in paramDict["files"].items():
+            for key, val in paramDict["files"].items():
-            if val != "-" and val != "":
+                if val != "-" and val != "":
-                uploadPath = os.path.join(UPLOAD_DIR, val)
+                    uploadPath = os.path.join(UPLOAD_DIR, val)
-                storePath = os.path.join(UPLOAD_DIR, f"{slot}", val)
+                    storePath = os.path.join(UPLOAD_DIR, f"{slot}", val)
-                storeDir = os.path.dirname(storePath)
+                    storeDir = os.path.dirname(storePath)
-                os.makedirs(storeDir, exist_ok=True)
+                    os.makedirs(storeDir, exist_ok=True)
-                shutil.move(uploadPath, storePath)
+                    shutil.move(uploadPath, storePath)
-                newFilesDict[key] = storePath
+                    newFilesDict[key] = storePath
-        paramDict["files"] = newFilesDict
+            paramDict["files"] = newFilesDict
 
-        props: LoadModelParams = LoadModelParams(
+            props: LoadModelParams = LoadModelParams(slot=slot, isHalf=isHalf, params=paramDict)
-            slot=slot, isHalf=isHalf, params=paramDict
-        )
 
-        info = self.voiceChangerManager.loadModel(props)
+            info = self.voiceChangerManager.loadModel(props)
-        json_compatible_item_data = jsonable_encoder(info)
+            json_compatible_item_data = jsonable_encoder(info)
-        return JSONResponse(content=json_compatible_item_data)
+            return JSONResponse(content=json_compatible_item_data)
+        except Exception as e:
+            print("[Voice Changer] ", e)
 
     def post_model_type(self, modelType: ModelType = Form(...)):
-        info = self.voiceChangerManager.switchModelType(modelType)
+        try:
-        json_compatible_item_data = jsonable_encoder(info)
+            info = self.voiceChangerManager.switchModelType(modelType)
-        return JSONResponse(content=json_compatible_item_data)
+            json_compatible_item_data = jsonable_encoder(info)
+            return JSONResponse(content=json_compatible_item_data)
+        except Exception as e:
+            print("[Voice Changer] ", e)
 
     def get_model_type(self):
-        info = self.voiceChangerManager.getModelType()
+        try:
-        json_compatible_item_data = jsonable_encoder(info)
+            info = self.voiceChangerManager.getModelType()
-        return JSONResponse(content=json_compatible_item_data)
+            json_compatible_item_data = jsonable_encoder(info)
+            return JSONResponse(content=json_compatible_item_data)
+        except Exception as e:
+            print("[Voice Changer] ", e)
 
     def get_onnx(self):
-        info = self.voiceChangerManager.export2onnx()
+        try:
-        json_compatible_item_data = jsonable_encoder(info)
+            info = self.voiceChangerManager.export2onnx()
-        return JSONResponse(content=json_compatible_item_data)
+            json_compatible_item_data = jsonable_encoder(info)
+            return JSONResponse(content=json_compatible_item_data)
+        except Exception as e:
+            print("[Voice Changer] ", e)
 
     def post_merge_models(self, request: str = Form(...)):
-        print(request)
+        try:
-        info = self.voiceChangerManager.merge_models(request)
+            print(request)
-        json_compatible_item_data = jsonable_encoder(info)
+            info = self.voiceChangerManager.merge_models(request)
-        return JSONResponse(content=json_compatible_item_data)
+            json_compatible_item_data = jsonable_encoder(info)
+            return JSONResponse(content=json_compatible_item_data)
+        except Exception as e:
+            print("[Voice Changer] ", e)
 
     def post_update_model_default(self):
-        info = self.voiceChangerManager.update_model_default()
+        try:
-        json_compatible_item_data = jsonable_encoder(info)
+            info = self.voiceChangerManager.update_model_default()
-        return JSONResponse(content=json_compatible_item_data)
+            json_compatible_item_data = jsonable_encoder(info)
+            return JSONResponse(content=json_compatible_item_data)
+        except Exception as e:
+            print("[Voice Changer] ", e)
 
     def post_update_model_info(self, newData: str = Form(...)):
-        info = self.voiceChangerManager.update_model_info(newData)
+        try:
-        json_compatible_item_data = jsonable_encoder(info)
+            info = self.voiceChangerManager.update_model_info(newData)
-        return JSONResponse(content=json_compatible_item_data)
+            json_compatible_item_data = jsonable_encoder(info)
+            return JSONResponse(content=json_compatible_item_data)
+        except Exception as e:
+            print("[Voice Changer] ", e)
 
     def post_upload_model_assets(self, params: str = Form(...)):
-        info = self.voiceChangerManager.upload_model_assets(params)
+        try:
-        json_compatible_item_data = jsonable_encoder(info)
+            info = self.voiceChangerManager.upload_model_assets(params)
-        return JSONResponse(content=json_compatible_item_data)
+            json_compatible_item_data = jsonable_encoder(info)
+            return JSONResponse(content=json_compatible_item_data)
+        except Exception as e:
+            print("[Voice Changer] ", e)

server/voice_changer/ModelSlotManager.py

@@ -0,0 +1,60 @@
+from const import UPLOAD_DIR
+from data.ModelSlot import ModelSlots, loadAllSlotInfo, loadSlotInfo, saveSlotInfo
+from voice_changer.utils.VoiceChangerParams import VoiceChangerParams
+import json
+import os
+import shutil
+
+
+class ModelSlotManager:
+    _instance = None
+
+    def __init__(self, params: VoiceChangerParams):
+        self.params = params
+        self.modelSlots = loadAllSlotInfo(self.params.model_dir)
+
+    @classmethod
+    def get_instance(cls, params: VoiceChangerParams):
+        if cls._instance is None:
+            cls._instance = cls(params)
+        return cls._instance
+
+    def _save_model_slot(self, slotIndex: int, slotInfo: ModelSlots):
+        saveSlotInfo(self.params.model_dir, slotIndex, slotInfo)
+        self.modelSlots = loadAllSlotInfo(self.params.model_dir)
+
+    def _load_model_slot(self, slotIndex: int):
+        return self.modelSlots[slotIndex]
+
+    def getAllSlotInfo(self):
+        return self.modelSlots
+
+    def get_slot_info(self, slotIndex: int):
+        return self._load_model_slot(slotIndex)
+
+    def save_model_slot(self, slotIndex: int, slotInfo: ModelSlots):
+        self._save_model_slot(slotIndex, slotInfo)
+
+    def update_model_info(self, newData: str):
+        print("[Voice Changer] UPDATE MODEL INFO", newData)
+        newDataDict = json.loads(newData)
+        slotInfo = self._load_model_slot(newDataDict["slot"])
+        setattr(slotInfo, newDataDict["key"], newDataDict["val"])
+        self._save_model_slot(newDataDict["slot"], slotInfo)
+
+    def store_model_assets(self, params: str):
+        print("[Voice Changer] UPLOAD ASSETS", params)
+        paramsDict = json.loads(params)
+        uploadPath = os.path.join(UPLOAD_DIR, paramsDict["file"])
+        storeDir = os.path.join(self.params.model_dir, str(paramsDict["slot"]))
+        storePath = os.path.join(
+            storeDir,
+            paramsDict["file"],
+        )
+        try:
+            shutil.move(uploadPath, storePath)
+            slotInfo = self._load_model_slot(paramsDict["slot"])
+            setattr(slotInfo, paramsDict["name"], storePath)
+            self._save_model_slot(paramsDict["slot"], slotInfo)
+        except Exception as e:
+            print("Exception::::", e)

server/voice_changer/RVC/RVC.py

@@ -5,8 +5,9 @@ from typing import cast
 import numpy as np
 import torch
 import torchaudio
-from data.ModelSlot import RVCModelSlot, loadAllSlotInfo
+from data.ModelSlot import RVCModelSlot
 from utils.downloader.SampleDownloader import getSampleInfos
+from voice_changer.ModelSlotManager import ModelSlotManager
 
 
 # avoiding parse arg error in RVC
@@ -65,21 +66,26 @@ class RVC:
         self.pitchExtractor = PitchExtractorManager.getPitchExtractor(self.settings.f0Detector)
         self.params = params
         EmbedderManager.initialize(params)
-        self.settings.modelSlots = loadAllSlotInfo(self.params.model_dir)
         print("[Voice Changer] RVC initialization: ", params)
+        self.modelSlotManager = ModelSlotManager.get_instance(self.params)
 
         # サンプルカタログ作成
         samples = getSampleInfos(params.sample_mode)
         self.settings.sampleModels = samples
         # 起動時にスロットにモデルがある場合はロードしておく
-        if len(self.settings.modelSlots) > 0:
+
-            for i, slot in enumerate(self.settings.modelSlots):
+        allSlots = self.modelSlotManager.getAllSlotInfo()
-                if len(slot.modelFile) > 0:
+        availableIndex = -1
-                    self.prepareModel(i)
+        for i, slot in enumerate(allSlots):
-                    self.settings.modelSlotIndex = i
+            if slot.modelFile is not None and slot.modelFile != "":
-                    self.switchModel(self.settings.modelSlotIndex)
+                availableIndex = i
-                    self.initialLoad = False
+                break
-                    break
+        if availableIndex >= 0:
+            self.prepareModel(availableIndex)
+            self.settings.modelSlotIndex = availableIndex
+            self.switchModel(self.settings.modelSlotIndex)
+            self.initialLoad = False
+
         self.prevVol = 0.0
 
     def getSampleInfo(self, id: str):
@@ -125,8 +131,8 @@ class RVC:
             slotInfo.indexFile = self.moveToModelDir(slotInfo.indexFile, slotDir)
         if slotInfo.iconFile is not None and len(slotInfo.iconFile) > 0:
             slotInfo.iconFile = self.moveToModelDir(slotInfo.iconFile, slotDir)
-        json.dump(asdict(slotInfo), open(os.path.join(slotDir, "params.json"), "w"))
+        # json.dump(asdict(slotInfo), open(os.path.join(slotDir, "params.json"), "w"))
-        self.settings.modelSlots = loadAllSlotInfo(self.params.model_dir)
+        self.modelSlotManager.save_model_slot(target_slot_idx, slotInfo)
 
         # 初回のみロード(起動時にスロットにモデルがあった場合はinitialLoadはFalseになっている)
         if self.initialLoad:
@@ -147,7 +153,8 @@ class RVC:
                 if val < 0:
                     return True
                 val = val % 1000  # Quick hack for same slot is selected
-                if self.settings.modelSlots[val].modelFile is None or self.settings.modelSlots[val].modelFile == "":
+                allModelSlots = self.modelSlotManager.getAllSlotInfo()
+                if allModelSlots[val].modelFile is None or allModelSlots[val].modelFile == "":
                     print("[Voice Changer] slot does not have model.")
                     return True
                 self.prepareModel(val)
@@ -174,7 +181,8 @@ class RVC:
         if slot < 0:
             print("[Voice Changer] Prepare Model of slot skip:", slot)
             return self.get_info()
-        modelSlot = self.settings.modelSlots[slot]
+        allModelSlots = self.modelSlotManager.getAllSlotInfo()
+        modelSlot = allModelSlots[slot]
 
         print("[Voice Changer] Prepare Model of slot:", slot)
 
@@ -208,7 +216,6 @@ class RVC:
         )
 
     def get_info(self):
-        self.settings.modelSlots = loadAllSlotInfo(self.params.model_dir)
         data = asdict(self.settings)
         if self.pipeline is not None:
             pipelineInfo = self.pipeline.getPipelineInfo()
@@ -293,9 +300,15 @@ class RVC:
         f0_up_key = self.settings.tran
         index_rate = self.settings.indexRatio
         protect = self.settings.protect
-        if_f0 = 1 if self.settings.modelSlots[self.currentSlot].f0 else 0
+
-        embOutputLayer = self.settings.modelSlots[self.currentSlot].embOutputLayer
+        # if_f0 = 1 if self.settings.modelSlots[self.currentSlot].f0 else 0
-        useFinalProj = self.settings.modelSlots[self.currentSlot].useFinalProj
+        # embOutputLayer = self.settings.modelSlots[self.currentSlot].embOutputLayer
+        # useFinalProj = self.settings.modelSlots[self.currentSlot].useFinalProj
+
+        if_f0 = 1 if self.modelSlotManager.get_slot_info(self.currentSlot).f0 else 0
+        embOutputLayer = self.modelSlotManager.get_slot_info(self.currentSlot).embOutputLayer
+        useFinalProj = self.modelSlotManager.get_slot_info(self.currentSlot).useFinalProj
+
         try:
             audio_out = self.pipeline.exec(
                 sid,
@@ -340,7 +353,8 @@ class RVC:
                 pass
 
     def export2onnx(self):
-        modelSlot = self.settings.modelSlots[self.settings.modelSlotIndex]
+        allModelSlots = self.modelSlotManager.getAllSlotInfo()
+        modelSlot = allModelSlots[self.settings.modelSlotIndex]
 
         if modelSlot.isONNX:
             print("[Voice Changer] export2onnx, No pyTorch filepath.")
@@ -359,7 +373,9 @@ class RVC:
         merged = merge_model(req)
         targetSlot = 0
         if req.slot < 0:
-            targetSlot = len(self.settings.modelSlots) - 1
+            # 最後尾のスロット番号を格納先とする。
+            allModelSlots = self.modelSlotManager.getAllSlotInfo()
+            targetSlot = len(allModelSlots) - 1
         else:
             targetSlot = req.slot
 
@@ -386,47 +402,37 @@ class RVC:
         self.currentSlot = self.settings.modelSlotIndex
 
     def update_model_default(self):
-        print("[Voice Changer] UPDATE MODEL DEFAULT!!")
+        # {"slot":9,"key":"name","val":"dogsdododg"}
-        slotDir = os.path.join(self.params.model_dir, str(self.currentSlot))
+        self.modelSlotManager.update_model_info(
-        params = json.load(open(os.path.join(slotDir, "params.json"), "r", encoding="utf-8"))
+            json.dumps(
-        params["defaultTune"] = self.settings.tran
+                {
-        params["defaultIndexRatio"] = self.settings.indexRatio
+                    "slot": self.currentSlot,
-        params["defaultProtect"] = self.settings.protect
+                    "key": "defaultTune",
-
+                    "val": self.settings.tran,
-        json.dump(params, open(os.path.join(slotDir, "params.json"), "w"))
+                }
-        self.settings.modelSlots = loadAllSlotInfo(self.params.model_dir)
+            )
+        )
+        self.modelSlotManager.update_model_info(
+            json.dumps(
+                {
+                    "slot": self.currentSlot,
+                    "key": "defaultIndexRatio",
+                    "val": self.settings.indexRatio,
+                }
+            )
+        )
+        self.modelSlotManager.update_model_info(
+            json.dumps(
+                {
+                    "slot": self.currentSlot,
+                    "key": "defaultProtect",
+                    "val": self.settings.protect,
+                }
+            )
+        )
 
     def update_model_info(self, newData: str):
-        print("[Voice Changer] UPDATE MODEL INFO", newData)
+        self.modelSlotManager.update_model_info(newData)
-        newDataDict = json.loads(newData)
-        try:
-            slotDir = os.path.join(self.params.model_dir, str(newDataDict["slot"]))
-        except Exception as e:
-            print("Exception::::", e)
-        params = json.load(open(os.path.join(slotDir, "params.json"), "r", encoding="utf-8"))
-        params[newDataDict["key"]] = newDataDict["val"]
-        json.dump(params, open(os.path.join(slotDir, "params.json"), "w"))
-        self.settings.modelSlots = loadAllSlotInfo(self.params.model_dir)
 
     def upload_model_assets(self, params: str):
-        print("[Voice Changer] UPLOAD ASSETS", params)
+        self.modelSlotManager.store_model_assets(params)
-        paramsDict = json.loads(params)
-        uploadPath = os.path.join(UPLOAD_DIR, paramsDict["file"])
-        storeDir = os.path.join(self.params.model_dir, str(paramsDict["slot"]))
-        storePath = os.path.join(
-            storeDir,
-            paramsDict["file"],
-        )
-        storeJson = os.path.join(
-            storeDir,
-            "params.json",
-        )
-        try:
-            shutil.move(uploadPath, storePath)
-            params = json.load(open(storeJson, "r", encoding="utf-8"))
-            params[paramsDict["name"]] = storePath  # type:ignore
-            json.dump(params, open(storeJson, "w"))
-        except Exception as e:
-            print("Exception::::", e)
-
-        self.settings.modelSlots = loadAllSlotInfo(self.params.model_dir)

server/voice_changer/RVC/RVCSettings.py

@@ -1,7 +1,8 @@
 from dataclasses import dataclass, field
 from ModelSample import RVCModelSample
-from const import MAX_SLOT_NUM
+
-from data.ModelSlot import ModelSlot, ModelSlots
+# from const import MAX_SLOT_NUM
+# from data.ModelSlot import ModelSlot, ModelSlots
 
 
 @dataclass
@@ -16,7 +17,7 @@ class RVCSettings:
     clusterInferRatio: float = 0.1
 
     framework: str = "PyTorch"  # PyTorch or ONNX
-    modelSlots: list[ModelSlots] = field(default_factory=lambda: [ModelSlot() for _x in range(MAX_SLOT_NUM)])
+    # modelSlots: list[ModelSlots] = field(default_factory=lambda: [ModelSlot() for _x in range(MAX_SLOT_NUM)])
 
     sampleModels: list[RVCModelSample] = field(default_factory=lambda: [])
 

server/voice_changer/VoiceChangerManager.py

@@ -1,4 +1,5 @@
 import numpy as np
+from data.ModelSlot import loadAllSlotInfo
 from utils.downloader.SampleDownloader import downloadSample
 from voice_changer.Local.ServerDevice import ServerDevice, ServerDeviceCallbacks
 from voice_changer.VoiceChanger import VoiceChanger
@@ -95,6 +96,7 @@ class VoiceChangerManager(ServerDeviceCallbacks):
     def get_info(self):
         data = asdict(self.settings)
         data["gpus"] = self.gpus
+        data["modelSlots"] = loadAllSlotInfo(self.params.model_dir)
 
         data["status"] = "OK"
 
@@ -143,6 +145,9 @@ class VoiceChangerManager(ServerDeviceCallbacks):
         self.voiceChanger.merge_models(request)
         return self.get_info()
 
+    def setEmitTo(self, emitTo: Callable[[Any], None]):
+        self.emitToFunc = emitTo
+
     def update_model_default(self):
         self.voiceChanger.update_model_default()
         return self.get_info()
@@ -154,6 +159,3 @@ class VoiceChangerManager(ServerDeviceCallbacks):
     def upload_model_assets(self, params: str):
         self.voiceChanger.upload_model_assets(params)
         return self.get_info()
-
-    def setEmitTo(self, emitTo: Callable[[Any], None]):
-        self.emitToFunc = emitTo