Upload functionbloom.py

functionbloom.py

@@ -0,0 +1,388 @@
+from typing import Optional, Dict
+import streamlit as st
+import requests
+import json
+import fitz  # PyMuPDF
+from fpdf import FPDF
+import os
+import tempfile
+from dotenv import load_dotenv
+import torch
+from transformers import DistilBertForSequenceClassification, DistilBertTokenizer
+from torch.nn.functional import softmax
+from doctr.models import ocr_predictor
+from doctr.io import DocumentFile
+import tempfile
+
+load_dotenv()
+
+model = DistilBertForSequenceClassification.from_pretrained('./fine_tuned_distilbert')
+tokenizer = DistilBertTokenizer.from_pretrained('./fine_tuned_distilbert')
+device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
+model.to(device)
+mapping = {"Remembering": 0, "Understanding": 1, "Applying": 2, "Analyzing": 3, "Evaluating": 4, "Creating": 5}
+reverse_mapping = {v: k for k, v in mapping.items()}
+modelocr = ocr_predictor(det_arch='db_resnet50', reco_arch='crnn_vgg16_bn', pretrained=True)
+
+def save_uploaded_file(uploaded_file):
+    if uploaded_file is not None:
+        file_extension = uploaded_file.name.split('.')[-1].lower()
+        temp_file = tempfile.NamedTemporaryFile(delete=False, suffix = f'.{file_extension}')
+        temp_file.write(uploaded_file.getvalue())
+        temp_file.close()
+        return temp_file.name
+    return None
+
+# Previous functions from Question Generator
+def get_pdf_path(pdf_source=None, uploaded_file=None):
+    try:
+        # If a file is uploaded locally
+        if uploaded_file is not None:
+            # Create a temporary file to save the uploaded PDF
+            temp_dir = tempfile.mkdtemp()
+            pdf_path = os.path.join(temp_dir, uploaded_file.name)
+            
+            # Save the uploaded file
+            with open(pdf_path, "wb") as pdf_file:
+                pdf_file.write(uploaded_file.getvalue())
+            return pdf_path
+
+        # If a URL is provided
+        if pdf_source:
+            response = requests.get(pdf_source, timeout=30)
+            response.raise_for_status()
+            
+            # Create a temporary file
+            temp_dir = tempfile.mkdtemp()
+            pdf_path = os.path.join(temp_dir, "downloaded.pdf")
+            
+            with open(pdf_path, "wb") as pdf_file:
+                pdf_file.write(response.content)
+            return pdf_path
+
+        # If no source is provided
+        st.error("No PDF source provided.")
+        return None
+    except Exception as e:
+        st.error(f"Error getting PDF: {e}")
+        return None
+    
+    
+def extract_text_pymupdf(pdf_path):
+    try:
+        doc = fitz.open(pdf_path)
+        pages_content = []
+        for page_num in range(len(doc)):
+            page = doc[page_num]
+            pages_content.append(page.get_text())
+        doc.close()
+        return " ".join(pages_content)  # Join all pages into one large context string
+    except Exception as e:
+        st.error(f"Error extracting text from PDF: {e}")
+        return ""
+    
+
+def get_bloom_taxonomy_scores(question: str) -> Dict[str, float]:
+    # Default scores in case of API failure
+    default_scores = {
+        "Remembering": 0.2,
+        "Understanding": 0.2,
+        "Applying": 0.15,
+        "Analyzing": 0.15,
+        "Evaluating": 0.15,
+        "Creating": 0.15
+    }
+    
+    try:
+        scores = predict_with_loaded_model(question)
+        for key, value in scores.items():
+            if not (0 <= value <= 1):
+                st.warning(f"Invalid score value for {key}. Using default scores.")
+                return default_scores           
+        return scores
+
+    except Exception as e:
+        st.warning(f"Unexpected error: {e}. Using default scores.")
+        return default_scores
+    
+    
+def generate_ai_response(api_key, assistant_context, user_query, role_description, response_instructions, bloom_taxonomy_weights, num_questions, question_length, include_numericals):
+    try:
+        url = f"https://generativelanguage.googleapis.com/v1beta/models/gemini-1.5-flash-latest:generateContent?key={api_key}"
+        
+        # Define length guidelines
+        length_guidelines = {
+            "Short": "Keep questions concise, around 10-15 words each.",
+            "Medium": "Create moderately detailed questions, around 20-25 words each.",
+            "Long": "Generate detailed, comprehensive questions, around 30-40 words each that may include multiple parts."
+        }
+        
+        prompt = f"""
+        You are a highly knowledgeable assistant. Your task is to assist the user with the following context from an academic paper.
+
+        **Role**: {role_description}
+
+        **Context**: {assistant_context}
+
+        **Instructions**: {response_instructions}
+        Question Length Requirement: {length_guidelines[question_length]}
+
+        **Bloom's Taxonomy Weights**:
+        Knowledge: {bloom_taxonomy_weights['Knowledge']}%
+        Comprehension: {bloom_taxonomy_weights['Comprehension']}%
+        Application: {bloom_taxonomy_weights['Application']}%
+        Analysis: {bloom_taxonomy_weights['Analysis']}%
+        Synthesis: {bloom_taxonomy_weights['Synthesis']}%
+        Evaluation: {bloom_taxonomy_weights['Evaluation']}%
+
+        **Query**: {user_query}
+
+        **Number of Questions**: {num_questions}
+
+        **Include Numericals**: {include_numericals}
+        """
+        
+        payload = {
+            "contents": [
+                {
+                    "parts": [
+                        {"text": prompt}
+                    ]
+                }
+            ]
+        }
+        headers = {"Content-Type": "application/json"}
+
+        response = requests.post(url, headers=headers, data=json.dumps(payload), timeout=60)
+        response.raise_for_status()
+        
+        result = response.json()
+        questions = result.get("candidates", [{}])[0].get("content", {}).get("parts", [{}])[0].get("text", "")
+        questions_list = [question.strip() for question in questions.split("\n") if question.strip()]
+        
+        # Get Bloom's taxonomy scores for each question with progress bar
+        questions_with_scores = []
+        progress_bar = st.progress(0)
+        for idx, question in enumerate(questions_list):
+            scores = get_bloom_taxonomy_scores(question)
+            if scores:  # Only add questions that got valid scores
+                questions_with_scores.append((question, scores))
+            progress_bar.progress((idx + 1) / len(questions_list))
+        
+        if not questions_with_scores:
+            st.warning("Could not get Bloom's Taxonomy scores for any questions. Using default scores.")
+            # Use default scores if no scores were obtained
+            questions_with_scores = [(q, get_bloom_taxonomy_scores("")) for q in questions_list]
+        
+        # Update session state with scores
+        st.session_state.question_scores = {q: s for q, s in questions_with_scores}
+        
+        # Return just the questions
+        return [q for q, _ in questions_with_scores]
+    except requests.RequestException as e:
+        st.error(f"API request error: {e}")
+        return []
+    except Exception as e:
+        st.error(f"Error generating questions: {e}")
+        return []
+
+def normalize_bloom_weights(bloom_weights):
+    total = sum(bloom_weights.values())
+    if total != 100:
+        normalization_factor = 100 / total
+        # Normalize each weight by multiplying it by the normalization factor
+        bloom_weights = {key: round(value * normalization_factor, 2) for key, value in bloom_weights.items()}
+    return bloom_weights
+
+def generate_pdf(questions, filename="questions.pdf"):
+    try:
+        pdf = FPDF()
+        pdf.set_auto_page_break(auto=True, margin=15)
+        pdf.add_page()
+
+        # Set font
+        pdf.set_font("Arial", size=12)
+        
+        # Add a title or heading
+        pdf.cell(200, 10, txt="Generated Questions", ln=True, align="C")
+
+        # Add space between title and questions
+        pdf.ln(10)
+
+        # Loop through questions and add them to the PDF
+        for i, question in enumerate(questions, 1):
+            # Using multi_cell for wrapping the text in case it's too long
+            pdf.multi_cell(0, 10, f"Q{i}: {question}")
+
+        # Save the generated PDF to the file
+        pdf.output(filename)
+        return filename
+    except Exception as e:
+        st.error(f"Error generating PDF: {e}")
+        return None
+
+def process_pdf_and_generate_questions(pdf_source, uploaded_file, api_key, role_description, response_instructions, bloom_taxonomy_weights, num_questions, question_length, include_numericals):
+    try:
+
+        pdf_path = get_pdf_path(pdf_source, uploaded_file)
+        if not pdf_path:
+            return []
+
+        # Extract text
+        pdf_text = extract_text_pymupdf(pdf_path)
+        if not pdf_text:
+            return []
+        # Generate questions
+        assistant_context = pdf_text
+        user_query = "Generate questions based on the above context."
+        normalized_bloom_weights = normalize_bloom_weights(bloom_taxonomy_weights)
+        questions = generate_ai_response(
+            api_key, 
+            assistant_context, 
+            user_query, 
+            role_description, 
+            response_instructions, 
+            normalized_bloom_weights, 
+            num_questions,
+            question_length,
+            include_numericals
+        )
+
+        # Clean up temporary PDF file
+        try:
+            os.remove(pdf_path)
+            # Remove the temporary directory
+            os.rmdir(os.path.dirname(pdf_path))
+        except Exception as e:
+            st.warning(f"Could not delete temporary PDF file: {e}")
+
+        return questions
+    except Exception as e:
+        st.error(f"Error processing PDF and generating questions: {e}")
+        return []
+    
+def get_bloom_taxonomy_details(question_scores: Optional[Dict[str, float]] = None) -> str:
+    """
+    Generate a detailed explanation of Bloom's Taxonomy scores.
+    Handles missing or invalid scores gracefully.
+    """
+    try:
+        if question_scores is None or not isinstance(question_scores, dict):
+            return "Bloom's Taxonomy scores not available"
+        
+        # Validate scores
+        valid_categories = {"Remembering", "Understanding", "Applying", 
+                          "Analyzing", "Evaluating", "Creating"}
+        
+        if not all(isinstance(score, (int, float)) for score in question_scores.values()):
+            return "Invalid score values detected"
+            
+        if not all(category in valid_categories for category in question_scores.keys()):
+            return "Invalid score categories detected"
+        
+        details_text = "Bloom's Taxonomy Analysis:\n\n"
+        
+        try:
+            # Sort scores by value in descending order
+            sorted_scores = sorted(question_scores.items(), key=lambda x: x[1], reverse=True)
+            
+            # Format each score as a percentage
+            for category, score in sorted_scores:
+                percentage = min(max(score * 100, 0), 100)  # Ensure percentage is between 0 and 100
+                details_text += f"{category}: {percentage:.1f}%\n"
+            
+            # Add the predicted level
+            predicted_level = max(question_scores.items(), key=lambda x: x[1])[0]
+            details_text += f"\nPredicted Level: {predicted_level}"
+            
+            return details_text.strip()
+            
+        except Exception as e:
+            return f"Error processing scores: {str(e)}"
+            
+    except Exception as e:
+        return f"Error generating taxonomy details: {str(e)}"
+    
+       
+def predict_with_loaded_model(text):
+    inputs = tokenizer(text, return_tensors='pt', padding=True, truncation=True, max_length=512)
+    input_ids = inputs['input_ids'].to(device)
+    model.eval()
+    with torch.no_grad():
+        outputs = model(input_ids)
+        logits = outputs.logits
+        probabilities = softmax(logits, dim=-1)
+    probabilities = probabilities.squeeze().cpu().numpy()
+    # Convert to float and format to 3 decimal places
+    class_probabilities = {reverse_mapping[i]: float(f"{prob:.3f}") for i, prob in enumerate(probabilities)}
+    return class_probabilities
+
+def process_document(input_path):
+    if input_path.lower().endswith(".pdf"):
+        doc = DocumentFile.from_pdf(input_path)
+        #print(f"Number of pages: {len(doc)}")
+    elif input_path.lower().endswith((".jpg", ".jpeg", ".png", ".bmp", ".tiff")):
+        doc = DocumentFile.from_images(input_path)
+    else:
+        raise ValueError("Unsupported file type. Please provide a PDF or an image file.")
+    result = modelocr(doc)
+    def calculate_average_confidence(result):
+        total_confidence = 0
+        word_count = 0
+        for page in result.pages:
+            for block in page.blocks:
+                for line in block.lines:
+                    for word in line.words:
+                        total_confidence += word.confidence
+                        word_count += 1
+        average_confidence = total_confidence / word_count if word_count > 0 else 0
+        return average_confidence
+    average_confidence = calculate_average_confidence(result)
+    string_result = result.render()
+    return {'Avg_Confidence': average_confidence, 'String':string_result.split('\n')}
+
+def sendtogemini(inputpath, question):
+    if inputpath and inputpath.lower().endswith((".pdf", ".jpg", ".jpeg", ".png")):
+        qw = process_document(inputpath)
+    elif question:
+        qw = {'String': [question]}
+    else:
+        raise ValueError("Unsupported file type. Please provide a PDF or an image file.")
+    questionset = str(qw['String'])
+    # send this prompt to gemini : 
+    questionset += """You are given a list of text fragments containing questions fragments extracted by an ocr model. Your task is to:
+    # only Merge the question fragments into complete and coherent questions.Don't answer then.
+    # Separate each question , start a new question with @ to make them easily distinguishable for further processing."""
+    url = f"https://generativelanguage.googleapis.com/v1beta/models/gemini-1.5-flash-latest:generateContent?key={os.getenv('GEMINI_API_KEY')}"
+    
+    payload = {
+            "contents": [
+                {
+                    "parts": [
+                        {"text": questionset}
+                    ]
+                }
+            ]
+        }
+    headers = {"Content-Type": "application/json"}
+
+    response = requests.post(url, headers=headers, data=json.dumps(payload), timeout=60)
+    result = response.json()
+    res1 = result.get("candidates", [{}])[0].get("content", {}).get("parts", [{}])[0].get("text", "")
+    question = []
+    for i in res1.split('\n'):
+        i = i.strip()
+        if len(i) > 0:
+            if i[0] == '@':
+                i = i[1:].strip().lower()
+                if i[0] == 'q':
+                    question.append(i[1:].strip())
+                else:
+                    question.append(i)
+    data = []
+    for i in question:
+        d = {}
+        d['question'] = i
+        d['score'] = predict_with_loaded_model(i)
+        data.append(d)
+    return data