Merge pull request #24 from lendot/sample-wav-multiple

Support for speakers with multiple speaker_wav files

Author: daswer123

xtts_api_server/RealtimeTTS/engines/coqui_engine.py

@@ -401,7 +401,7 @@ def send_command(self, command, data):
         message = {'command': command, 'data': data}
         self.parent_synthesize_pipe.send(message)            
 
-    def set_cloning_reference(self, cloning_reference_wav: str):
+    def set_cloning_reference(self, cloning_reference_wav: Union[str, List[str]]):
         """
         Send an 'update_reference' command and wait for a response.
         """
@@ -594,7 +594,7 @@ def get_voices(self):
 
         return voice_file_names 
 
-    def set_voice(self, voice: str):
+    def set_voice(self, voice: Union[str, List[str]]):
         """
         Sets the voice to be used for speech synthesis.
         """
@@ -637,4 +637,4 @@ def shutdown(self):
 
         # Wait for the process to terminate
         self.synthesize_process.join()
-        logging.info('Worker process has been terminated')
+        logging.info('Worker process has been terminated')

xtts_api_server/server.py

@@ -147,7 +147,7 @@ def get_folders():
     output_folder = XTTS.output_folder
     return {"speaker_folder": speaker_folder, "output_folder": output_folder}
 
-@app.get("/sample/{file_name}")
+@app.get("/sample/{file_name:path}")
 def get_sample(file_name: str):
     file_path = os.path.join(XTTS.speaker_folder, file_name)
     if os.path.isfile(file_path):
@@ -184,7 +184,7 @@ async def tts_to_audio(request: SynthesisRequest):
                 raise HTTPException(status_code=400,
                                     detail="Language code sent is either unsupported or misspelled.")
 
-            speaker_wav = XTTS.get_speaker_path(request.speaker_wav)
+            speaker_wav = XTTS.get_speaker_wav(request.speaker_wav)
             language = request.language[0:2]
 
             if stream.is_playing() and not STREAM_PLAY_SYNC:
@@ -262,4 +262,4 @@ async def tts_to_file(request: SynthesisFileRequest):
         raise HTTPException(status_code=500, detail=f"An error occurred: {str(e)}")
 
 if __name__ == "__main__":
-    uvicorn.run(app,host="0.0.0.0",port=8002)
+    uvicorn.run(app,host="0.0.0.0",port=8002)

xtts_api_server/tts_funcs.py

@@ -111,19 +111,18 @@ def switch_model_device(self):
                 # Clearing the cache to free up VRAM
                 torch.cuda.empty_cache()
 
-    def get_or_create_latents(self, speaker_wav):
-        if speaker_wav not in self.latents_cache:
+    def get_or_create_latents(self, speaker_name, speaker_wav):
+        if speaker_name not in self.latents_cache:
+            logger.info(f"creating latents for {speaker_name}: {speaker_wav}")
             gpt_cond_latent, speaker_embedding = self.model.get_conditioning_latents(speaker_wav)
-            self.latents_cache[speaker_wav] = (gpt_cond_latent, speaker_embedding)
-        return self.latents_cache[speaker_wav]
+            self.latents_cache[speaker_name] = (gpt_cond_latent, speaker_embedding)
+        return self.latents_cache[speaker_name]
 
     def create_latents_for_all(self):
-        speakers_list = self.get_speakers()
+        speakers_list = self._get_speakers()
 
-        for speaker_name in speakers_list:
-            speaker_wav = os.path.join(self.speaker_folder, speaker_name+".wav")
-
-            self.get_or_create_latents(speaker_wav)
+        for speaker in speakers_list:
+            self.get_or_create_latents(speaker['speaker_name'],speaker['speaker_wav'])
 
         logger.info(f"Latents created for all {len(speakers_list)} speakers.")
 
@@ -137,7 +136,7 @@ def create_directories(self):
             if not os.path.exists(absolute_path):
                 # If the folder does not exist, create it
                 os.makedirs(absolute_path)
-                print(f"Folder in the path {absolute_path} has been created")
+                logger.info(f"Folder in the path {absolute_path} has been created")
 
     def set_speaker_folder(self, folder):
         if os.path.exists(folder) and os.path.isdir(folder):
@@ -155,38 +154,78 @@ def set_out_folder(self, folder):
         else:
             raise ValueError("Provided path is not a valid directory")
 
-    def list_speakers(self):
-        speakers_list = [f for f in os.listdir(self.speaker_folder) if f.endswith('.wav')]
-        return speakers_list
+    def get_wav_files(self, directory):
+        """ Finds all the wav files in a directory. """
+        wav_files = [f for f in os.listdir(directory) if f.endswith('.wav')]
+        return wav_files
+
+    def _get_speakers(self):
+        """
+        Gets info on all the speakers.
+
+        Returns a list of {speaker_name,speaker_wav,preview} dicts
+        """
+        speakers = []
+        for f in os.listdir(self.speaker_folder):
+            full_path = os.path.join(self.speaker_folder,f)
+            if os.path.isdir(full_path):
+                # multi-sample voice
+                subdir_files = self.get_wav_files(full_path) 
+                if len(subdir_files) == 0:
+                    # no wav files in directory
+                    continue
+
+                speaker_name = f
+                speaker_wav = [os.path.join(self.speaker_folder,f,s) for s in subdir_files]
+                # use the first file found as the preview
+                preview = os.path.join(f,subdir_files[0])
+                speakers.append({
+                        'speaker_name': speaker_name,
+                        'speaker_wav': speaker_wav,
+                        'preview': preview
+                        })
+
+            elif f.endswith('.wav'):
+                speaker_name = os.path.splitext(f)[0]
+                speaker_wav = full_path 
+                preview = f
+                speakers.append({
+                        'speaker_name': speaker_name,
+                        'speaker_wav': speaker_wav,
+                        'preview': preview
+                        })
+        return speakers
 
     def get_speakers(self):
-        # Use os.path.splitext to split off the extension and take only the name
-        speakers_list = [os.path.splitext(f)[0] for f in os.listdir(self.speaker_folder) if f.endswith('.wav')]
-        return speakers_list
+        """ Gets available speakers """
+        speakers = [ s['speaker_name'] for s in self._get_speakers() ] 
+        return speakers
+
     # Special format for SillyTavern
     def get_speakers_special(self):
-        speakers_list = []
         BASE_URL = os.getenv('BASE_URL', '127.0.0.1:8020')
         TUNNEL_URL = os.getenv('TUNNEL_URL', '')
 
-        preview_url = "" 
-        for file in os.listdir(self.speaker_folder):
-            
+        speakers_special = []
+
+        speakers = self._get_speakers()
+
+        for speaker in speakers:
             if TUNNEL_URL == "":
-                preview_url = f"{BASE_URL}/sample/{file}"
+                preview_url = f"{BASE_URL}/sample/{speaker['preview']}"
             else:
-                preview_url = f"{TUNNEL_URL}/sample/{file}"
+                preview_url = f"{TUNNEL_URL}/sample/{speaker['preview']}"
 
-            if file.endswith('.wav'):
-                speaker_name = os.path.splitext(file)[0]
-                speaker = {
-                    'name': speaker_name,
-                    'voice_id': speaker_name,
+            speaker_special = {
+                    'name': speaker['speaker_name'],
+                    'voice_id': speaker['speaker_name'],
                     'preview_url': preview_url
-                }
-                speakers_list.append(speaker)
-        return speakers_list
-    
+            }
+            speakers_special.append(speaker_special)
+
+        return speakers_special
+
+
     def list_languages(self):
         return reversed_supported_languages
 
@@ -197,11 +236,11 @@ def clean_text(self,text):
         text = re.sub(r'"\s?(.*?)\s?"', r"'\1'", text)
         return text
 
-    def local_generation(self,text,speaker_wav,language,output_file):
+    def local_generation(self,text,speaker_name,speaker_wav,language,output_file):
         # Log time
         generate_start_time = time.time()  # Record the start time of loading the model
 
-        gpt_cond_latent, speaker_embedding = self.get_or_create_latents(speaker_wav)
+        gpt_cond_latent, speaker_embedding = self.get_or_create_latents(speaker_name, speaker_wav)
 
         out = self.model.inference(
             text,
@@ -230,29 +269,37 @@ def api_generation(self,text,speaker_wav,language,output_file):
                 language=language,
                 file_path=output_file,
         )
-    
-    def get_speaker_path(self,speaker_name_or_path):
-        # Check if the speaker path is a .wav file or just the name
+
+    def get_speaker_wav(self, speaker_name_or_path):
+        """ Gets the speaker_wav(s) for a given speaker name. """
         if speaker_name_or_path.endswith('.wav'):
-                if os.path.isabs(speaker_name_or_path):
-                    # If it's an absolute path for the speaker file
-                    speaker_wav = speaker_name_or_path
-                else:
-                    # It's just a filename; append it to the speakers folder
-                    speaker_wav = os.path.join(self.speaker_folder, speaker_name_or_path)
+            # it's a file name
+            if os.path.isabs(spekaer_name_or_path):
+                # absolute path; nothing to do
+                speaker_wav = speaker_name_or_path
+            else:
+                # make it a full path
+                speaker_wav = os.path.join(self.speaker_folder, speaker_name_or_path)
         else:
-                # Look for the corresponding .wav in our list of speakers
-                speakers_list = self.list_speakers()
-                if f"{speaker_name_or_path}.wav" in speakers_list:
-                    speaker_wav = os.path.join(self.speaker_folder, f"{speaker_name_or_path}.wav")
-                else:
-                    raise ValueError(f"Speaker {speaker_name_or_path} not found.")
+            # it's a speaker name
+            full_path = os.path.join(self.speaker_folder, speaker_name_or_path) 
+            wav_file = f"{full_path}.wav"
+            if os.path.isdir(full_path):
+                # multi-sample speaker
+                speaker_wav = [ os.path.join(full_path,wav) for wav in self.get_wav_files(full_path) ]
+                if len(speaker_wav) == 0:
+                    raise ValueError(f"no wav files found in {full_path}")
+            elif os.path.isfile(wav_file):
+                speaker_wav = wav_file
+            else:
+                raise ValueError(f"Speaker {speaker_name_or_path} not found.")
+
         return speaker_wav
 
 
     def process_tts_to_file(self, text, speaker_name_or_path, language, file_name_or_path="out.wav"):
         try:
-            speaker_wav = self.get_speaker_path(speaker_name_or_path)
+            speaker_wav = self.get_speaker_wav(speaker_name_or_path)
             # Determine output path based on whether a full path or a file name was provided
             if os.path.isabs(file_name_or_path):
                 # An absolute path was provided by user; use as is.
@@ -268,7 +315,7 @@ def process_tts_to_file(self, text, speaker_name_or_path, language, file_name_or
 
             # Define generation if model via api or locally
             if self.model_source == "local":
-                self.local_generation(clear_text,speaker_wav,language,output_file)
+                self.local_generation(clear_text,speaker_name_or_path,speaker_wav,language,output_file)
             else:
                 self.api_generation(clear_text,speaker_wav,language,output_file)
 
@@ -282,4 +329,4 @@ def process_tts_to_file(self, text, speaker_name_or_path, language, file_name_or
 
 
 
-        
+