feat: app.py

app.py

@@ -0,0 +1,242 @@
+from __future__ import annotations
+import gradio as gr
+from config.settings import SETTINGS
+from services.model_router import ModelRouter
+from quiz.models import QuizSession
+from quiz.engine import QuizEngine
+from storage.local_db import DB
+from storage.mastery import MasteryStore
+from agents.orchestrator import LearningOrchestrator
+from ui.theme import build_champ_theme, CHAMP_CSS
+
+print("[StudyWithChampAI] Initializing...")
+_db = DB(SETTINGS.db_path)
+_mastery_store = MasteryStore(_db)
+_router = ModelRouter(
+    ocr_model=SETTINGS.ocr_model_id,
+    reasoning_model=SETTINGS.reasoning_model_id,
+    multilingual_model=SETTINGS.multilingual_model_id,
+    speech_model=SETTINGS.speech_model_id,
+    hf_api_key=SETTINGS.hf_api_key,
+    featherless_api_key=SETTINGS.featherless_api_key,
+    max_tokens=SETTINGS.max_new_tokens,
+    temperature=SETTINGS.temperature,
+)
+_orchestrator = LearningOrchestrator(
+    _router, _db, _mastery_store, questions_per_quest=SETTINGS.questions_per_quest)
+_engine = QuizEngine()
+_QUEST_CACHE: list = []
+_SOURCE_TEXT_CACHE: str = ""  # retained for revision quest generation
+print("[StudyWithChampAI] Ready.")
+
+
+def handle_process(file_obj, pasted_text: str, audio_obj, language: str):
+    global _QUEST_CACHE, _SOURCE_TEXT_CACHE
+    yield "Processing...", []
+    try:
+        quests = None
+        if audio_obj is not None:
+            yield "Transcribing voice input with Whisper...", []
+            with open(audio_obj, "rb") as f:
+                audio_bytes = f.read()
+            quests = _orchestrator.process_voice(audio_bytes, language)
+        elif file_obj is not None:
+            yield "Processing file with MiniCPM-V...", []
+            quests = _orchestrator.process_file(file_obj.name, language)
+        elif pasted_text and pasted_text.strip():
+            _SOURCE_TEXT_CACHE = pasted_text.strip()
+            yield "Extracting concepts with MiniCPM-V...", []
+            quests = _orchestrator.process_text(pasted_text.strip(), language)
+        else:
+            yield "No input provided. Upload a file, paste text, or record audio.", []; return
+
+        _QUEST_CACHE = quests
+        count = sum(len(q.questions) for q in quests)
+        yield (f"Generated {len(quests)} quests with {count} questions "
+               f"({count - len(quests)} regular + {len(quests)} boss battles). Go to **Quest Map**!",
+               [q.name for q in quests])
+    except Exception as exc:
+        yield f"Error: {exc}", []
+
+
+def handle_select_quest(quest_name: str):
+    quest = next((q for q in _QUEST_CACHE if q.name == quest_name), None)
+    if quest is None:
+        return None, "Quest not found."
+    session = QuizSession(quest_name=quest.name)
+    session.questions = quest.questions
+    return session, f"Quest '{quest_name}' ready! Go to Battle Mode."
+
+
+def handle_answer(selected_option: str, session: QuizSession, question):
+    if session is None or question is None:
+        return "No active session.", "", "", "", "", gr.update(visible=False), gr.update(visible=True), session
+    selected_idx = question.options.index(selected_option) if selected_option in question.options else -1
+    result = _engine.submit_next(session, selected_idx)
+    tutor_hint = ""
+    if not result.is_correct:
+        tutor_hint = _orchestrator.get_tutor_hint(question, selected_option)
+    boss_label = "👑 BOSS BATTLE! " if result.was_boss else ""
+    return (
+        f"{boss_label}{'✅ CORRECT!' if result.is_correct else '❌ WRONG!'}",
+        f"+{result.xp_delta} XP" if result.xp_delta > 0 else "",
+        result.streak_label,
+        question.explanation,
+        tutor_hint,
+        gr.update(visible=True),
+        gr.update(visible=False),
+        session,
+    )
+
+
+def handle_translate(hint_text: str, target_lang: str):
+    if not hint_text:
+        return "_No hint to translate._"
+    return _orchestrator.translate_hint(hint_text, target_lang)
+
+
+def handle_session_complete(session: QuizSession):
+    if session is None:
+        return "No session.", "", "", "", ""
+    result = _orchestrator.complete_quest(session)
+    all_mastery = _mastery_store.all_mastery()
+    mastery_lines = "\n".join(
+        f"- **{t}**: {'█' * int(v * 10)}{'░' * (10 - int(v * 10))} {int(v*100)}%"
+        for t, v in all_mastery.items())
+    weak = result.get("weak_topics", [])
+    weak_lines = ("\n".join(f"- {t} — revision needed" for t in weak)
+                  if weak else "_All topics strong!_ 🏆")
+    from quiz.scoring import compute_grade
+    grade = compute_grade(session.score, len(session.questions))
+    return (f"### {session.score}/{len(session.questions)} correct",
+            f"### Grade: {grade}", f"### XP Earned: {session.xp_earned}",
+            mastery_lines or "_No mastery data yet._", weak_lines)
+
+
+def handle_revision_quest(session: QuizSession):
+    """Generate an adaptive revision quest for weak topics and add it to the cache."""
+    global _QUEST_CACHE
+    weak = _mastery_store.weak_topics()
+    if not weak:
+        return "No weak topics found — all strong!", [q.name for q in _QUEST_CACHE]
+    try:
+        revision = _orchestrator.generate_revision_quest(weak, source_text=_SOURCE_TEXT_CACHE)
+        _QUEST_CACHE.append(revision)
+        return (f"Revision quest unlocked: **{revision.name}**\nGo to Quest Map!",
+                [q.name for q in _QUEST_CACHE])
+    except Exception as exc:
+        return f"Error generating revision quest: {exc}", [q.name for q in _QUEST_CACHE]
+
+
+def build_app() -> gr.Blocks:
+    with gr.Blocks(theme=build_champ_theme(), css=CHAMP_CSS, title="StudyWithChampAI") as demo:
+        gr.Markdown("# StudyWithChampAI\n### Turn notes into quests. Turn studying into progression.")
+
+        quest_names_state = gr.State([])
+        session_state = gr.State(None)
+        current_q_state = gr.State(None)
+        q_idx_state = gr.State(0)
+
+        with gr.Tab("Import Material"):
+            gr.Markdown("### Upload PDF, image, voice, or paste text.\nMiniCPM-V reads and understands your material.")
+            with gr.Row():
+                file_input = gr.File(label="Upload File", file_types=[".pdf",".txt",".png",".jpg",".jpeg"])
+                audio_input = gr.Audio(label="Record Voice", type="filepath", sources=["microphone"])
+            text_input = gr.Textbox(label="Or paste notes here", lines=6, placeholder="Paste study material...")
+            lang_dropdown = gr.Dropdown(choices=["English","Hindi","Bengali","Spanish","French"],
+                                        value="English", label="Quiz Language")
+            process_btn = gr.Button("Generate Quests ⚔️", variant="primary", size="lg")
+            status_md = gr.Markdown("_Ready._")
+            process_btn.click(fn=handle_process,
+                              inputs=[file_input, text_input, audio_input, lang_dropdown],
+                              outputs=[status_md, quest_names_state])
+
+        with gr.Tab("Quest Map"):
+            gr.Markdown("### Select a quest to begin.")
+            quest_radio = gr.Radio(choices=[], label="Available Quests", interactive=True)
+            start_btn = gr.Button("Enter Battle ⚔️", variant="primary")
+            quest_status_md = gr.Markdown("")
+            quest_names_state.change(fn=lambda n: gr.update(choices=n, visible=bool(n)),
+                                     inputs=[quest_names_state], outputs=[quest_radio])
+            start_btn.click(fn=handle_select_quest, inputs=[quest_radio],
+                            outputs=[session_state, quest_status_md])
+
+        with gr.Tab("Battle Mode"):
+            gr.Markdown("### Answer to earn XP! Boss battle always awaits at quest end.")
+            question_md = gr.Markdown("_Select a quest first._")
+            progress_md = gr.Markdown("")
+            answer_radio = gr.Radio(choices=[], label="Your Answer", interactive=True)
+            submit_btn = gr.Button("Submit Answer", variant="primary")
+
+            with gr.Group(visible=False) as feedback_group:
+                result_md = gr.Markdown("")
+                xp_md = gr.Markdown("")
+                streak_md = gr.Markdown("")
+                explanation_md = gr.Markdown("")
+                tutor_md = gr.Markdown("")
+                with gr.Row():
+                    translate_lang = gr.Dropdown(choices=["Hindi","Bengali","Spanish","French"],
+                                                 value="Hindi", label="Translate hint to", scale=1)
+                    translate_btn = gr.Button("Translate 🌐", scale=1)
+                translated_md = gr.Markdown("")
+                next_btn = gr.Button("Next Question →")
+
+            def load_question(session, idx):
+                if session is None:
+                    return "_No active session._", "", gr.update(choices=[], visible=False), gr.update(visible=False), None, idx
+                if idx >= len(session.questions):
+                    return "_Quest complete! Go to Results tab._", "", gr.update(choices=[], visible=False), gr.update(visible=False), None, idx
+                q = session.questions[idx]
+                header = "👑 **BOSS BATTLE** 👑\n\n" if q.is_boss else ""
+                return (f"{header}**Q{idx+1}/{len(session.questions)}**\n\n{q.text}",
+                        f"Progress: {idx}/{len(session.questions)} | XP: {session.xp_earned}",
+                        gr.update(choices=q.options, visible=True, value=None),
+                        gr.update(visible=True), q, idx)
+
+            session_state.change(fn=load_question, inputs=[session_state, q_idx_state],
+                                 outputs=[question_md, progress_md, answer_radio, submit_btn, current_q_state, q_idx_state])
+            submit_btn.click(fn=handle_answer, inputs=[answer_radio, session_state, current_q_state],
+                             outputs=[result_md, xp_md, streak_md, explanation_md, tutor_md,
+                                      feedback_group, submit_btn, session_state])
+            translate_btn.click(fn=handle_translate, inputs=[tutor_md, translate_lang],
+                                outputs=[translated_md])
+
+            def advance_question(session, idx):
+                new_idx = idx + 1
+                if session is None or new_idx >= len(session.questions):
+                    return ("_Quest complete! Go to Results._", "",
+                            gr.update(choices=[], visible=False), gr.update(visible=False),
+                            gr.update(visible=False), None, new_idx)
+                q = session.questions[new_idx]
+                header = "👑 **BOSS BATTLE** 👑\n\n" if q.is_boss else ""
+                return (f"{header}**Q{new_idx+1}/{len(session.questions)}**\n\n{q.text}",
+                        f"Progress: {new_idx}/{len(session.questions)} | XP: {session.xp_earned}",
+                        gr.update(choices=q.options, visible=True, value=None),
+                        gr.update(visible=True), gr.update(visible=False), q, new_idx)
+
+            next_btn.click(fn=advance_question, inputs=[session_state, q_idx_state],
+                           outputs=[question_md, progress_md, answer_radio, submit_btn,
+                                    feedback_group, current_q_state, q_idx_state])
+
+        with gr.Tab("Results"):
+            gr.Markdown("## Quest Complete! 🏆")
+            score_md = gr.Markdown("")
+            grade_md = gr.Markdown("")
+            xp_total_md = gr.Markdown("")
+            mastery_md = gr.Markdown("### Mastery Map")
+            weak_md = gr.Markdown("")
+            with gr.Row():
+                finish_btn = gr.Button("View Results", variant="primary")
+                revision_btn = gr.Button("⚔️ Generate Revision Quest", variant="secondary")
+            revision_status_md = gr.Markdown("")
+            finish_btn.click(fn=handle_session_complete, inputs=[session_state],
+                             outputs=[score_md, grade_md, xp_total_md, mastery_md, weak_md])
+            revision_btn.click(fn=handle_revision_quest, inputs=[session_state],
+                               outputs=[revision_status_md, quest_names_state])
+
+    return demo
+
+
+if __name__ == "__main__":
+    app = build_app()
+    app.launch(share=False, server_name="0.0.0.0", server_port=7860)