back to main

app.py

@@ -633,241 +633,38 @@ def align_and_grade_pipeline(qp_path, ms_path, ans_path, imprint=False):
         print("❌ Pipeline error:", e)
         return f"❌ Error: {e}", None, None, None, None
 
-# ---------------- MASS GRADING HELPERS ----------------
-def process_single_triplet(triplet_data, triplet_index):
-    """Process a single QP+MS+AS triplet and return results."""
-    try:
-        qp_path = triplet_data["qp_file"]
-        ms_path = triplet_data["ms_file"] 
-        ans_path = triplet_data["ans_file"]
-        imprint_flag = triplet_data["imprint"]
-        
-        print(f"🔄 Processing triplet {triplet_index + 1}: {os.path.basename(ans_path)}")
-        
-        qpms_text, as_text, grading_text, grading_pdf_path, imprinted_pdf_path = align_and_grade_pipeline(
-            qp_path, ms_path, ans_path, imprint=imprint_flag
-        )
-        
-        return {
-            "triplet_index": triplet_index,
-            "status": "✅ Complete",
-            "grading_pdf": grading_pdf_path,
-            "imprinted_pdf": imprinted_pdf_path if imprint_flag else None,
-            "error": None
-        }
-    except Exception as e:
-        print(f"❌ Error processing triplet {triplet_index + 1}: {e}")
-        return {
-            "triplet_index": triplet_index,
-            "status": f"❌ Error: {str(e)[:50]}...",
-            "grading_pdf": None,
-            "imprinted_pdf": None,
-            "error": str(e)
-        }
-
-def process_all_triplets(triplets_data):
-    """Process all triplets sequentially and return results."""
-    results = []
-    for i, triplet in enumerate(triplets_data):
-        if triplet["qp_file"] and triplet["ms_file"] and triplet["ans_file"]:
-            result = process_single_triplet(triplet, i)
-            results.append(result)
-            yield results  # Yield intermediate results for progress updates
-    return results
-
-# ---------------- GRADIO UI ----------------
-with gr.Blocks(title="🎓 AI Mass Grading System", theme=gr.themes.Soft()) as demo:
-    gr.Markdown("# 🎓 AI Mass Grading System")
-    gr.Markdown("Upload multiple sets of Question Papers, Markschemes, and Answer Sheets for batch processing.")
-    
-    with gr.Row():
-        with gr.Column():
-            qp_files = gr.File(
-                label="📄 Upload Question Papers (PDF)", 
-                file_count="multiple", 
-                file_types=[".pdf"]
-            )
-        with gr.Column():
-            ms_files = gr.File(
-                label="📋 Upload Markschemes (PDF)", 
-                file_count="multiple", 
-                file_types=[".pdf"]
-            )
-        with gr.Column():
-            ans_files = gr.File(
-                label="📝 Upload Answer Sheets (PDF)", 
-                file_count="multiple", 
-                file_types=[".pdf"]
-            )
-    
-    imprint_all = gr.Checkbox(label="✍ Imprint Marks on All Answer Sheets", value=False)
-    
-    with gr.Row():
-        start_grading_btn = gr.Button("🚀 Start Mass Grading", variant="primary", size="lg")
-    
-    # Status and results
-    status_info = gr.Markdown("", visible=False)
-    status_table = gr.Dataframe(
-        headers=["Set", "QP File", "MS File", "Answer Sheet", "Status"],
-        datatype=["number", "str", "str", "str", "str"],
-        label="📊 Grading Progress",
-        visible=False
-    )
-    
-    results_section = gr.Column(visible=False)
-    with results_section:
-        gr.Markdown("## 📥 Download Results")
-        results_container = gr.Column()
-    
-    def create_triplets_from_files(qp_list, ms_list, ans_list, imprint_setting):
-        """Create triplets from uploaded file lists."""
-        if not qp_list or not ms_list or not ans_list:
-            return []
-        
-        # Convert single files to lists
-        if not isinstance(qp_list, list):
-            qp_list = [qp_list] if qp_list else []
-        if not isinstance(ms_list, list):
-            ms_list = [ms_list] if ms_list else []
-        if not isinstance(ans_list, list):
-            ans_list = [ans_list] if ans_list else []
-        
-        # Create triplets by matching files by index
-        triplets = []
-        max_files = max(len(qp_list), len(ms_list), len(ans_list))
-        
-        for i in range(max_files):
-            qp_file = qp_list[i].name if i < len(qp_list) else None
-            ms_file = ms_list[i].name if i < len(ms_list) else None
-            ans_file = ans_list[i].name if i < len(ans_list) else None
-            
-            # Only create triplet if all three files are present
-            if qp_file and ms_file and ans_file:
-                triplets.append({
-                    "qp_file": qp_file,
-                    "ms_file": ms_file,
-                    "ans_file": ans_file,
-                    "imprint": imprint_setting
-                })
-        
-        return triplets
-    
-    def update_status_display(qp_list, ms_list, ans_list):
-        """Update the status display showing file matching."""
-        if not qp_list and not ms_list and not ans_list:
-            return gr.update(visible=False), gr.update(visible=False, value=[])
-        
-        # Convert single files to lists
-        if not isinstance(qp_list, list):
-            qp_list = [qp_list] if qp_list else []
-        if not isinstance(ms_list, list):
-            ms_list = [ms_list] if ms_list else []
-        if not isinstance(ans_list, list):
-            ans_list = [ans_list] if ans_list else []
-        
-        max_files = max(len(qp_list), len(ms_list), len(ans_list))
-        
-        if max_files == 0:
-            return gr.update(visible=False), gr.update(visible=False, value=[])
-        
-        # Create status info
-        complete_sets = 0
-        table_data = []
-        
-        for i in range(max_files):
-            qp_name = os.path.basename(qp_list[i].name) if i < len(qp_list) else "❌ Missing"
-            ms_name = os.path.basename(ms_list[i].name) if i < len(ms_list) else "❌ Missing"
-            ans_name = os.path.basename(ans_list[i].name) if i < len(ans_list) else "❌ Missing"
-            
-            # Determine status
-            if i < len(qp_list) and i < len(ms_list) and i < len(ans_list):
-                status = "✅ Ready"
-                complete_sets += 1
-            else:
-                status = "⏳ Incomplete"
-            
-            table_data.append([i + 1, qp_name, ms_name, ans_name, status])
-        
-        status_text = f"**📊 File Summary:** {complete_sets} complete set(s) ready for grading out of {max_files} total."
-        
-        return (
-            gr.update(visible=True, value=status_text),
-            gr.update(visible=True, value=table_data)
-        )
-    
-    def process_mass_grading(qp_list, ms_list, ans_list, imprint_setting):
-        """Process all complete triplets."""
-        triplets = create_triplets_from_files(qp_list, ms_list, ans_list, imprint_setting)
-        
-        if not triplets:
-            return (
-                gr.update(value="❌ No complete sets found. Please ensure each set has QP, MS, and AS files."),
-                gr.update(visible=False),
-                gr.update(visible=False)
-            )
-        
-        # Update status to show processing
-        table_data = []
-        for i, triplet in enumerate(triplets):
-            table_data.append([
-                i + 1,
-                os.path.basename(triplet["qp_file"]),
-                os.path.basename(triplet["ms_file"]),
-                os.path.basename(triplet["ans_file"]),
-                "🔄 Processing..."
-            ])
-        
-        # Process each triplet
-        results = []
-        result_components = []
-        
-        for i, triplet in enumerate(triplets):
-            print(f"\n🔄 Processing Set {i + 1}/{len(triplets)}")
-            
-            try:
-                result = process_single_triplet(triplet, i)
-                results.append(result)
-                
-                # Update table status
-                table_data[i][4] = result["status"]
-                
-                # Create download components
-                if result["grading_pdf"]:
-                    result_components.append(
-                        gr.File(value=result["grading_pdf"], label=f"📄 Set {i+1} - Grading PDF", visible=True)
-                    )
-                if result["imprinted_pdf"]:
-                    result_components.append(
-                        gr.File(value=result["imprinted_pdf"], label=f"✍️ Set {i+1} - Imprinted PDF", visible=True)
-                    )
-                    
-            except Exception as e:
-                print(f"❌ Error processing set {i + 1}: {e}")
-                table_data[i][4] = f"❌ Error: {str(e)[:30]}..."
-        
-        # Final status
-        successful = len([r for r in results if r.get("status") == "✅ Complete"])
-        status_text = f"**🏁 Processing Complete:** {successful}/{len(triplets)} sets processed successfully."
-        
-        return (
-            gr.update(value=status_text),
-            gr.update(value=table_data),
-            gr.update(visible=True)
-        )
-    
-    # Event handlers
-    for file_input in [qp_files, ms_files, ans_files]:
-        file_input.change(
-            fn=update_status_display,
-            inputs=[qp_files, ms_files, ans_files],
-            outputs=[status_info, status_table]
-        )
-    
-    start_grading_btn.click(
-        fn=process_mass_grading,
-        inputs=[qp_files, ms_files, ans_files, imprint_all],
-        outputs=[status_info, status_table, results_section]
-    )
+# ---------------- GRADIO UI ---------------- 
+with gr.Blocks(title=" AI Grading (Final Flow )") as demo: 
+    gr.Markdown("## 📘  AI Grading — Final Flow") 
+
+    with gr.Row(): 
+        qp_file = gr.File(label="📄 Upload Question Paper (PDF)") 
+        ms_file = gr.File(label="📄 Upload Markscheme (PDF)") 
+        ans_file = gr.File(label="📝 Upload Student Answer Sheet (PDF)") 
+
+    imprint_toggle = gr.Checkbox(label="✍ Imprint Marks on Student Answer Sheet", value=False) 
+    run_button = gr.Button("🚀 Run Pipeline") 
+
+    with gr.Row(): 
+        grading_pdf_file = gr.File(label="📥 Download Grading PDF") 
+        imprint_pdf_file = gr.File(label="📥 Download Imprinted PDF (Optional)") 
+
+    def run_pipeline(qp_file_obj, ms_file_obj, ans_file_obj, imprint_flag): 
+        qp_path = qp_file_obj.name 
+        ms_path = ms_file_obj.name 
+        ans_path = ans_file_obj.name 
+
+        qpms_text, as_text, grading_text, grading_pdf_path, imprinted_pdf_path = align_and_grade_pipeline( 
+            qp_path, ms_path, ans_path, imprint=imprint_flag 
+        ) 
+
+        return grading_pdf_path, imprinted_pdf_path 
+
+    run_button.click( 
+        fn=run_pipeline, 
+        inputs=[qp_file, ms_file, ans_file, imprint_toggle], 
+        outputs=[grading_pdf_file, imprint_pdf_file] 
+    ) 
 
-if __name__ == "__main__":
+if __name__ == "__main__": 
     demo.launch()