Update app.py

app.py

@@ -1,16 +1,13 @@
 #!/usr/bin/env python3
 import streamlit as st
 from gradio_client import Client
+from groq import Groq
 from PIL import Image
 import moviepy.editor as mp
 from natsort import natsorted
-from pydantic import BaseModel, Field
-from typing import List, Dict, Type, Optional, TypedDict
-from langgraph.graph import StateGraph, START, END
-from langchain_groq import ChatGroq
-from langchain_core.messages import SystemMessage
 import os
 from dotenv import load_dotenv
+import json
 
 # Load environment variables
 load_dotenv()
@@ -18,227 +15,145 @@ load_dotenv()
 # Constants
 HF_TOKEN = os.getenv("HF_TOKEN")
 GROQ_API_KEY = os.getenv("GROQ_API_KEY")
-IMAGE_GENERATION_SPACE_NAME = "stabilityai/stable-diffusion-3.5-large-turbo"  # Updated to correct Space
-SUPPORTED_FORMATS = ["mp3", "wav", "ogg", "flac", "aac", "m4a"]
+IMAGE_GENERATION_SPACE_NAME = "stabilityai/stable-diffusion-3.5-large-turbo"
 
-# Pydantic Models
-class SingleScene(BaseModel):
-    text: str = Field(description="Actual Segment of text(a scene) from the complete story")
-    image_prompts: List[str] = Field(
-        description="""List of detailed and descriptive image prompts for the segment
-        prompt format: [theme: {atmosphere/mood}] [style: {artistic/photorealistic}] [focus: {main subject}] [details: {specific elements}] [lighting: {day/night/mystic}] [perspective: {close-up/wide-angle}]"
-        Example: "theme: eerie forest | style: cinematic realism | focus: abandoned cabin | details: broken windows, overgrown vines | lighting: moonlit fog | perspective: wide-angle shot"
-        """
-    )
+# Initialize Groq client
+groq_client = Groq(api_key=GROQ_API_KEY)
 
-class ScenesResponseSchema(BaseModel):
-    scenes: List[SingleScene]
-
-# Structured Output Extractor
-class State(TypedDict):
-    messages: list
-    output: Optional[BaseModel]
-
-class StructuredOutputExtractor:
-    def __init__(self, response_schema: Type[BaseModel]):
-        self.response_schema = response_schema
-        self.llm = ChatGroq(model="deepseek-r1-distill-llama-70b", api_key=GROQ_API_KEY)
-        self.structured_llm = self.llm.with_structured_output(response_schema)
-        self._build_graph()
-
-    def _build_graph(self):
-        graph_builder = StateGraph(State)
-        graph_builder.add_node("extract", self._extract_structured_info)
-        graph_builder.add_edge(START, "extract")
-        graph_builder.add_edge("extract", END)
-        self.graph = graph_builder.compile()
-
-    def _extract_structured_info(self, state: dict):
-        query = state['messages'][-1].content
-        try:
-            output = self.structured_llm.invoke(query)
-            return {"output": output}
-        except Exception as e:
-            st.error(f"Error during extraction: {e}")
-            return {"output": None}
-
-    def extract(self, query: str) -> Optional[BaseModel]:
-        result = self.graph.invoke({"messages": [SystemMessage(content=query)]})
-        return result.get('output')
+# LLM Models (free options)
+LLM_MODELS = {
+    "Mixtral 8x7B (Groq)": "mixtral-8x7b-32768",
+    "Mistral 7B (HF)": "mistralai/Mixtral-7B-Instruct-v0.1",
+    "LLaMA 13B (HF)": "meta-llama/Llama-13b-hf"  # Note: May require approval; replace if needed
+}
 
 # Utility Functions
-def calculate_read_time(text: str, words_per_minute: int = 155) -> str:
-    try:
-        if not text or not isinstance(text, str):
-            return "Invalid input: Text must be a non-empty string."
-        words = text.split()
-        word_count = len(words)
-        total_seconds = (word_count / words_per_minute) * 60
-        hours = int(total_seconds // 3600)
-        minutes = int((total_seconds % 3600) // 60)
-        seconds = int(total_seconds % 60)
-        if hours > 0:
-            return f"Reading time: {hours} hour(s), {minutes} minute(s), and {seconds} second(s)."
-        elif minutes > 0:
-            return f"Reading time: {minutes} minute(s) and {seconds} second(s)."
-        else:
-            return f"Reading time: {seconds} second(s)."
-    except Exception as e:
-        return f"An error occurred: {e}"
-
-def get_scenes(text_script: str):
-    read_time = calculate_read_time(text_script)
+def generate_tutor_output(subject, difficulty, student_input, model):
     prompt = f"""
-    ROLE: Story to Scene Generator
-    Tasks: For the given story
-    1. Read it Completely and Understand the Complete Context
-    2. Rewrite the story in tiny scenes(but without even changing a word) with highly detailed and context aware list of image prompts to visualize each scene
-    3. Never Describe complete scene in a single image prompt use multiple prompts
-    RULE OF THUMB: 12 image prompts / 1 min audio
-
-    Estimated Read Time: {read_time}\n\n
-    Complete Story: {text_script}
+    You are an expert tutor in {subject} at the {difficulty} level. 
+    The student has provided the following input: "{student_input}"
+    
+    Please generate:
+    1. A brief, engaging lesson on the topic (2-3 paragraphs)
+    2. A thought-provoking question to check understanding
+    3. Constructive feedback on the student's input
+    
+    Format your response as a JSON object with keys: "lesson", "question", "feedback"
     """
-    extractor = StructuredOutputExtractor(response_schema=ScenesResponseSchema)
-    result = extractor.extract(prompt)
-    return result.model_dump() if result else {}
-
-def generate_audio(text, language_code, speaker, path='test_audio.mp3'):
-    try:
-        client = Client("habib926653/Multilingual-TTS")
-        result = client.predict(
-            text=text,
-            language_code=language_code,
-            speaker=speaker,
-            api_name="/text_to_speech_edge"
+    
+    if model.startswith("mixtral"):  # Groq model
+        completion = groq_client.chat.completions.create(
+            messages=[{
+                "role": "system",
+                "content": f"You are the world's best AI tutor for {subject}, renowned for clear, engaging explanations."
+            }, {
+                "role": "user",
+                "content": prompt
+            }],
+            model=model,
+            max_tokens=1000
         )
-        audio_file_path = result[1]
-        with open(audio_file_path, 'rb') as f:
-            audio_bytes = f.read()
-        with open(path, 'wb') as f:
-            f.write(audio_bytes)
-        return {"audio_file": path}
-    except Exception as e:
-        st.error(f"Error during audio generation: {e}")
-        return {"error": str(e)}
-
-def generate_image(prompt, path='test_image.png'):
+        return json.loads(completion.choices[0].message.content)
+    else:  # Hugging Face models
+        try:
+            client = Client("https://api-inference.huggingface.co/models/" + model, hf_token=HF_TOKEN)
+            response = client.predict(prompt, api_name="/generate")
+            return json.loads(response)
+        except:
+            st.warning(f"HF model {model} failed, falling back to Mixtral.")
+            return generate_tutor_output(subject, difficulty, student_input, "mixtral-8x7b-32768")
+
+def generate_image(prompt, path='temp_image.png'):
     try:
-        client = Client("stabilityai/stable-diffusion-3.5-large-turbo", hf_token=HF_TOKEN)
+        client = Client(IMAGE_GENERATION_SPACE_NAME, hf_token=HF_TOKEN)
         result = client.predict(
-            prompt=prompt,  # Text prompt for image generation
-            width=1280,
-            height=720,
-            api_name="/generate_image"
+            prompt=prompt,
+            width=512,  # Reduced for speed
+            height=512,
+            api_name="/predict"  # Correct endpoint
         )
         image = Image.open(result)
         image.save(path)
-        return result
+        return path
     except Exception as e:
-        st.error(f"Error during image generation: {e}")
-        return {"error": str(e)}
-
-def generate_video_assets(scenes: Dict, language: str, speaker: str, base_path: str = "media") -> str:
-    try:
-        if not os.path.exists(base_path):
-            os.makedirs(base_path)
-        scenes_list = scenes.get("scenes", [])
-        video_folder = os.path.join(base_path, f"video_{len(os.listdir(base_path)) + 1}")
-        os.makedirs(video_folder, exist_ok=True)
-        images_folder = os.path.join(video_folder, "images")
-        audio_folder = os.path.join(video_folder, "audio")
-        os.makedirs(images_folder, exist_ok=True)
-        os.makedirs(audio_folder, exist_ok=True)
-
-        for scene_count, scene in enumerate(scenes_list):
-            text = scene.get("text", "")
-            image_prompts = scene.get("image_prompts", [])
-            audio_path = os.path.join(audio_folder, f"scene_{scene_count + 1}.mp3")
-            audio_result = generate_audio(text, language, speaker, path=audio_path)
-            if "error" in audio_result:
-                continue
-            scene_images_folder = os.path.join(images_folder, f"scene_{scene_count + 1}")
-            os.makedirs(scene_images_folder, exist_ok=True)
-            for count, prompt in enumerate(image_prompts):
-                image_path = os.path.join(scene_images_folder, f"scene_{scene_count + 1}_image_{count + 1}.png")
-                generate_image(prompt=prompt, path=image_path)
-
-        return video_folder
-    except Exception as e:
-        st.error(f"Error during video asset generation: {e}")
-        return ""
+        st.error(f"Error generating image: {e}")
+        return None
 
-def generate_video(video_folder: str, output_filename: str = "final_video.mp4"):
+def generate_video(images, audio_text, language, speaker, path='temp_video.mp4'):
     try:
-        audio_folder = os.path.join(video_folder, "audio")
-        images_folder = os.path.join(video_folder, "images")
-        final_clips = []
-        scene_folders = [
-            os.path.join(images_folder, scene)
-            for scene in natsorted(os.listdir(images_folder))
-            if os.path.isdir(os.path.join(images_folder, scene))
-        ]
-        for scene_path in scene_folders:
-            scene_name = os.path.basename(scene_path)
-            audio_path = os.path.join(audio_folder, f"{scene_name}.mp3")
-            if not os.path.exists(audio_path):
-                continue
-            image_files = natsorted([
-                os.path.join(scene_path, img)
-                for img in os.listdir(scene_path)
-                if img.lower().endswith(('.png', '.jpg', '.jpeg'))
-            ])
-            if not image_files:
-                continue
-            audio_clip = mp.AudioFileClip(audio_path)
-            duration_per_image = audio_clip.duration / len(image_files)
-            image_clips = [mp.ImageClip(img).set_duration(duration_per_image) for img in image_files]
-            scene_video = mp.concatenate_videoclips(image_clips, method="compose").set_audio(audio_clip)
-            final_clips.append(scene_video)
-        if not final_clips:
-            st.error("No valid scenes processed.")
-            return None
-        final_video = mp.concatenate_videoclips(final_clips, method="compose")
-        output_path = os.path.join(video_folder, output_filename)
-        final_video.write_videofile(output_path, fps=24, codec='libx264')
-        return output_path
+        audio_client = Client("habib926653/Multilingual-TTS")
+        audio_result = audio_client.predict(
+            text=audio_text,
+            language_code=language,
+            speaker=speaker,
+            api_name="/text_to_speech_edge"
+        )
+        audio_file = audio_result[1]
+        with open(audio_file, 'rb') as f:
+            audio_bytes = f.read()
+        audio_path = "temp_audio.mp3"
+        with open(audio_path, 'wb') as f:
+            f.write(audio_bytes)
+        
+        audio_clip = mp.AudioFileClip(audio_path)
+        duration_per_image = audio_clip.duration / len(images)
+        image_clips = [mp.ImageClip(img).set_duration(duration_per_image) for img in images if img]
+        video = mp.concatenate_videoclips(image_clips, method="compose").set_audio(audio_clip)
+        video.write_videofile(path, fps=24, codec='libx264')
+        return path
     except Exception as e:
-        st.error(f"Error during video generation: {e}")
+        st.error(f"Error generating video: {e}")
         return None
 
 # Streamlit App
 def main():
-    st.markdown("<h1 style='text-align: center;'>Text to Video Generator</h1>", unsafe_allow_html=True)
-    st.markdown("<p style='text-align: center;'>Leave a Like if it works for you! ❤️</p>", unsafe_allow_html=True)
-
-    text_script = st.text_area("Enter your script/story (max 1500 characters):", max_chars=1500)
-    language = st.selectbox("Choose Language:", ["Urdu", "English"])
-    client = Client("habib926653/Multilingual-TTS")
-    speakers_response = client.predict(language=language, api_name="/get_speakers")
-    speakers = [choice[0] for choice in speakers_response["choices"]]
-    selected_speaker = st.selectbox("Choose Speaker:", speakers)
-
-    if st.button("Generate Video"):
-        if text_script:
-            with st.spinner("Generating video... This may take a few minutes."):
-                scenes = get_scenes(text_script)
-                if not scenes:
-                    st.error("Failed to generate scenes.")
-                else:
-                    video_assets_folder = generate_video_assets(scenes, language, selected_speaker)
-                    if video_assets_folder:
-                        generated_video_path = generate_video(video_assets_folder)
-                        if generated_video_path:
-                            st.success("Video generated successfully!")
-                            st.video(generated_video_path)
+    st.markdown("<h1 style='text-align: center;'>EduAI: Your Interactive Tutor</h1>", unsafe_allow_html=True)
+    st.markdown("<p style='text-align: center;'>Learn, Ask, Visualize! ❤️</p>", unsafe_allow_html=True)
+
+    subject = st.selectbox("Choose Subject:", ["Math", "Science", "History", "Literature", "Code", "AI"])
+    difficulty = st.selectbox("Difficulty Level:", ["Beginner", "Intermediate", "Advanced"])
+    model = st.selectbox("Choose LLM Model:", list(LLM_MODELS.keys()))
+    student_input = st.text_area("Your Question/Input (max 1500 chars):", max_chars=1500)
+
+    if 'tutor_response' not in st.session_state:
+        st.session_state.tutor_response = None
+
+    if st.button("Generate Answer & Question"):
+        if student_input:
+            with st.spinner("Generating your lesson..."):
+                response = generate_tutor_output(subject, difficulty, student_input, LLM_MODELS[model])
+                st.session_state.tutor_response = response
         else:
-            st.warning("Please enter some text to generate a video.")
-
-    st.markdown("### 🔥 See How It Works (Example)")
-    example_script = """
-    One hot summer day, a thirsty crow was flying in search of water. He looked everywhere, but he couldn't find a single drop. Tired and exhausted, he finally spotted a clay pot with a little water at the bottom.
-    """
-    st.markdown(f"**Example Script:** {example_script}")
+            st.warning("Please provide an input!")
+
+    if st.session_state.tutor_response:
+        st.markdown("### Lesson")
+        st.write(st.session_state.tutor_response["lesson"])
+        st.markdown("### Comprehension Question")
+        st.write(st.session_state.tutor_response["question"])
+        st.markdown("### Feedback")
+        st.write(st.session_state.tutor_response["feedback"])
+
+        col1, col2 = st.columns(2)
+        with col1:
+            if st.button("Generate Image"):
+                with st.spinner("Creating image..."):
+                    image_path = generate_image(st.session_state.tutor_response["lesson"])
+                    if image_path:
+                        st.image(image_path, caption="Visual of your lesson")
+        with col2:
+            if st.button("Generate Video"):
+                with st.spinner("Creating video..."):
+                    audio_client = Client("habib926653/Multilingual-TTS")
+                    speakers_response = audio_client.predict(language="English", api_name="/get_speakers")
+                    speaker = speakers_response["choices"][0][0]
+                    images = [generate_image(st.session_state.tutor_response["lesson"])]
+                    video_path = generate_video(images, st.session_state.tutor_response["lesson"], "English", speaker)
+                    if video_path:
+                        st.video(video_path)
+
+    st.markdown("---")
+    st.markdown("<p style='text-align: center;'>Built for learning, powered by AI!</p>", unsafe_allow_html=True)
 
 if __name__ == "__main__":
     main()