Upload 2 files

rubrics.py

@@ -0,0 +1,112 @@
+import streamlit as st
+from pymongo import MongoClient
+from openai import OpenAI
+from bson import ObjectId
+import json
+from dotenv import load_dotenv
+import os
+
+load_dotenv()
+MONGO_URI = os.getenv('MONGO_URI')
+OPENAI_API_KEY = os.getenv('OPENAI_KEY')
+
+client = MongoClient(MONGO_URI)
+db = client['novascholar_db']
+# db.create_collection("rubrics")
+rubrics_collection = db['rubrics']
+resources_collection = db['resources']
+courses_collection = db['courses']
+
+def generate_rubrics(api_key, session_title, outcome_description, taxonomy, pre_class_material):
+    prompt = f"""
+    You are an expert educational AI assistant specializing in instructional design. Generate a detailed rubric for the session titled "{session_title}". The rubric should be aligned with Bloom's Taxonomy level "{taxonomy}" and use numerical scoring levels (4,3,2,1) instead of descriptive levels. Use the following context:
+
+    Session Outcome Description:
+    {outcome_description}
+
+    Pre-class Material:
+    {pre_class_material}
+
+    Please generate the rubric in JSON format with these specifications:
+    1. Use numerical levels (4=Highest, 1=Lowest) instead of descriptive levels
+    2. Include 4-5 relevant criteria based on the session outcome
+    3. Each criterion should have clear descriptors for each numerical level
+    4. Focus on objectively measurable aspects for evaluation
+    5. Structure should be suitable for evaluating assignments and test answers
+
+    ***IMPORTANT: DO NOT INCLUDE THE WORD JSON IN THE OUTPUT STRING, DO NOT INCLUDE BACKTICKS (```) IN THE OUTPUT, AND DO NOT INCLUDE ANY OTHER TEXT, OTHER THAN THE ACTUAL JSON RESPONSE. START THE RESPONSE STRING WITH AN OPEN CURLY BRACE {{ AND END WITH A CLOSING CURLY BRACE }}.***
+    """
+
+    messages = [
+        {
+            "role": "system",
+            "content": "You are an expert educational AI assistant specializing in instructional design.",
+        },
+        {
+            "role": "user",
+            "content": prompt
+        },
+    ]
+
+    try:
+        client = OpenAI(api_key=api_key)
+        response = client.chat.completions.create(
+            model="gpt-4-0125-preview",
+            messages=messages
+        )
+        return response.choices[0].message.content
+    except Exception as e:
+        st.error(f"Failed to generate rubrics: {e}")
+        return None
+
+def display_rubrics_tab(session, course_id):
+    st.subheader("Generated Rubrics")
+
+    # Fetch session details from the courses collection
+    course_data = courses_collection.find_one(
+        {"course_id": course_id, "sessions.session_id": session['session_id']},
+        {"sessions.$": 1}
+    )
+    
+    if course_data and 'sessions' in course_data and len(course_data['sessions']) > 0:
+        session_data = course_data['sessions'][0]
+        
+        # Extract session learning outcomes
+        if 'session_learning_outcomes' in session_data and len(session_data['session_learning_outcomes']) > 0:
+            outcome = session_data['session_learning_outcomes'][0]
+            outcome_description = outcome.get('outcome_description', '')
+            taxonomy_level = outcome.get('bloom_taxonomy_level', '')
+            
+            # Display fetched information
+            st.markdown("### Session Information")
+            st.markdown(f"**Session Title:** {session['title']}")
+            st.markdown(f"**Learning Outcome:** {outcome_description}")
+            st.markdown(f"**Taxonomy Level:** {taxonomy_level}")
+
+            # Fetch pre-class material
+            pre_class_material_docs = resources_collection.find({"session_id": session['session_id']})
+            pre_class_material = "\n".join([f"{doc.get('title', 'No Title')}: {doc.get('url', 'No URL')}" for doc in pre_class_material_docs])
+
+            if st.button("Generate Rubric"):
+                rubric = generate_rubrics(
+                    OPENAI_API_KEY,
+                    session['title'],
+                    outcome_description,
+                    taxonomy_level,
+                    pre_class_material
+                )
+                
+                if rubric:
+                    st.json(rubric)
+                    if st.button("Save Rubric"):
+                        rubric_data = {
+                            "course_id": course_id,
+                            "session_id": session['session_id'],
+                            "rubric": json.loads(rubric)
+                        }
+                        rubrics_collection.insert_one(rubric_data)
+                        st.success("Rubric saved successfully!")
+        else:
+            st.error("No learning outcomes found for this session")
+    else:
+        st.error("Session data not found")

subjective_test_evaluation.py

@@ -0,0 +1,252 @@
+import openai
+from pymongo import MongoClient
+from datetime import datetime
+import os
+from dotenv import load_dotenv
+import re
+import streamlit as st
+from bson import ObjectId
+
+load_dotenv()
+MONGO_URI = os.getenv('MONGO_URI')
+OPENAI_API_KEY = os.getenv('OPENAI_KEY')
+
+client = MongoClient(MONGO_URI)
+db = client['novascholar_db']
+rubrics_collection = db['rubrics']
+resources_collection = db['resources']
+subjective_tests_collection = db['subjective_tests']
+subjective_test_analysis_collection = db['subjective_test_analysis']
+
+openai.api_key = OPENAI_API_KEY
+
+def evaluate_subjective_answers(test_id, student_id, course_id):
+    """Evaluate subjective test answers using OpenAI."""
+    try:
+        # Get test and submission details
+        test_doc = subjective_tests_collection.find_one({
+            "_id": ObjectId(test_id),
+            "course_id": course_id
+        })
+        if not test_doc:
+            return {
+                "content_analysis": "Error: Test not found",
+                "analyzed_at": datetime.utcnow(),
+                "correctness_score": 0
+            }
+
+        submission = next(
+            (sub for sub in test_doc.get('submissions', []) if sub['student_id'] == student_id),
+            None
+        )
+
+        if not submission:
+            return {
+                "content_analysis": "Error: Submission not found",
+                "analyzed_at": datetime.utcnow(),
+                "correctness_score": 0
+            }
+
+        # Rest of the evaluation logic remains the same
+        questions = test_doc.get('questions', [])
+        student_answers = submission.get('answers', [])
+
+        if not questions or not student_answers:
+            return {
+                "content_analysis": "Error: No questions or answers found",
+                "analyzed_at": datetime.utcnow(),
+                "correctness_score": 0
+            }
+
+        # Retrieve rubrics for the session
+        rubric_doc = rubrics_collection.find_one({
+            "session_id": test_doc['session_id'],
+            "course_id": course_id
+        })
+        
+        if not rubric_doc:
+            return {
+                "content_analysis": "Error: Rubric not found",
+                "analyzed_at": datetime.utcnow(),
+                "correctness_score": 0
+            }
+
+        rubric = rubric_doc.get('rubric', {})
+
+        # Retrieve pre-class materials
+        pre_class_materials = resources_collection.find({
+            "session_id": test_doc['session_id'],
+            "course_id": course_id
+        })
+        pre_class_content = "\n".join([material.get('text_content', '') for material in pre_class_materials])
+
+        # Analyze each question
+        all_analyses = []
+        total_score = 0
+
+        for i, (question, answer) in enumerate(zip(questions, student_answers), 1):
+            analysis_content = f"Question {i}: {question['question']}\nAnswer: {answer}\n\nRubric: {rubric}\n\nPre-class Materials: {pre_class_content}\n\n"
+
+            prompt_template = f"""As an educational assessor, evaluate this student's answer based on the provided rubric criteria and pre-class materials. Follow these assessment guidelines:
+
+                1. Evaluation Process:
+                - Use each rubric criterion (scored 1-4) for internal assessment
+                - Compare response with pre-class materials
+                - Check alignment with all rubric requirements
+                - Calculate final score: sum of criteria scores converted to 10-point scale
+
+                Pre-class Materials:
+                {pre_class_content}
+
+                Rubric Criteria:
+                {rubric}
+
+                Question and Answer:
+                {analysis_content}
+
+                Provide your assessment in the following format:
+
+                **Score and Evidence**
+                - Score: [X]/10
+                - Evidence for deduction: [One-line reference to most significant gap or inaccuracy]
+                **Key Areas for Improvement**
+                - [Concise improvement point 1]
+                - [Concise improvement point 2]
+                - [Concise improvement point 3]
+            """
+
+            response = openai.Completion.create(
+                model="text-davinci-003",
+                prompt=prompt_template,
+                max_tokens=500,
+                temperature=0.7
+            )
+
+            individual_analysis = response.choices[0].text.strip()
+
+            try:
+                score_match = re.search(r'Score: (\d+)', individual_analysis)
+                question_score = int(score_match.group(1)) if score_match else 0
+                total_score += question_score
+            except:
+                question_score = 0
+
+            formatted_analysis = f"\n\n## Question {i} Analysis\n\n{individual_analysis}"
+            all_analyses.append(formatted_analysis)
+
+        average_score = round(total_score / len(questions)) if questions else 0
+        combined_analysis = "\n".join(all_analyses)
+
+        return {
+            "content_analysis": combined_analysis,
+            "analyzed_at": datetime.utcnow(),
+            "correctness_score": average_score
+        }
+
+    except Exception as e:
+        return {
+            "content_analysis": f"Error evaluating answers: {str(e)}",
+            "analyzed_at": datetime.utcnow(),
+            "correctness_score": 0
+        }
+
+def display_evaluation_to_faculty(session_id, student_id, course_id):
+    """Display submitted tests with improved error handling and debugging"""
+    st.subheader("Evaluate Subjective Tests")
+    
+    try:
+        # Convert all IDs to strings for consistent comparison
+        session_id = str(session_id)
+        student_id = str(student_id)
+        course_id = str(course_id)
+        
+        print(f"Searching for tests with session_id: {session_id}, student_id: {student_id}, course_id: {course_id}")
+        
+        # Query for tests
+        query = {
+            "session_id": session_id,
+            "course_id": course_id,
+            "submissions": {
+                "$elemMatch": {
+                    "student_id": student_id
+                }
+            }
+        }
+        
+        # Log the query for debugging
+        print(f"MongoDB Query: {query}")
+        
+        # Fetch tests
+        tests = list(subjective_tests_collection.find(query))
+        print(f"Found {len(tests)} tests matching query")
+        
+        if not tests:
+            # Check if any tests exist for this session
+            all_session_tests = list(subjective_tests_collection.find({
+                "session_id": session_id,
+                "course_id": course_id
+            }))
+            
+            if all_session_tests:
+                print(f"Found {len(all_session_tests)} tests for this session, but no submissions from student {student_id}")
+                st.warning("No submitted tests found for this student, but tests exist for this session.")
+            else:
+                print("No tests found for this session at all")
+                st.info("No tests have been created for this session yet.")
+            return
+
+        # Display tests and handle evaluation
+        for test in tests:
+            with st.expander(f"Test: {test.get('title', 'Untitled Test')}", expanded=True):
+                # Find student submission
+                submission = next(
+                    (sub for sub in test.get('submissions', []) 
+                     if sub['student_id'] == student_id),
+                    None
+                )
+                
+                if submission:
+                    st.write("### Student's Answers")
+                    for i, (question, answer) in enumerate(zip(test['questions'], submission['answers'])):
+                        st.markdown(f"**Q{i+1}:** {question['question']}")
+                        st.markdown(f"**A{i+1}:** {answer}")
+                        st.markdown("---")
+                    
+                    # Generate/display analysis
+                    if st.button(f"Generate Analysis for {test.get('title')}"):
+                        with st.spinner("Analyzing responses..."):
+                            analysis = evaluate_subjective_answers(
+                                str(test['_id']),
+                                student_id,
+                                course_id
+                            )
+                            
+                            if analysis:
+                                st.markdown("### Analysis")
+                                st.markdown(analysis['content_analysis'])
+                                st.metric("Score", f"{analysis['correctness_score']}/10")
+                else:
+                    st.error("Submission data not found for this student")
+                    
+    except Exception as e:
+        st.error("An error occurred while loading the tests")
+        with st.expander("Error Details"):
+            st.write(f"Error: {str(e)}")
+            st.write(f"Session ID: {session_id}")
+            st.write(f"Student ID: {student_id}")
+            st.write(f"Course ID: {course_id}")
+
+def check_test_submission(session_id, student_id, course_id):
+    """Utility function to check test submission status"""
+    try:
+        query = {
+            "session_id": str(session_id),
+            "course_id": str(course_id),
+            "submissions.student_id": str(student_id)
+        }
+        
+        test = subjective_tests_collection.find_one(query)
+        return bool(test)
+    except Exception as e:
+        print(f"Error checking submission: {e}")
+        return False