from datetime import datetime, timedelta from typing import Dict, List from bson import ObjectId import pandas as pd import streamlit as st import json import google.generativeai as genai from dotenv import load_dotenv import os from pymongo import MongoClient import plotly.express as px load_dotenv() MONGO_URI = os.getenv('MONGO_URI') GEMINI_API_KEY = os.getenv('GEMINI_KEY') genai.configure(api_key=GEMINI_API_KEY) model = genai.GenerativeModel("gemini-1.5-flash") client = MongoClient(MONGO_URI) db = client['novascholar_db'] courses_collection = db['courses'] def generate_interval_questions(context: str, num_questions: int) -> List[Dict]: """Generate all interval questions at once""" prompt = f""" Generate {num_questions} quick check-in questions based on the following context. Return ONLY a JSON array with this exact structure: [ {{ "question_text": "Your question here", "type": "mcq OR short_answer", "options": ["option1", "option2", "option3", "option4"], "correct_option": "correct answer", "explanation": "brief explanation" }} ] Requirements: - Each question should be answerable in 30-60 seconds - Test understanding of different aspects of the material - For MCQ, include exactly 4 options - Questions should build in complexity - **MAKE SURE THE NUMBER OF SHORT ANSWER QUESTIONS ARE MORE THAN MCQS** - **MAKE SURE THE NUMBER OF MCQS DOES NOT EXCEED 2** Context: {context} """ try: response = model.generate_content( prompt, generation_config=genai.GenerationConfig( temperature=0.3, response_mime_type="application/json" ) ) if not response.text: raise ValueError("Empty response from model") questions = json.loads(response.text) return questions except Exception as e: st.error(f"Error generating questions: {e}") return None class IntervalQuestionManager: def __init__(self, session_id: str, course_id: str): self.session_id = session_id self.course_id = course_id self.questions = [] self.current_index = 0 self.interval_minutes = 10 # self.start_time = self._get_start_time() self.start_time = None # self.questions = self._load_existing_questions() self._initialize_from_db() # def _get_start_time(self): # """Retrieve start time from database""" # try: # session_data = courses_collection.find_one( # { # "course_id": self.course_id, # "sessions.session_id": self.session_id # }, # {"sessions.$": 1} # ) # if session_data and session_data.get('sessions'): # session = session_data['sessions'][0] # interval_data = session.get('in_class', {}).get('interval_questions', {}) # if interval_data and 'start_time' in interval_data: # return interval_data['start_time'] # return None # except Exception as e: # st.error(f"Error getting start time: {e}") # return None # def _load_existing_questions(self) -> List[Dict]: # """Load existing interval questions from database""" # try: # session_data = courses_collection.find_one( # { # "course_id": self.course_id, # "sessions.session_id": self.session_id # }, # {"sessions.$": 1} # ) # if session_data and session_data.get('sessions'): # session = session_data['sessions'][0] # interval_data = session.get('in_class', {}).get('interval_questions', {}) # if interval_data and interval_data.get('questions'): # print("Found existing questions:", len(interval_data['questions'])) # return interval_data['questions'] # return [] # except Exception as e: # st.error(f"Error loading questions: {e}") # return [] def _initialize_from_db(self): """Load existing questions and settings from database""" try: session_data = courses_collection.find_one( { "course_id": self.course_id, "sessions.session_id": self.session_id }, {"sessions.$": 1} ) if session_data and session_data.get('sessions'): session = session_data['sessions'][0] interval_data = session.get('in_class', {}).get('interval_questions', {}) if interval_data: self.questions = interval_data.get('questions', []) self.interval_minutes = interval_data.get('interval_minutes', 10) self.start_time = interval_data.get('start_time') print(f"Loaded {len(self.questions)} questions, interval: {self.interval_minutes} mins") except Exception as e: st.error(f"Error initializing from database: {e}") def initialize_questions(self, context: str, num_questions: int, interval_minutes: int): if self.questions: # If questions already exist, don't regenerate return True """Initialize all questions and save to database""" questions = generate_interval_questions(context, num_questions) if not questions: return False self.questions = [] self.interval_minutes = interval_minutes for i, q in enumerate(questions): question_doc = { "_id": ObjectId(), "question_text": q["question_text"], "type": q["type"], "options": q.get("options", []), "correct_option": q.get("correct_option", ""), "explanation": q.get("explanation", ""), "display_after": i * interval_minutes, # When to display this question "active": False, "submissions": [] } self.questions.append(question_doc) # Save to database result = courses_collection.update_one( { "course_id": self.course_id, "sessions.session_id": self.session_id }, { "$set": { "sessions.$.in_class.interval_questions": { "questions": self.questions, "interval_minutes": self.interval_minutes, "start_time": None } } } ) return result.modified_count > 0 # def start_questions(self): # """Start the interval questions""" # try: # self.start_time = datetime.now() # # Update start time in database # result = courses_collection.update_one( # { # "course_id": self.course_id, # "sessions.session_id": self.session_id # }, # { # "$set": { # "sessions.$.in_class.interval_questions.start_time": self.start_time # } # } # ) # # Activate first question # if self.questions: # self.questions[0]["active"] = True # # Update first question's active status in database # courses_collection.update_one( # { # "course_id": self.course_id, # "sessions.session_id": self.session_id, # "sessions.in_class.questions._id": self.questions[0]["_id"] # }, # { # "$set": { # "sessions.$.in_class.questions.$.active": True, # "sessions.$.in_class.questions.$.started_at": self.start_time # } # } # ) # if result.modified_count > 0: # st.success("Questions started successfully!") # else: # st.error("Could not start questions. Please try again.") # except Exception as e: # st.error(f"Error starting questions: {str(e)}") def start_questions(self): """Start the interval questions""" try: self.start_time = datetime.now() result = courses_collection.update_one( { "course_id": self.course_id, "sessions.session_id": self.session_id }, { "$set": { "sessions.$.in_class.interval_questions.start_time": self.start_time } } ) if result.modified_count > 0: st.success("Questions started successfully!") else: st.error("Could not start questions. Please try again.") except Exception as e: st.error(f"Error starting questions: {str(e)}") # def start_questions(self): # """Start the interval questions""" # self.start_time = datetime.now() # courses_collection.update_one( # { # "course_id": self.course_id, # "sessions.session_id": self.session_id # }, # { # "$set": { # "sessions.$.in_class.interval_questions.start_time": self.start_time # } # } # ) # def get_current_question(self) -> Dict: # # """Get the currently active question based on elapsed time""" # # if not self.start_time: # # return None # # elapsed_minutes = (datetime.now() - self.start_time).total_seconds() / 60 # # # Find the latest question that should be displayed # # for question in reversed(self.questions): # # if elapsed_minutes >= question["display_after"]: # # return question # # return None # """Get current question based on elapsed time""" # try: # if not self.start_time: # return None # questions = self.questions # if not questions: # # Get questions from database # session_data = courses_collection.find_one( # { # "course_id": self.course_id, # "sessions.session_id": self.session_id # }, # {"sessions.$": 1} # ) # if not session_data or not session_data.get('sessions'): # return None # session = session_data['sessions'][0] # interval_data = session.get('in_class', {}).get('interval_questions', {}) # if not interval_data: # return None # questions = interval_data.get('questions', []) # interval_minutes = interval_data.get('interval_minutes', 10) # # Calculate elapsed time # elapsed_minutes = (datetime.now() - self.start_time).total_seconds() / 60 # question_index = int(elapsed_minutes // interval_minutes) # # Debug logging # print(f"Elapsed minutes: {elapsed_minutes}") # print(f"Question index: {question_index}") # print(f"Total questions: {len(questions)}") # # Return current question if valid index # if 0 <= question_index < len(questions): # return questions[question_index] # return None # except Exception as e: # st.error(f"Error getting current question: {e}") # return None def get_current_question(self) -> Dict: """Get current question based on elapsed time""" try: if not self.start_time or not self.questions: return None elapsed_minutes = (datetime.now() - self.start_time).total_seconds() / 60 question_index = int(elapsed_minutes // self.interval_minutes) # Debug logging print(f"Elapsed minutes: {elapsed_minutes}") print(f"Question index: {question_index}") print(f"Total questions: {len(self.questions)}") print(f"Interval minutes: {self.interval_minutes}") if 0 <= question_index < len(self.questions): return self.questions[question_index] return None except Exception as e: st.error(f"Error getting current question: {e}") return None def display_response_distribution(course_id: str, session_id: str, question_id: str): """Display real-time distribution of student responses""" try: session_data = courses_collection.find_one( { "course_id": course_id, "sessions.session_id": session_id }, {"sessions.$": 1} ) if not session_data or not session_data.get('sessions'): st.warning("No responses found.") return session = session_data['sessions'][0] question = next( (q for q in session['in_class']['interval_questions']['questions'] if str(q['_id']) == question_id), None ) if not question: st.warning("Question not found.") return submissions = question.get('submissions', []) if not submissions: st.info("No responses received yet.") return # Calculate response distribution if question['type'] == 'mcq': responses = {} for submission in submissions: answer = submission['answer'] responses[answer] = responses.get(answer, 0) + 1 # Create DataFrame for visualization df = pd.DataFrame(list(responses.items()), columns=['Option', 'Count']) df['Percentage'] = df['Count'] / len(submissions) * 100 # Display metrics total_responses = len(submissions) st.metric("Total Responses", total_responses) # Display bar chart fig = px.bar(df, x='Option', y='Count', title='Response Distribution', text='Percentage') fig.update_traces(texttemplate='%{text:.1f}%', textposition='outside') st.plotly_chart(fig) # Display detailed table st.dataframe(df) else: # Short answer responses st.markdown("##### Responses Received") # for submission in submissions: # st.markdown(f"- {submission['answer']}") st.metric("Total Responses", len(submissions)) except Exception as e: st.error(f"Error displaying responses: {str(e)}") def display_interval_question(course_id, session, question: Dict, user_type: str): """Display the interval question with different views for faculty/students""" if not question: return st.markdown("#### Quick Questions ⚡") try: if user_type == "faculty": with st.container(): st.markdown("##### Current Active Question") st.markdown(f"**Question:** {question['question_text']}") if question['type'] == 'mcq': display_response_distribution(course_id, session['session_id'], str(question['_id'])) # Add controls to end question period col1, col2 = st.columns([3,1]) with col2: if st.button("End Question"): end_question_period(course_id, session['session_id'], str(question['_id'])) st.rerun() # Display all questions overview st.markdown("##### All Questions Overview") # Get all questions from the session session_data = courses_collection.find_one( { "course_id": course_id, "sessions.session_id": session['session_id'] }, {"sessions.$": 1} ) if session_data and session_data.get('sessions'): interval_questions = session_data['sessions'][0].get('in_class', {}).get('interval_questions', {}) all_questions = interval_questions.get('questions', []) for idx, q in enumerate(all_questions, 1): with st.expander(f"Question {idx}", expanded=False): st.markdown(f"**Type:** {q['type'].upper()}") st.markdown(f"**Question:** {q['question_text']}") if q['type'] == 'mcq': st.markdown("**Options:**") for opt in q['options']: if opt == q['correct_option']: st.markdown(f"✅ {opt}") else: st.markdown(f"- {opt}") st.markdown(f"**Explanation:** {q['explanation']}") # Show when this question will be displayed display_time = timedelta(minutes=q['display_after']) st.info(f"Will be displayed after: {display_time}") # Show submission count if any submission_count = len(q.get('submissions', [])) if submission_count > 0: st.metric("Responses received", submission_count) else: # Student view # Check if student has already submitted # existing_submission = next( # (sub for sub in question.get('submissions', []) # if sub['student_id'] == st.session_state.user_id), # None # ) # Check for existing submission existing_submission = _check_existing_submission( course_id, session['session_id'], str(question['_id']), st.session_state.user_id ) if existing_submission: st.success("Response submitted!") st.markdown(f"Your answer: **{existing_submission['answer']}**") else: with st.form(key=f"question_form_{question['_id']}"): st.markdown(f"**Question:** {question['question_text']}") if question['type'] == 'mcq': response = st.radio( "Select your answer:", options=question['options'], key=f"mcq_{question['_id']}" ) else: response = st.text_area( "Your answer:", key=f"text_{question['_id']}" ) if st.form_submit_button("Submit"): if response: submit_response( course_id, session['session_id'], str(question['_id']), st.session_state.user_id, response ) st.rerun() else: st.error("Please provide an answer before submitting.") except Exception as e: st.error(f"Error displaying question: {str(e)}") def _check_existing_submission(course_id: str, session_id: str, question_id: str, student_id: str) -> Dict: """Check if student has already submitted an answer""" try: session_data = courses_collection.find_one( { "course_id": course_id, "sessions.session_id": session_id, "sessions.in_class.questions._id": ObjectId(question_id) }, {"sessions.$": 1} ) if session_data and session_data.get('sessions'): session = session_data['sessions'][0] question = next( (q for q in session['in_class']['questions'] if str(q['_id']) == question_id), None ) if question: return next( (sub for sub in question.get('submissions', []) if sub['student_id'] == student_id), None ) return None except Exception as e: st.error(f"Error checking submission: {e}") return None def end_question_period(course_id: str, session_id: str, question_id: str): """End the question period and mark the question as inactive""" try: result = courses_collection.update_one( { "course_id": course_id, "sessions.session_id": session_id, "sessions.in_class.questions._id": ObjectId(question_id) }, { "$set": { "sessions.$.in_class.questions.$.active": False, "sessions.$.in_class.questions.$.ended_at": datetime.utcnow() } } ) if result.modified_count > 0: st.success("Question period ended successfully!") st.rerun() else: st.warning("Could not end question period. Question may not exist.") except Exception as e: st.error(f"Error ending question period: {e}") # def submit_response(course_id: str, session_id: str, question_id: str, student_id: str, response: str): # """Submit a student's response to a question""" # try: # # First find the specific session and question # session_query = { # "course_id": course_id, # "sessions": { # "$elemMatch": { # "session_id": session_id, # "in_class.interval_questions.questions._id": ObjectId(question_id) # } # } # } # # Check for existing submission # session_doc = courses_collection.find_one(session_query) # if not session_doc: # st.error("Question not found") # return # session = next((s for s in session_doc['sessions'] if s['session_id'] == session_id), None) # if not session: # st.error("Session not found") # return # question = next((q for q in session['in_class']['questions'] # if str(q['_id']) == question_id), None) # if not question: # st.error("Question not found") # return # # Check for existing submission # if any(sub['student_id'] == student_id for sub in question.get('submissions', [])): # st.warning("You have already submitted an answer for this question.") # return # # Add new submission # result = courses_collection.update_one( # { # "course_id": course_id, # "sessions.session_id": session_id, # "sessions.in_class.interval_questions.questions._id": ObjectId(question_id) # }, # { # "$push": { # "sessions.$[session].in_class.interval_questions.questions.$[question].submissions": { # "student_id": student_id, # "answer": response, # "submitted_at": datetime.utcnow() # } # } # }, # array_filters=[ # {"session.session_id": session_id}, # {"question._id": ObjectId(question_id)} # ] # ) # if result.modified_count > 0: # st.success("Response submitted successfully!") # st.rerun() # else: # st.error("Could not submit response. Question may not exist or be inactive.") # except Exception as e: # st.error(f"Error submitting response: {str(e)}") def submit_response(course_id: str, session_id: str, question_id: str, student_id: str, response: str): """Submit a student's response to a question""" try: # Debug prints print(f"Submitting response for question {question_id}") print(f"Course: {course_id}, Session: {session_id}") # First find the specific session and question with correct path session_doc = courses_collection.find_one( { "course_id": course_id, "sessions.session_id": session_id }, {"sessions.$": 1} ) if not session_doc: print("Session document not found") st.error("Session not found") return session = session_doc['sessions'][0] # Get interval questions interval_questions = session.get('in_class', {}).get('interval_questions', {}) if not interval_questions: print("No interval questions found") st.error("No questions found") return questions = interval_questions.get('questions', []) # Find specific question question = next( (q for q in questions if str(q['_id']) == question_id), None ) if not question: print(f"Question {question_id} not found in questions list") st.error("Question not found") return # Check for existing submission if any(sub['student_id'] == student_id for sub in question.get('submissions', [])): st.warning("You have already submitted an answer for this question.") return # Add new submission with correct path result = courses_collection.update_one( { "course_id": course_id, "sessions.session_id": session_id, "sessions.in_class.interval_questions.questions._id": ObjectId(question_id) }, { "$push": { "sessions.$[ses].in_class.interval_questions.questions.$[q].submissions": { "student_id": student_id, "answer": response, "submitted_at": datetime.utcnow() } } }, array_filters=[ {"ses.session_id": session_id}, {"q._id": ObjectId(question_id)} ] ) print(f"Update result: {result.modified_count}") if result.modified_count > 0: st.success("Response submitted successfully!") st.rerun() else: st.error("Could not submit response. Please try again.") except Exception as e: print(f"Error details: {str(e)}") st.error(f"Error submitting response: {str(e)}")