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)