Spaces:
Build error
Build error
| import streamlit as st | |
| from src.services.ai_service import AITutorService | |
| from src.utils.session import get_tutor_context | |
| from datetime import datetime | |
| import speech_recognition as sr | |
| import pyttsx3 | |
| import threading | |
| import queue | |
| import time | |
| class AITutor: | |
| def __init__(self): | |
| self.service = AITutorService() | |
| self.initialize_speech_components() | |
| def initialize_speech_components(self): | |
| """Initialize text-to-speech and speech recognition""" | |
| # Initialize text-to-speech engine | |
| if 'tts_engine' not in st.session_state: | |
| st.session_state.tts_engine = pyttsx3.init() | |
| # Configure voice properties | |
| st.session_state.tts_engine.setProperty('rate', 150) | |
| st.session_state.tts_engine.setProperty('volume', 0.9) | |
| # Get available voices and set a female voice if available | |
| voices = st.session_state.tts_engine.getProperty('voices') | |
| female_voice = next((voice for voice in voices if 'female' in voice.name.lower()), voices[0]) | |
| st.session_state.tts_engine.setProperty('voice', female_voice.id) | |
| # Initialize speech recognition | |
| if 'speech_recognizer' not in st.session_state: | |
| st.session_state.speech_recognizer = sr.Recognizer() | |
| st.session_state.speech_recognizer.energy_threshold = 4000 | |
| st.session_state.audio_queue = queue.Queue() | |
| def speak(self, text: str): | |
| """Make the AI tutor speak the given text""" | |
| def speak_text(): | |
| st.session_state.tts_engine.say(text) | |
| st.session_state.tts_engine.runAndWait() | |
| # Run speech in a separate thread to avoid blocking | |
| thread = threading.Thread(target=speak_text) | |
| thread.start() | |
| def listen(self): | |
| """Listen for user speech input""" | |
| try: | |
| with sr.Microphone() as source: | |
| st.write("🎤 Listening...") | |
| audio = st.session_state.speech_recognizer.listen(source, timeout=5) | |
| text = st.session_state.speech_recognizer.recognize_google(audio) | |
| return text | |
| except sr.WaitTimeoutError: | |
| st.warning("No speech detected. Please try again.") | |
| except sr.UnknownValueError: | |
| st.warning("Could not understand audio. Please try again.") | |
| except sr.RequestError: | |
| st.error("Could not access speech recognition service. Please try typing instead.") | |
| return None | |
| def display_chat_interface(self): | |
| """Display the enhanced chat interface with avatar and speech""" | |
| st.header("AI Tutor") | |
| # Voice interaction controls | |
| col1, col2 = st.columns(2) | |
| with col1: | |
| voice_enabled = st.toggle("Enable Voice", value=False, key="voice_enabled") | |
| with col2: | |
| if voice_enabled: | |
| if st.button("🎤 Start Speaking"): | |
| user_input = self.listen() | |
| if user_input: | |
| self.handle_user_input(user_input) | |
| # Display avatar | |
| self.service.display_avatar(state='neutral') | |
| # Topic selection | |
| topics = [None, 'Physics', 'Mathematics', 'Computer Science', 'Artificial Intelligence'] | |
| selected_topic = st.selectbox( | |
| "Select Topic", | |
| topics, | |
| format_func=lambda x: 'All Topics' if x is None else x, | |
| key="topic_selector" | |
| ) | |
| context = get_tutor_context() | |
| if selected_topic != context['current_topic']: | |
| context['current_topic'] = selected_topic | |
| # Display chat container | |
| chat_container = st.container() | |
| with chat_container: | |
| # Display chat history with avatar states | |
| for message in context['chat_history']: | |
| with st.chat_message(message["role"]): | |
| st.write(message["content"]) | |
| if message["role"] == "assistant": | |
| self.service.display_avatar(state='happy') | |
| if voice_enabled and message.get('speak', True): | |
| self.speak(message["content"]) | |
| message['speak'] = False # Prevent speaking the same message again | |
| # Chat input | |
| if prompt := st.chat_input("Ask your question or click the microphone to speak"): | |
| self.handle_user_input(prompt) | |
| def handle_user_input(self, user_input: str): | |
| """Process user input and generate response""" | |
| # Show thinking avatar | |
| self.service.display_avatar(state='thinking') | |
| # Add user message | |
| context = get_tutor_context() | |
| context['chat_history'].append({ | |
| "role": "user", | |
| "content": user_input | |
| }) | |
| # Generate and display AI response | |
| response = self.service.generate_response(user_input) | |
| # Add AI response | |
| context['chat_history'].append({ | |
| "role": "assistant", | |
| "content": response, | |
| "speak": True # Mark for speaking | |
| }) | |
| # Show happy avatar and rerun | |
| self.service.display_avatar(state='happy') | |
| st.rerun() | |
| def display_learning_metrics(self): | |
| """Display learning progress and engagement metrics""" | |
| with st.sidebar: | |
| st.subheader("Learning Metrics") | |
| context = get_tutor_context() | |
| # Engagement score | |
| metrics = context['engagement_metrics'] | |
| if metrics: | |
| avg_sentiment = sum(m['sentiment_score'] for m in metrics) / len(metrics) | |
| st.metric( | |
| "Engagement Score", | |
| f"{avg_sentiment:.2f}", | |
| delta="0.1" if avg_sentiment > 0.5 else "-0.1" | |
| ) | |
| # Interaction stats | |
| if context['chat_history']: | |
| st.metric( | |
| "Questions Asked", | |
| len([m for m in context['chat_history'] if m['role'] == 'user']) | |
| ) | |
| # Topic focus | |
| if context['current_topic']: | |
| st.info(f"Current focus: {context['current_topic']}") |