clean & organise the source code

core/audio_files.py

@@ -0,0 +1,99 @@
+#coding: utf-8
+from pydub import AudioSegment
+#from openai import OpenAI
+#from io import BytesIO
+#from typing import Any
+#from typing import Dict
+#from typing import IO
+from typing import List
+from typing import Optional
+from typing import Tuple
+from typing import Union
+import base64
+import io
+
+def concatenate_audio_files(audio_list: List[Tuple[Union[bytes, str], float]]) -> Optional[bytes]:
+    """
+    Concatène plusieurs fichiers audio avec des effets sonores.
+
+    Args:
+        audio_list (List[Tuple[Union[bytes, str], float]]): Une liste de tuples, chacun contenant
+            des octets audio (ou une chaîne base64) et la durée.
+
+    Returns:
+        Optional[bytes]: L'audio concaténé sous forme d'octets, ou None en cas d'erreur.
+    """
+    # Créer un segment audio vide
+    final_audio = AudioSegment.empty()
+    
+    try:
+        # Charger les effets sonores
+        begin_sound = AudioSegment.from_mp3(
+            "sound-effects/voice-message-play-begin/voice-message-play-begin-1.mp3"
+        )
+        end_sound = AudioSegment.from_mp3(
+            "sound-effects/voice-message-play-ending/voice-message-play-ending-1.mp3"
+        )
+        
+        # 5 secondes de silence
+        silence = AudioSegment.silent(duration=1500)  # 1500 ms = 1.5 secondes
+        
+        for audio_data, _ in audio_list:
+            # Convertir en bytes si c'est une chaîne base64
+            if isinstance(audio_data, str):
+                audio_bytes = base64.b64decode(audio_data)
+            else:
+                audio_bytes = audio_data
+
+            # Convertir les octets en un segment audio
+            segment = AudioSegment.from_mp3(io.BytesIO(audio_bytes))
+            
+            # Ajouter le son de début, le segment TTS, le son de fin et le silence
+            final_audio += begin_sound + segment + end_sound + silence
+        
+        
+        # Convertir le segment audio final en octets
+        buffer = io.BytesIO()
+        final_audio.export(buffer, format="mp3")
+        return buffer.getvalue()
+    except IOError as e:
+        print(f"Erreur lors de la lecture ou de l'écriture des fichiers audio : {e}")
+        return None
+    except Exception as e:
+        print(f"Une erreur inattendue s'est produite : {e}")
+        return None
+
+
+def split_audio(audio_file, max_size_mb: int = 25) -> List[bytes]:
+    """
+    Divise un fichier audio en segments de taille maximale spécifiée.
+
+    Args:
+        audio_file: Fichier audio ouvert en mode binaire.
+        max_size_mb (int): Taille maximale de chaque segment en Mo.
+
+    Returns:
+        List[bytes]: Liste des segments audio divisés sous forme de bytes.
+    """
+    try:
+        audio_file.seek(0)
+        audio = AudioSegment.from_file(audio_file)
+        duration_ms = len(audio)
+        segment_duration_ms = int(
+            (max_size_mb * 1024 * 1024 * 8) /
+            (audio.frame_rate * audio.sample_width * audio.channels)
+        )
+
+        segments = []
+        for start in range(0, duration_ms, segment_duration_ms):
+            end = min(start + segment_duration_ms, duration_ms)
+            segment = audio[start:end]
+            
+            with io.BytesIO() as buffer:
+                segment.export(buffer, format="mp3")
+                segments.append(buffer.getvalue())
+
+        return segments
+    except Exception as e:
+        print(f"Une erreur s'est produite lors de la division de l'audio : {e}")
+        return []

core/speech_to_text.py

@@ -3,7 +3,16 @@
 import requests  # Pour envoyer des requêtes HTTP à l'API
 import json      # Pour traiter les réponses JSON de l'API
 from os import getenv 
-
+from pydub import AudioSegment
+from openai import OpenAI
+from io import BytesIO
+#from typing import Any
+#from typing import Dict
+from typing import IO
+#from typing import List
+from typing import Optional
+#from typing import Tuple
+from typing import Union
 
 def huggingface_endpoints_stt(fichier_audio: str) -> str:
     # Définir l'URL de l'endpoint d'inférence sur Hugging Face
@@ -41,6 +50,73 @@ def huggingface_endpoints_stt(fichier_audio: str) -> str:
             # En cas d'erreur, afficher le code de statut et le message
             raise Exception(f"Erreur API: {response.status_code}, {response.text}")
 
+
+
+# ############################################################
+
+
+def transcribe_audio(filepath: Union[str, IO], language: Optional[str] = None) -> str:
+    """
+    Transcrit un fichier audio temporaire en texte.
+
+    Args:
+        filepath Chemin vers le fichier audio temporaire à transcrire.
+        language (Optional[str]): La langue de l'audio. Par défaut None.
+
+    Returns:
+        str: Le texte transcrit.
+    """
+    max_size_mb = 25
+    client = OpenAI(api_key=getenv("OPENAI_API_KEY"))
+    try:
+        transcriptions = []
+        with open(filepath if isinstance(filepath, str) else filepath.name, "rb") as f:
+            # filepath peut etre un chemin vers un fichier audio ou un objet IO
+            # verifier si le fichier audio fait plus de 25 Mo
+
+            # Diviser l'audio en segments de taille maximale
+            #segments = split_audio(f, max_size_mb)
+            f.seek(0)
+            audio = AudioSegment.from_file(f)
+            duration_ms = len(audio)
+            segment_duration_ms = int(
+                (max_size_mb * 1024 * 1024 * 8) /
+                (audio.frame_rate * audio.sample_width * audio.channels)
+            )
+            
+            for start in range(0, duration_ms, segment_duration_ms):
+                end = min(start + segment_duration_ms, duration_ms)
+                segment = audio[start:end]
+
+                buffer = BytesIO()
+                segment.export(buffer, format="mp3")
+                buffer.seek(0)
+
+                if not( language ):
+                    response = client.audio.transcriptions.create(
+                        model="whisper-1",
+                        file=("audio.mp3", buffer),
+                        response_format="text"
+                    )
+                else:
+                    response = client.audio.transcriptions.create(
+                        model="whisper-1",
+                        file=("audio.mp3", buffer),
+                        language=language,
+                        response_format="text"
+                    )
+
+                transcriptions.append(response)
+                
+        return " ".join(transcriptions)
+    except Exception as e:
+        print(f"Erreur lors de la transcription de l'audio : {e}")
+        return ""
+
+
+
+
+
 # Exemple d'utilisation de la fonction
 if __name__ == "__main__":
     fichier_audio = "sample_1.wav"  # Remplacez par votre fichier audio
@@ -56,3 +132,7 @@ if __name__ == "__main__":
 """
 Supported content types are:\n                application/json, application/json; charset=UTF-8, text/csv, text/plain, image/png, image/jpeg, image/jpg, image/tiff, image/bmp, image/gif, image/webp, image/x-image, audio/x-flac, audio/flac, audio/mpeg, audio/x-mpeg-3, audio/wave, audio/wav, audio/x-wav, audio/ogg, audio/x-audio, audio/webm, audio/webm;codecs=opus, audio/AMR, audio/amr, audio/AMR-WB, audio/AMR-WB+, audio/m4a, audio/x-m4a\n
 """
+
+
+
+

pages/main.py

@@ -33,8 +33,9 @@ from core.files import read_file
 from core.text_to_speech import openai_tts
 from core.DetectLanguage import detect_language
 from core.speech_to_text import huggingface_endpoints_stt
-
-
+from core.speech_to_text import transcribe_audio
+from core.audio_files import concatenate_audio_files
+from core.audio_files import split_audio
 
 # Au début du fichier, après les imports
 st.set_page_config(
@@ -69,151 +70,6 @@ def process_tts_message(text_response: str) -> Tuple[Optional[bytes], Optional[f
         st.error(f"Une erreur s'est produite lors de la conversion texte-parole : {e}")
         return None, None
 
-# ecrire ici la fonction: split_audio
-def split_audio(audio_file, max_size_mb: int = 25) -> List[bytes]:
-    """
-    Divise un fichier audio en segments de taille maximale spécifiée.
-
-    Args:
-        audio_file: Fichier audio ouvert en mode binaire.
-        max_size_mb (int): Taille maximale de chaque segment en Mo.
-
-    Returns:
-        List[bytes]: Liste des segments audio divisés sous forme de bytes.
-    """
-    try:
-        audio_file.seek(0)
-        audio = AudioSegment.from_file(audio_file)
-        duration_ms = len(audio)
-        segment_duration_ms = int(
-            (max_size_mb * 1024 * 1024 * 8) /
-            (audio.frame_rate * audio.sample_width * audio.channels)
-        )
-
-        segments = []
-        for start in range(0, duration_ms, segment_duration_ms):
-            end = min(start + segment_duration_ms, duration_ms)
-            segment = audio[start:end]
-            
-            with io.BytesIO() as buffer:
-                segment.export(buffer, format="mp3")
-                segments.append(buffer.getvalue())
-
-        return segments
-    except Exception as e:
-        print(f"Une erreur s'est produite lors de la division de l'audio : {e}")
-        return []
-
-
-def transcribe_audio(filepath: Union[str, IO], language: Optional[str] = None) -> str:
-    """
-    Transcrit un fichier audio temporaire en texte.
-
-    Args:
-        filepath Chemin vers le fichier audio temporaire à transcrire.
-        language (Optional[str]): La langue de l'audio. Par défaut None.
-
-    Returns:
-        str: Le texte transcrit.
-    """
-    max_size_mb = 25
-    
-    try:
-        transcriptions = []
-        with open(filepath if isinstance(filepath, str) else filepath.name, "rb") as f:
-            # filepath peut etre un chemin vers un fichier audio ou un objet IO
-            # verifier si le fichier audio fait plus de 25 Mo
-
-            # Diviser l'audio en segments de taille maximale
-            #segments = split_audio(f, max_size_mb)
-            f.seek(0)
-            audio = AudioSegment.from_file(f)
-            duration_ms = len(audio)
-            segment_duration_ms = int(
-                (max_size_mb * 1024 * 1024 * 8) /
-                (audio.frame_rate * audio.sample_width * audio.channels)
-            )
-            
-            for start in range(0, duration_ms, segment_duration_ms):
-                end = min(start + segment_duration_ms, duration_ms)
-                segment = audio[start:end]
-
-                buffer = BytesIO()
-                segment.export(buffer, format="mp3")
-                buffer.seek(0)
-
-                if not( language ):
-                    response = client.audio.transcriptions.create(
-                        model="whisper-1",
-                        file=("audio.mp3", buffer),
-                        response_format="text"
-                    )
-                else:
-                    response = client.audio.transcriptions.create(
-                        model="whisper-1",
-                        file=("audio.mp3", buffer),
-                        language=language,
-                        response_format="text"
-                    )
-
-                transcriptions.append(response)
-                
-        return " ".join(transcriptions)
-    except Exception as e:
-        print(f"Erreur lors de la transcription de l'audio : {e}")
-        return ""
-
-
-def concatenate_audio_files(audio_list: List[Tuple[Union[bytes, str], float]]) -> Optional[bytes]:
-    """
-    Concatène plusieurs fichiers audio avec des effets sonores.
-
-    Args:
-        audio_list (List[Tuple[Union[bytes, str], float]]): Une liste de tuples, chacun contenant
-            des octets audio (ou une chaîne base64) et la durée.
-
-    Returns:
-        Optional[bytes]: L'audio concaténé sous forme d'octets, ou None en cas d'erreur.
-    """
-    # Créer un segment audio vide
-    final_audio = AudioSegment.empty()
-    
-    try:
-        # Charger les effets sonores
-        begin_sound = AudioSegment.from_mp3(
-            "sound-effects/voice-message-play-begin/voice-message-play-begin-1.mp3"
-        )
-        end_sound = AudioSegment.from_mp3(
-            "sound-effects/voice-message-play-ending/voice-message-play-ending-1.mp3"
-        )
-        
-        # 5 secondes de silence
-        silence = AudioSegment.silent(duration=1500)  # 1500 ms = 1.5 secondes
-        
-        for audio_data, _ in audio_list:
-            # Convertir en bytes si c'est une chaîne base64
-            if isinstance(audio_data, str):
-                audio_bytes = base64.b64decode(audio_data)
-            else:
-                audio_bytes = audio_data
-
-            # Convertir les octets en un segment audio
-            segment = AudioSegment.from_mp3(io.BytesIO(audio_bytes))
-            
-            # Ajouter le son de début, le segment TTS, le son de fin et le silence
-            final_audio += begin_sound + segment + end_sound + silence
-        
-        
-        # Convertir le segment audio final en octets
-        buffer = io.BytesIO()
-        final_audio.export(buffer, format="mp3")
-        return buffer.getvalue()
-    except IOError as e:
-        print(f"Erreur lors de la lecture ou de l'écriture des fichiers audio : {e}")
-        return None
-    except Exception as e:
-        print(f"Une erreur inattendue s'est produite : {e}")
-        return None
 
 def process_message(
     message: str,
@@ -252,7 +108,6 @@ def process_message(
         st.error(f"Une erreur s'est produite lors de la génération de la réponse : {e}")
         return ""    
 
-
 class GlobalSystemPrompts:
     """Class to store global system prompts."""
 
@@ -318,7 +173,6 @@ LANGUAGES_EMOJI = {
     "Vietnamese": "🇻🇳", "Welsh": "🏴󠁧󠁢󠁷󠁬󠁳󠁿"
 }
 
-
 def convert_iso6391_to_language_name(language_code: str, 
                                      filter_mode=True) -> str:
     """
@@ -430,9 +284,6 @@ def init_process_mode(
         return system_prompt, operation_prompt
     return "", ""
 
-
-
-
 @st.dialog("Settings")
 def tts_settings(name__tts_voice, 
          state__tts_with_text, 
@@ -467,7 +318,6 @@ def tts_settings(name__tts_voice,
         #st.session_state.
         st.rerun()
 
-
 @st.fragment
 def recorder_released():
     if "rec_widget" in st.session_state:
@@ -681,13 +531,11 @@ def main_page():
                         icon="ℹ️"):
             st.subheader(f"version: {__version__}")
             st.info(get_translation("info_app"))
-        
 
         with st.expander(f"{get_translation('selection_langue')}", 
                         expanded=True, 
                         icon="🌐"):
             # Conteneur pour la sélection de langue
-
             # Sélection multiple des langues de destination
             st.multiselect(
                 label=get_translation("langues_destination"),
@@ -707,14 +555,6 @@ def main_page():
         )
         
         st.experimental_audio_input("Record a voice message",on_change=recorder_released, key="rec_widget")
-        #audiorecorder(
-        #    start_prompt=get_translation("cliquez_enregistrer"),
-        #    stop_prompt=get_translation("cliquez_arreter"),
-        #    pause_prompt=get_translation("cliquez_pause"),
-        #    show_visualizer=True,
-        #    key="vocal_chat_input"
-        #)
-
 
         if st.session_state.user_input:
             
@@ -757,7 +597,6 @@ def main_page():
                     st.error("Erreur : Les prompts système ou d'opération sont vides.")
                     raise ValueError("Les prompts système ou d'opération ne peuvent pas être vides.")
                 
-                
                 with st.status(f'({target_language_name}) - {get_translation("traduction_en_cours")}', expanded=True) as response_status:
                     with st.chat_message("assistant", avatar="👻"):
                         message_placeholder = st.empty()
@@ -767,7 +606,6 @@ def main_page():
                             st.session_state.system_prompt
                         )
 
-
                         response_status.update(label=f'({target_language_name}) - {get_translation("traduction_en_cours")}', state="running", expanded=True)
                         for response_chunk in st.session_state.response_generator:
                             message_placeholder.markdown(response_chunk)
@@ -787,13 +625,11 @@ def main_page():
                                 else:
                                     response_status.update(label=f'({target_language_name}) - {get_translation("erreur_synthese_vocale")}', state="error", expanded=False)
 
-
                             else:
                                 response_status.update(label=f'({target_language_name}) - {get_translation("traduction_terminee")}', state="complete", expanded=False)
                         else:
                             response_status.update(label=f'({target_language_name}) - {get_translation("erreur_traduction")}', state="error", expanded=False)
 
-
             if st.session_state.audio_list:
                 with st.status(f"{get_translation('concatenation_audio_en_cours')}", expanded=False) as audio_status:
                     audio_status.update(label=f"{get_translation('concatenation_audio_en_cours')}", state="running", expanded=False)
@@ -805,7 +641,6 @@ def main_page():
                             st.session_state.timestamp = time.strftime("%Y%m%d-%H%M%S")
                             st.session_state.langues = "_".join([lang["iso-639-1"] for lang in st.session_state.selected_languages])
                             st.session_state.nom_fichier = f"reponse_audio_{st.session_state.langues}_{st.session_state.timestamp}.mp3"
-                            
 
                             st.audio(st.session_state.final_audio, format="audio/mp3", autoplay=st.session_state.autoplay_tts)
 
@@ -819,23 +654,12 @@ def main_page():
                                 key=f"download_button_{st.session_state.langues}_{st.session_state.timestamp}",
                             )
 
-                            # ##
                             audio_status.update(label=f"{get_translation('concatenation_audio_terminee')}", state="complete", expanded=True)
                     except Exception as e:
                         st.error(f"{get_translation('erreur_concatenation_audio')} : {str(e)}")
                         
-                        # ##
                         audio_status.update(label=f"{get_translation('erreur_concatenation_audio')} : {str(e)}", state="error", expanded=True)
                     
-            #clear_inputs_garbages()
-            # Interface utilisateur pour l'enregistrement audio
-            # st.write(f"🗣️ {get_translation('enregistrez_message')}")
-
-
-
-        
-                
-
 
 
 def clear_inputs_garbages(sessions_state_list: Optional[list] =