app.py

import streamlit as st
from huggingface_hub import InferenceClient
import re
import edge_tts
import asyncio
from concurrent.futures import ThreadPoolExecutor
import tempfile
from pydub import AudioSegment

# Initialize Hugging Face InferenceClient
client_hf = InferenceClient("mistralai/Mixtral-8x7B-Instruct-v0.1")

# Define the async function for text-to-speech conversion using Edge TTS
async def text_to_speech_edge(text, language_code):
    voice = {"fr": "fr-FR-RemyMultilingualNeural"}[language_code]
    communicate = edge_tts.Communicate(text, voice)
    with tempfile.NamedTemporaryFile(delete=False, suffix=".mp3") as tmp_file:
        tmp_path = tmp_file.name
    await communicate.save(tmp_path)
    return tmp_path

# Helper function to run async functions from within Streamlit (synchronous context)
def run_in_threadpool(func, *args, **kwargs):
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    future = asyncio.ensure_future(func(*args, **kwargs))
    return loop.run_until_complete(future)

def concatenate_audio(paths):
    combined = AudioSegment.empty()
    for path in paths:
        audio = AudioSegment.from_mp3(path)
        combined += audio
    combined_path = tempfile.mktemp(suffix=".mp3")
    combined.export(combined_path, format="mp3")
    return combined_path

# Modified function to work with async Edge TTS
def dictee_to_audio_segmented(dictee):
    sentences = segmenter_texte(dictee)
    audio_urls = []
    with ThreadPoolExecutor() as executor:
        for sentence in sentences:
            processed_sentence = replace_punctuation(sentence)
            audio_path = executor.submit(run_in_threadpool, text_to_speech_edge, processed_sentence, "fr").result()
            audio_urls.append(audio_path)
    return audio_urls

def generer_dictee(classe, longueur):
    prompt = f"Créer une dictée pour la classe {classe} d'une longueur d'environ {longueur} mots. Il est important de créer le texte uniquement de la dictée et de ne pas ajouter de consignes ou d'indications supplémentaires."
    generate_kwargs = {
        "temperature": 0.7,
        "max_new_tokens": 1000,
        "top_p": 0.95,
        "repetition_penalty": 1.2,
        "do_sample": True,
    }
    formatted_prompt = f"<s>[INST] {prompt} [/INST]"
    stream = client_hf.text_generation(formatted_prompt, **generate_kwargs, stream=True, details=True, return_full_text=False)
    dictee = ""
    for response in stream:
        dictee += response.token.text
    dictee = dictee.replace("</s>", "").strip()
    return dictee

def correction_dictee(dictee_utilisateur):
    prompt = f"Voici une dictée crée: {st.session_state.dictée} | Voici la dictée faite par l'utilisateur : {dictee_utilisateur} - Corrige la dictée en donnant les explications, utilise les syntax du markdown pour une meilleur comprehesion de la correction. Il est important de comparer la dictée de l'utilisateur avec uniquement celle crée."
    generate_kwargs = {
        "temperature": 0.7,
        "max_new_tokens": 2000,  # Ajustez selon la longueur attendue de la correction
        "top_p": 0.95,
        "repetition_penalty": 1.2,
        "do_sample": True,
    }
    formatted_prompt = f"<s>[INST] {prompt} [/INST]"
    stream = client_hf.text_generation(formatted_prompt, **generate_kwargs, stream=True, details=True, return_full_text=False)
    correction = ""
    for response in stream:
        correction += response.token.text
    correction = correction.replace("</s>", "").strip()
    return correction

def replace_punctuation(text):
    replacements = {
        ".": " point.",
        ",": " virgule,",
        ";": " point-virgule;",
        ":": " deux-points:",
        "!": " point d'exclamation!",
        "?": " point d'interrogation?",
    }
    for key, value in replacements.items():
        text = text.replace(key, value)
    return text

def segmenter_texte(texte):
    sentences = re.split(r'(?<=[.!?]) +', texte)
    return sentences

# Stocker la dictée dans une variable de session pour qu'elle persiste
if 'dictée' not in st.session_state:
    st.session_state.dictée = None

# Streamlit App Interface
st.set_page_config(layout="wide")
st.title('Générateur de Dictée')

# Ajouter un container pour le paramètre de mode
with st.container():
    mode = st.radio("Mode:", ["S'entrainer", "Entrainer"], key="mode")

# Ajouter un container pour les paramètres de la dictée
params_placeholder = st.container()

if mode == "S'entrainer":
    st.subheader("Paramètres de la dictée pour s'entrainer")
else:
    st.subheader("Paramètres de la dictée pour entrainer quelqu'un d'autre")

# Utiliser des colonnes pour organiser les éléments
col1, col2 = st.columns(2)

with params_placeholder:
    with col1:
        classe = st.selectbox("Classe", ["CP", "CE1", "CE2", "CM1", "CM2", "6ème", "5ème", "4ème", "3ème", "Seconde", "Premiere", "Terminale"], index=2)
    with col2:
        longueur = st.slider("Longueur de la dictée (nombre de mots)", 50, 500, 200)

    bouton_generer = st.button('Générer la Dictée')

if bouton_generer:
    # Masquer les paramètres
    params_placeholder.empty()

    with st.spinner("Génération de la dictée en cours..."):
        dictee = generer_dictee(classe, longueur)
        st.session_state.dictée = dictee

if st.session_state.dictée:
    # Afficher la dictée ou les audios en fonction du mode
    if mode == "S'entrainer":
        audio_urls = dictee_to_audio_segmented(st.session_state.dictée)
        concatenated_audio_path = concatenate_audio(audio_urls)

        col1, col2 = st.columns(2)

        with col1:
            with st.expander("Écouter la dictée complète"):
                st.audio(concatenated_audio_path, format='audio/wav', start_time=0)
            with st.expander("Phrases de la Dictée"):
                for idx, url in enumerate(audio_urls, start=1):
                    st.markdown(f"**Phrase {idx}:**")
                    st.audio(url, format='audio/wav')

        with col2:
            dictee_utilisateur = st.text_input("Écrivez la dictée ici:")
            if st.button('Correction'):
                with st.spinner("Génération de la correction en cours..."):
                    correction = correction_dictee(dictee_utilisateur)
                    st.markdown("Voici la correction :")
                    st.markdown(correction)

    elif mode == "Entrainer":
        st.text_area("Voici votre dictée :", st.session_state.dictée, height=300)

    # Ajouter un bouton "Retour" pour réinitialiser la session
    col1, col2 = st.columns(2)
    with col2:
        if st.button('Retour'):
            st.session_state.dictée = None
            # Réafficher les paramètres
            with params_placeholder:
                with col1:
                    classe = st.selectbox("Classe", ["CP", "CE1", "CE2", "CM1", "CM2", "6ème", "5ème", "4ème", "3ème", "Seconde", "Premiere", "Terminale"], index=2)
                with col2:
                    longueur = st.slider("Longueur de la dictée (nombre de mots)", 50, 500, 200)
                bouton_generer = st.button('Générer la Dictée')