Add automated training scheduler - checks every 2 minutes and trains when 10+ feedback samples available

app.py

@@ -7,7 +7,10 @@ from flask import Flask, request, jsonify
 from inference_core import run_pipeline_for_image, download_image_from_url, upload_to_cloudinary, model, device
 from scripts.feedback_learning_pipeline import initialize_feedback_pipeline, run_feedback_training
 from scripts.model_versioning import initialize_model_tracker
+from apscheduler.schedulers.background import BackgroundScheduler
+from datetime import datetime
 import os
+import atexit
 
 app = Flask(__name__)
 
@@ -18,12 +21,67 @@ feedback_pipeline = initialize_feedback_pipeline(model, device)
 model_tracker = initialize_model_tracker()
 
 
+# ===== Automated Training Scheduler =====
+def automated_training_check():
+    """
+    Background task that checks for new feedback and triggers training automatically
+    Runs periodically to keep the model updated with user corrections
+    """
+    try:
+        print(f"\n[Automated Training] Running scheduled check at {datetime.now()}")
+        
+        # Check if there's enough feedback to warrant training
+        stats = feedback_pipeline.get_feedback_stats()
+        
+        if stats.get("ready_for_retraining", False):
+            unprocessed = stats.get("total_feedback_in_db", 0) - stats.get("total_processed", 0)
+            print(f"[Automated Training] Found {unprocessed} unprocessed feedback samples")
+            print(f"[Automated Training] Starting training cycle...")
+            
+            # Trigger training
+            results = run_feedback_training(feedback_pipeline)
+            
+            if results.get("status") == "success":
+                print(f"[Automated Training] ✓ Training completed successfully")
+                print(f"[Automated Training] Processed {results.get('corrections_processed')} corrections")
+            else:
+                print(f"[Automated Training] Training status: {results.get('status')}")
+        else:
+            unprocessed = stats.get("total_feedback_in_db", 0) - stats.get("total_processed", 0)
+            print(f"[Automated Training] Not enough feedback for training ({unprocessed} new samples)")
+            
+    except Exception as e:
+        print(f"[Automated Training] Error during automated check: {e}")
+
+
+# Initialize background scheduler
+scheduler = BackgroundScheduler(daemon=True)
+
+# Schedule training checks every 2 minutes
+# You can adjust the interval: hours, minutes, seconds
+scheduler.add_job(
+    func=automated_training_check,
+    trigger="interval",
+    minutes=2,  # Check every 2 minutes
+    id='automated_training',
+    name='Automated Feedback Training Check',
+    replace_existing=True
+)
+
+# Start the scheduler
+scheduler.start()
+print("[Automated Training] Scheduler started - checking for new feedback every 2 minutes")
+
+# Shutdown scheduler gracefully when app exits
+atexit.register(lambda: scheduler.shutdown())
+
+
 @app.route("/", methods=["GET"])
 def home():
     """Home page with API documentation"""
     return jsonify({
         "service": "Anomaly Detection API with Feedback Learning",
-        "version": "2.0",
+        "version": "2.1",
         "endpoints": {
             "/health": "GET - Health check",
             "/infer": "POST - Run inference on image URL",
@@ -43,7 +101,8 @@ def home():
         },
         "feedback_info": {
             "description": "User corrections are automatically fetched from Supabase",
-            "training_trigger": "Automatic when 10+ new feedback samples available",
+            "automated_training": "Checks for new feedback every 2 minutes and trains automatically",
+            "training_threshold": "10+ new feedback samples triggers training",
             "manual_training": "POST /feedback/train to trigger immediately"
         },
         "versioning_info": {

requirements.txt

@@ -33,3 +33,6 @@ requests
 # Supabase for feedback storage
 supabase
 
+# Background task scheduling
+APScheduler
+