', unsafe_allow_html=True)
col1, col2, col3 = st.columns(3)
with col1:
concept_understanding = st.selectbox(
"Filter by Understanding",
["All", "Strong", "Moderate", "Needs Improvement"]
)
with col2:
participation_level = st.selectbox(
"Filter by Participation",
["All", "High (>80%)", "Medium (50-80%)", "Low (<50%)"]
)
with col3:
struggling_topic = st.selectbox(
"Filter by Struggling Topic",
["All"] + list(set([topic for student in analytics["student_analytics"]
for topic in student["struggling_topics"]]))
)
# st.markdown('
', unsafe_allow_html=True)
# Display student metrics in a grid
# st.markdown('', unsafe_allow_html=True)
student_ids = [student["student_id"] for student in analytics["student_analytics"]]
student_names = {str(student["_id"]): student["full_name"] for student in students_collection.find({"_id": {"$in": [ObjectId(sid) for sid in student_ids]}})}
# Display student metrics in a grid
st.markdown('
', unsafe_allow_html=True)
for student in analytics["student_analytics"]:
# Apply filters
if (concept_understanding != "All" and
student["engagement_metrics"]["concept_understanding"].replace("_", " ").title() != concept_understanding):
continue
participation = student["engagement_metrics"]["participation_level"] * 100
if participation_level != "All":
if participation_level == "High (>80%)" and participation <= 80:
continue
elif participation_level == "Medium (50-80%)" and (participation < 50 or participation > 80):
continue
elif participation_level == "Low (<50%)" and participation >= 50:
continue
if struggling_topic != "All" and struggling_topic not in student["struggling_topics"]:
continue
student_name = student_names.get(student["student_id"], "Unknown Student")
st.markdown(f"""
Participation
{student["engagement_metrics"]["participation_level"]*100:.1f}%
Understanding
{student["engagement_metrics"]["concept_understanding"].replace('_', ' ').title()}
Struggling Topics:
{", ".join(student["struggling_topics"]) if student["struggling_topics"] else "None"}
{student["personalized_recommendation"]}
""", unsafe_allow_html=True)
st.markdown('
', unsafe_allow_html=True)
def reset_analytics_state():
"""
Helper function to reset the analytics state when needed
(e.g., when loading a new session or when data needs to be refreshed)
"""
if 'analytics_data' in st.session_state:
del st.session_state.analytics_data
if 'expanded_topic' in st.session_state:
del st.session_state.expanded_topic
if 'topic_indices' in st.session_state:
del st.session_state.topic_indice
def display_session_analytics(session, course_id):
"""Display session analytics for faculty"""
st.header("Session Analytics")
# Display Pre-class Analytics
display_preclass_analytics2(session, course_id)
# Display In-class Analytics
display_inclass_analytics(session, course_id)
# Display Post-class Analytics
display_postclass_analytics(session, course_id)
# def upload_preclass_materials(session_id, course_id):
# """Upload pre-class materials for a session"""
# st.subheader("Upload Pre-class Materials")
# # File upload section
# uploaded_file = st.file_uploader("Upload Material", type=['txt', 'pdf', 'docx'])
# if uploaded_file is not None:
# with st.spinner("Processing document..."):
# file_name = uploaded_file.name
# file_content = extract_text_from_file(uploaded_file)
# if file_content:
# material_type = st.selectbox("Select Material Type", ["pdf", "docx", "txt"])
# if st.button("Upload Material"):
# upload_resource(course_id, session_id, file_name, uploaded_file, material_type)
# # Search for the newly uploaded resource's _id in resources_collection
# resource_id = resources_collection.find_one({"file_name": file_name})["_id"]
# create_vector_store(file_content, resource_id)
# st.success("Material uploaded successfully!")
# # Display existing materials
# materials = resources_collection.find({"course_id": course_id, "session_id": session_id})
# for material in materials:
# st.markdown(f"""
# * **{material['file_name']}** ({material['material_type']})
# Uploaded on: {material['uploaded_at'].strftime('%Y-%m-%d %H:%M')}
# """)
def display_quiz_tab(student_id, course_id, session_id):
"""Display quizzes for students"""
st.header("Course Quizzes")
# Get available quizzes for this session, excluding pre-class quizzes
# quizzes = quizzes_collection.find({
# "course_id": course_id,
# "session_id": session_id,
# "status": "active",
# "quiz_type": {"$ne": "pre_class"}
# })
quizzes = list(quizzes_collection.find({
"course_id": course_id,
"session_id": session_id,
"status": "active",
"quiz_type": {"$ne": "pre_class"}
}))
# Get available quizzes for this session
# quizzes = quizzes_collection.find({
# "course_id": course_id,
# "session_id": session_id,
# "status": "active"
# })
# quizzes = list(quizzes)
if not quizzes:
st.info("No quizzes available for this session.")
return
# for quiz in quizzes:
# with st.expander(f"📝 {quiz['title']}", expanded=True):
# # Check if student has already taken this quiz
# existing_score = get_student_quiz_score(quiz['_id'], student_id)
# if existing_score is not None:
# st.success(f"Quiz completed! Your score: {existing_score:.1f}%")
# # Display correct answers after submission
# st.subheader("Quiz Review")
# for i, question in enumerate(quiz['questions']):
# st.markdown(f"**Question {i+1}:** {question['question']}")
# for opt in question['options']:
# if opt.startswith(question['correct_option']):
# st.markdown(f"✅ {opt}")
# else:
# st.markdown(f"- {opt}")
# else:
# # Display quiz questions
# st.write("Please select your answers:")
# # Create a form for quiz submission
# with st.form(f"quiz_form_{quiz['_id']}"):
# student_answers = {}
# for i, question in enumerate(quiz['questions']):
# st.markdown(f"**Question {i+1}:** {question['question']}")
# options = [opt for opt in question['options']]
# student_answers[str(i)] = st.radio(
# f"Select answer for question {i+1}:",
# options=options,
# key=f"q_{quiz['_id']}_{i}"
# )
# # Submit button
# if st.form_submit_button("Submit Quiz"):
# print(student_answers)
# score = submit_quiz_answers(quiz['_id'], student_id, student_answers)
# if score is not None:
# st.success(f"Quiz submitted successfully! Your score: {score:.1f}%")
# st.rerun() # Refresh to show results
# else:
# st.error("Error submitting quiz. Please try again.")
for idx, quiz in enumerate(quizzes):
with st.expander(f"📝 {quiz.get('title', 'Quiz')}", expanded=True):
# Check if student has already taken this quiz
existing_score = get_student_quiz_score(quiz['_id'], student_id)
if existing_score is not None:
st.success(f"Quiz completed! Your score: {existing_score:.1f}%")
# Display correct answers after submission
st.subheader("Quiz Review")
for i, question in enumerate(quiz['questions']):
st.markdown(f"**Question {i+1}:** {question['question']}")
for opt in question['options']:
if opt == question['correct_option']:
st.markdown(f"✅ {opt}")
else:
st.markdown(f"- {opt}")
else:
# Display quiz questions
st.write("Please select your answers:")
# Create a form with unique key using both quiz ID and index
form_key = f"quiz_form_{quiz['_id']}_{idx}"
with st.form(key=form_key):
student_answers = {}
for i, question in enumerate(quiz['questions']):
st.markdown(f"**Question {i+1}:** {question['question']}")
options = question['options']
answer = st.radio(
f"Select answer:",
options=options,
key=f"q_{quiz['_id']}_{idx}_{i}", # Make radio button key unique too
index=None
)
if answer:
student_answers[str(i)] = answer
# Submit button
if st.form_submit_button("Submit Quiz"):
if len(student_answers) != len(quiz['questions']):
st.error("Please answer all questions before submitting.")
else:
score = submit_quiz_answers(quiz['_id'], student_id, student_answers)
if score is not None:
st.success(f"Quiz submitted successfully! Your score: {score:.1f}%")
st.rerun()
else:
st.error("Error submitting quiz. Please try again.")
def display_session_content(student_id, course_id, session, username, user_type):
st.title(f"{session['title']}")
# Check if the date is a string or a datetime object
if isinstance(session['date'], str):
# Convert date string to datetime object
session_date = datetime.fromisoformat(session['date'])
else:
session_date = session['date']
course_name = courses_collection.find_one({"course_id": course_id})['title']
st.markdown(f"**Date:** {format_datetime(session_date)}")
st.markdown(f"**Course Name:** {course_name}")
# Find the course_id of the session in
if user_type == 'student':
# Create all tabs at once for students
tabs = st.tabs([
"Pre-class Work",
"In-class Work",
"Post-class Work",
"Quizzes",
"Subjective Tests",
"Group Work",
"End Terms",
"Code Playground"
])
if len(tabs) <= 8:
with tabs[0]:
display_preclass_content(session, student_id, course_id)
with tabs[1]:
display_in_class_content(session, user_type, course_id, user_type)
with tabs[2]:
display_post_class_content(session, student_id, course_id)
with tabs[3]:
display_quiz_tab(student_id, course_id, session['session_id'])
with tabs[4]:
display_subjective_test_tab(student_id, course_id, session['session_id'])
with tabs[5]:
st.subheader("Group Work")
st.info("Group work content will be available soon.")
with tabs[6]:
st.subheader("End Terms")
st.info("End term content will be available soon.")
with tabs[7]:
# st.subheader("Code Playground")
display_code_playground(session, student_id, course_id)
else:
st.error("Error creating tabs. Please try again.")
else: # faculty user
# Create all tabs at once for faculty
tabs = st.tabs([
"Pre-class Work",
"In-class Work",
"Post-class Work",
"Pre-class Analytics",
"In-class Analytics",
"Post-class Analytics",
"Rubrics",
"End Terms",
"Evaluate Subjective Tests"
])
with tabs[0]:
upload_preclass_materials(session['session_id'], course_id, student_id)
with tabs[1]:
display_in_class_content(session, user_type, course_id, user_type)
with tabs[2]:
display_post_class_content(session, student_id, course_id)
with tabs[3]:
display_preclass_analytics2(session, course_id)
with tabs[4]:
display_inclass_analytics(session, course_id)
with tabs[5]:
display_postclass_analytics(session, course_id)
with tabs[6]:
display_rubrics_tab(session, course_id)
with tabs[7]:
st.subheader("End Terms")
st.info("End term content will be available soon.")
with tabs[8]: # New tab for evaluation
display_evaluation_to_faculty(session['session_id'], student_id, course_id)
def parse_model_response(response_text):
"""Enhanced parser for model responses with better error handling.
Args:
response_text (str): Raw response text from the model
Returns:
dict or list: Parsed response object
Raises:
ValueError: If parsing fails
"""
import json
import ast
import re
# Remove markdown formatting and whitespace
cleaned_text = re.sub(r'```[a-zA-Z]*\n', '', response_text)
cleaned_text = cleaned_text.replace('```', '').strip()
# Try multiple parsing methods
parsing_methods = [
# Method 1: Direct JSON parsing
lambda x: json.loads(x),
# Method 2: AST literal evaluation
lambda x: ast.literal_eval(x),
# Method 3: Extract and parse content between curly braces
lambda x: json.loads(re.search(r'\{.*\}', x, re.DOTALL).group()),
# Method 4: Extract and parse content between square brackets
lambda x: json.loads(re.search(r'\[.*\]', x, re.DOTALL).group()),
# Method 5: Try to fix common JSON formatting issues and parse
lambda x: json.loads(x.replace("'", '"').replace('\n', '\\n'))
]
last_error = None
for parse_method in parsing_methods:
try:
result = parse_method(cleaned_text)
if result: # Ensure we have actual content
return result
except Exception as e:
last_error = e
continue
raise ValueError(f"Could not parse the model's response: {last_error}")
def generate_questions(context, num_questions, session_title, session_description):
"""Generate subjective questions based on context or session details"""
try:
# Construct the prompt
prompt = f"""You are a professional educator creating {num_questions} subjective questions.
Topic: {session_title}
Description: {session_description}
{'Context: ' + context if context else ''}
Generate exactly {num_questions} questions in this specific format:
[
{{"question": "Write your first question here?"}},
{{"question": "Write your second question here?"}}
]
Requirements:
1. Questions must require detailed explanations
2. Focus on critical thinking and analysis
3. Ask for specific examples or case studies
4. Questions should test deep understanding
IMPORTANT: Return ONLY the JSON array. Do not include any additional text or explanations.
"""
# Generate response
response = model.generate_content(prompt)
questions = parse_model_response(response.text)
# Validate response
if not isinstance(questions, list):
raise ValueError("Generated content is not a list")
if len(questions) != num_questions:
raise ValueError(f"Generated {len(questions)} questions instead of {num_questions}")
# Validate each question
for q in questions:
if not isinstance(q, dict) or 'question' not in q:
raise ValueError("Invalid question format")
return questions
except Exception as e:
print(f"Error generating questions: {str(e)}")
return None
def generate_synoptic(questions, context, session_title, num_questions):
"""Generate synoptic answers for the questions with improved error handling and response validation.
Args:
questions (list): List of question dictionaries
context (str): Additional context for answer generation
session_title (str): Title of the session
num_questions (int): Expected number of questions
Returns:
list: List of synoptic answers or None if generation fails
"""
try:
# First, let's validate our input
if not questions or not isinstance(questions, list):
raise ValueError("Questions must be provided as a non-empty list")
# Format questions for better prompt clarity
formatted_questions = "\n".join(
f"{i+1}. {q['question']}"
for i, q in enumerate(questions)
)
# Construct a more structured prompt
prompt = f"""You are a subject matter expert creating detailed model answers for {num_questions} questions about {session_title}.
Here are the questions:
{formatted_questions}
{f'Additional context: {context}' if context else ''}
Please provide {num_questions} comprehensive answers following this JSON format:
{{
"answers": [
{{
"answer": "Your detailed answer for question 1...",
"key_points": ["Point 1", "Point 2", "Point 3"]
}},
{{
"answer": "Your detailed answer for question 2...",
"key_points": ["Point 1", "Point 2", "Point 3"]
}}
]
}}
Requirements for each answer:
1. Minimum 150 words
2. Include specific examples and evidence
3. Reference key concepts and terminology
4. Demonstrate critical analysis
5. Structure with clear introduction, body, and conclusion
IMPORTANT: Return ONLY the JSON object with the answers array. No additional text.
"""
# Generate response
response = model.generate_content(prompt)
# Parse and validate the response
parsed_response = parse_model_response(response.text)
# Additional validation of parsed response
if not isinstance(parsed_response, (dict, list)):
raise ValueError("Response must be a dictionary or list")
# Handle both possible valid response formats
if isinstance(parsed_response, dict):
answers = parsed_response.get('answers', [])
else: # If it's a list
answers = parsed_response
# Validate answer count
if len(answers) != num_questions:
raise ValueError(f"Expected {num_questions} answers, got {len(answers)}")
# Extract just the answer texts for consistency with existing code
final_answers = []
for ans in answers:
if isinstance(ans, dict):
answer_text = ans.get('answer', '')
key_points = ans.get('key_points', [])
formatted_answer = f"{answer_text}\n\nKey Points:\n" + "\n".join(f"• {point}" for point in key_points)
final_answers.append(formatted_answer)
else:
final_answers.append(str(ans))
# Final validation of the answers
for i, answer in enumerate(final_answers):
if not answer or len(answer.split()) < 50: # Basic length check
raise ValueError(f"Answer {i+1} is too short or empty")
# Save the synoptic to the synoptic_store collection
synoptic_data = {
"session_title": session_title,
"questions": questions,
"synoptic": final_answers,
"created_at": datetime.utcnow()
}
synoptic_store_collection.insert_one(synoptic_data)
return final_answers
except Exception as e:
# Log the error for debugging
print(f"Error in generate_synoptic: {str(e)}")
print(f"Response text: {response.text if 'response' in locals() else 'No response generated'}")
return None
def save_subjective_test(course_id, session_id, title, questions):
"""Save subjective test to database with proper ID handling"""
try:
# Ensure proper string format for IDs
course_id = str(course_id)
session_id = str(session_id)
# Format questions
formatted_questions = []
for q in questions:
formatted_question = {
"question": q["question"],
"expected_points": [],
"difficulty_level": "medium",
"suggested_time": "5 minutes"
}
formatted_questions.append(formatted_question)
test_data = {
"course_id": course_id,
"session_id": session_id,
"title": title,
"questions": formatted_questions,
"created_at": datetime.utcnow(),
"status": "active",
"submissions": []
}
result = subjective_tests_collection.insert_one(test_data)
return str(result.inserted_id)
except Exception as e:
print(f"Error saving test: {e}")
return None
def submit_subjective_test(test_id, student_id, answers):
"""Submit test answers with proper ID handling"""
try:
# Ensure IDs are strings
test_id = str(test_id)
student_id = str(student_id)
# Create submission document
submission = {
"student_id": student_id,
"answers": answers,
"submitted_at": datetime.utcnow(),
"status": "submitted"
}
# Update test document with new submission
result = subjective_tests_collection.update_one(
{"_id": ObjectId(test_id)},
{"$push": {"submissions": submission}}
)
return result.modified_count > 0
except Exception as e:
print(f"Error submitting test: {e}")
return False
def display_subjective_test_tab(student_id, course_id, session_id):
"""Display subjective tests and results for students"""
st.header("Subjective Tests")
try:
subjective_tests = list(subjective_tests_collection.find({
"course_id": course_id,
"session_id": session_id,
"status": "active"
}))
if not subjective_tests:
st.info("No subjective tests available for this session.")
return
# Create tabs for Tests and Results
test_tab, results_tab = st.tabs(["Available Tests", "Test Results"])
with test_tab:
for test in subjective_tests:
with st.expander(f"📝 {test['title']}", expanded=True):
# Check for existing submission
existing_submission = next(
(sub for sub in test.get('submissions', [])
if sub['student_id'] == str(student_id)),
None
)
if existing_submission:
st.success("Test completed! Your answers have been submitted.")
st.subheader("Your Answers")
for i, ans in enumerate(existing_submission['answers']):
st.markdown(f"**Question {i+1}:** {test['questions'][i]['question']}")
st.markdown(f"**Your Answer:** {ans}")
st.markdown("---")
else:
st.write("Please write your answers:")
with st.form(key=f"subjective_test_form_{test['_id']}"):
student_answers = []
for i, question in enumerate(test['questions']):
st.markdown(f"**Question {i+1}:** {question['question']}")
answer = st.text_area(
"Your answer:",
key=f"q_{test['_id']}_{i}",
height=200
)
student_answers.append(answer)
if st.form_submit_button("Submit Test"):
if all(answer.strip() for answer in student_answers):
success = submit_subjective_test(
test['_id'],
str(student_id),
student_answers
)
if success:
st.success("Test submitted successfully!")
st.rerun()
else:
st.error("Error submitting test. Please try again.")
else:
st.error("Please answer all questions before submitting.")
with results_tab:
# Display results for completed tests
completed_tests = [
test for test in subjective_tests
if any(sub['student_id'] == str(student_id) for sub in test.get('submissions', []))
]
if not completed_tests:
st.info("You haven't completed any tests yet.")
return
# Create a selectbox for choosing which test results to view
test_options = {
f"{test['title']} (Submitted: {next(sub['submitted_at'].strftime('%Y-%m-%d') for sub in test['submissions'] if sub['student_id'] == str(student_id))})"
: test['_id']
for test in completed_tests
}
selected_test = st.selectbox(
"Select a test to view results:",
options=list(test_options.keys())
)
if selected_test:
test_id = test_options[selected_test]
display_test_results(test_id, student_id)
except Exception as e:
st.error("An error occurred while loading the tests. Please try again later.")
print(f"Error in display_subjective_test_tab: {str(e)}")
pre_subjective_test_evaluation_collection = db["pre_subjective_test_evaluation"]
pre_subjective_tests_collection = db["pre_subjective_tests"]
def display_pre_subjective_test_tab(student_id, course_id, session_id):
"""Display subjective tests and results for students"""
st.markdown("### Pre-class Subjective Questions ")
try:
print("course_id", course_id, "session_id", session_id, "student_id", student_id)
subjective_tests = list(
pre_subjective_tests_collection.find(
{"course_id": course_id,"session_id" : str(session_id), "status": "active"}
)
)
print("subjective_tests", subjective_tests)
if not subjective_tests:
print("bruh")
st.info("No pre subjective questions available for this session.")
return
# Create tabs for Tests and Results
test_tab, results_tab = st.tabs(["Available Tests", "Test Results"])
with test_tab:
for test in subjective_tests:
with st.expander(f"📝 {test['title']}", expanded=True):
# Check for existing submission
existing_submission = next(
(
sub
for sub in test.get("submissions", [])
if sub["student_id"] == str(student_id)
),
None,
)
if existing_submission:
st.success("Test completed! Your answers have been submitted.")
st.subheader("Your Answers")
for i, ans in enumerate(existing_submission["answers"]):
st.markdown(
f"**Question {i+1}:** {test['questions'][i]['question']}"
)
st.markdown(f"**Your Answer:** {ans}")
st.markdown("---")
else:
st.write("Please write your answers:")
with st.form(key=f"pre_subjective_test_form_{test['_id']}"):
student_answers = []
for i, question in enumerate(test["questions"]):
st.markdown(
f"**Question {i+1}:** {question['question']}"
)
answer = st.text_area(
"Your answer:",
key=f"q_{test['_id']}_{i}",
height=200,
)
student_answers.append(answer)
if st.form_submit_button("Submit Pre Test"):
if all(answer.strip() for answer in student_answers):
success = submit_pre_subjective_test(
test["_id"], str(student_id), student_answers
)
if success:
st.success("Test submitted successfully!")
st.rerun()
else:
st.error(
"Error submitting test. Please try again."
)
else:
st.error(
"Please answer all questions before submitting."
)
with results_tab:
# Display results for completed tests
completed_tests = [
test
for test in subjective_tests
if any(
sub["student_id"] == str(student_id)
for sub in test.get("submissions", [])
)
]
if not completed_tests:
st.info("You haven't completed any tests yet.")
return
# Create a selectbox for choosing which test results to view
test_options = {
f"{test['title']} (Submitted: {next(sub['submitted_at'].strftime('%Y-%m-%d') for sub in test['submissions'] if sub['student_id'] == str(student_id))})": test[
"_id"
]
for test in completed_tests
}
selected_test = st.selectbox(
"Select a test to view results:", options=list(test_options.keys())
)
if selected_test:
test_id = test_options[selected_test]
display_pre_test_results(test_id, student_id)
except Exception as e:
st.error("An error occurred while loading the tests. Please try again later.")
print(f"Error in display_subjective_test_tab: {str(e)}")
def display_test_results(test_id, student_id):
"""
Display test results and analysis for a student
Args:
test_id: ObjectId or str of the test
student_id: str of the student ID
"""
try:
# Fetch analysis from evaluation collection
analysis = subjective_test_evaluation_collection.find_one({
"test_id": test_id,
"student_id": str(student_id)
})
if not analysis:
st.info("Analysis will be available soon. Please check back later.")
return
st.header("Test Analysis")
# Display overall evaluation summary if available
if "overall_summary" in analysis:
with st.expander("Overall Performance Summary", expanded=True):
st.markdown(analysis["overall_summary"])
# Display individual question evaluations
st.subheader("Question-wise Analysis")
for eval_item in analysis.get('evaluations', []):
with st.expander(f"Question {eval_item['question_number']}", expanded=True):
st.markdown("**Question:**")
st.markdown(eval_item['question'])
st.markdown("**Your Answer:**")
st.markdown(eval_item['answer'])
st.markdown("**Evaluation:**")
st.markdown(eval_item['evaluation'])
# Extract and display score if available
if "Score:" in eval_item['evaluation']:
score_line = next((line for line in eval_item['evaluation'].split('\n') if "Score:" in line), None)
if score_line:
score = score_line.split("Score:")[1].strip()
st.metric("Score", score)
# Display improvement points if available
if "Key Areas for Improvement" in eval_item['evaluation']:
st.markdown("**Areas for Improvement:**")
improvement_section = eval_item['evaluation'].split("Key Areas for Improvement")[1]
points = [point.strip('- ').strip() for point in improvement_section.split('\n') if point.strip().startswith('-')]
for point in points:
if point: # Only display non-empty points
st.markdown(f"• {point}")
# Display evaluation timestamp
if "evaluated_at" in analysis:
st.caption(f"Analysis generated on: {analysis['evaluated_at'].strftime('%Y-%m-%d %H:%M:%S UTC')}")
except Exception as e:
st.error("An error occurred while loading the analysis. Please try again later.")
print(f"Error in display_test_results: {str(e)}")