Spaces:
Running
Running
| 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) | |