import os import json import time import threading import asyncio from fastapi import FastAPI, HTTPException, BackgroundTasks from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import JSONResponse, FileResponse from fastapi.staticfiles import StaticFiles import uvicorn from typing import Dict from pathlib import Path import subprocess from datetime import datetime import torch # Import core functionality from vision_analyzer import ( main_processing_loop, processing_status, log_message, ) # FastAPI App Definition app = FastAPI(title="Video Analysis API", description="API to access video frame analysis results and extracted images", version="1.0.0") # Add CORS middleware to allow cross-origin requests app.add_middleware( CORSMiddleware, allow_origins=["*"], # Allows all origins allow_credentials=True, allow_methods=["*"], # Allows all methods allow_headers=["*"], ) # Global variables for processing and frame tracking processing_thread = None frame_locks = {} # Dict to track frame locks: {course: {frame: {"locked_by": id, "locked_at": timestamp}}} processed_frames = {} # Dict to track processed frames: {course: {frame: {"processed_by": id, "processed_at": timestamp}}} LOCK_TIMEOUT = 300 # 5 minutes timeout for locks TRACKING_FILE = os.path.join(os.path.dirname(os.path.abspath(__file__)), "frame_tracking.json") def save_tracking_state(): """Save frame tracking state to disk""" state = { "frame_locks": frame_locks, "processed_frames": processed_frames } try: with open(TRACKING_FILE, "w") as f: json.dump(state, f, indent=2) except Exception as e: log_message(f"Error saving tracking state: {e}") def load_tracking_state(): """Load frame tracking state from disk""" global frame_locks, processed_frames try: with open(TRACKING_FILE, "r") as f: state = json.load(f) frame_locks = state.get("frame_locks", {}) processed_frames = state.get("processed_frames", {}) except FileNotFoundError: log_message("No previous tracking state found") except Exception as e: log_message(f"Error loading tracking state: {e}") def check_frame_lock(course: str, frame: str) -> bool: """Check if frame is locked and lock hasn't expired""" if course in frame_locks and frame in frame_locks[course]: lock = frame_locks[course][frame] if time.time() - lock["locked_at"] < LOCK_TIMEOUT: return True # Lock expired, remove it del frame_locks[course][frame] save_tracking_state() return False def lock_frame(course: str, frame: str, requester_id: str) -> bool: """Attempt to lock a frame for processing""" if check_frame_lock(course, frame): return False if course not in frame_locks: frame_locks[course] = {} frame_locks[course][frame] = { "locked_by": requester_id, "locked_at": time.time() } save_tracking_state() return True def mark_frame_processed(course: str, frame: str, requester_id: str): """Mark a frame as successfully processed""" if course not in processed_frames: processed_frames[course] = {} processed_frames[course][frame] = { "processed_by": requester_id, "processed_at": time.time() } # Remove the lock if it exists if course in frame_locks and frame in frame_locks[course]: del frame_locks[course][frame] save_tracking_state() def log_message(message): """Add a log message with timestamp""" timestamp = datetime.now().strftime("%H:%M:%S") log_entry = f"[{timestamp}] {message}" processing_status["logs"].append(log_entry) # Keep only the last 100 logs if len(processing_status["logs"]) > 100: processing_status["logs"] = processing_status["logs"][-100:] print(log_entry) @app.on_event("startup") async def startup_event(): """Initialize frame tracking and start processing loop""" # Load frame tracking state load_tracking_state() log_message("āœ“ Loaded frame tracking state") # Start processing thread global processing_thread if not (processing_thread and processing_thread.is_alive()): log_message("šŸš€ Starting RAR extraction, frame extraction, and vision analysis pipeline in background...") processing_thread = threading.Thread(target=main_processing_loop) processing_thread.daemon = True processing_thread.start() @app.get("/") async def root(): """Root endpoint that returns basic info""" return { "message": "Video Analysis API", "status": "running", "endpoints": { "/status": "Get processing status", "/courses": "List all available course folders", "/images/{course_folder}": "List images in a course folder", "/images/{course_folder}/{frame_filename}": "Get specific frame image", "/start-processing": "Start processing pipeline", "/stop-processing": "Stop processing pipeline" } } @app.get("/status") async def get_status(): """Get current processing status""" return { "processing_status": processing_status } # ===== NEW IMAGE SERVING ENDPOINTS ===== @app.post("/middleware/release/frame/{course_folder}/{video}/{frame}") async def release_frame(course_folder: str, video: str, frame: str, requester_id: str): """Release a frame lock""" if course_folder in frame_locks and frame in frame_locks[course_folder]: lock = frame_locks[course_folder][frame] if lock["locked_by"] == requester_id: del frame_locks[course_folder][frame] save_tracking_state() return {"status": "released"} return {"status": "not_found"} @app.post("/middleware/release/course/{course_folder}") async def release_course(course_folder: str, requester_id: str): """Release all frame locks for a course""" if course_folder in frame_locks: # Only release frames locked by this requester frames_to_release = [ frame for frame, lock in frame_locks[course_folder].items() if lock["locked_by"] == requester_id ] for frame in frames_to_release: del frame_locks[course_folder][frame] save_tracking_state() return {"status": "released"} """ List all available course folders with their image counts """ if not os.path.exists(FRAMES_OUTPUT_FOLDER): return {"courses": [], "message": "Frames output folder does not exist yet"} courses = [] for folder in os.listdir(FRAMES_OUTPUT_FOLDER): folder_path = os.path.join(FRAMES_OUTPUT_FOLDER, folder) if os.path.isdir(folder_path): # Count image files image_count = len([f for f in os.listdir(folder_path) if f.lower().endswith(('.png', '.jpg', '.jpeg'))]) courses.append({ "course_folder": folder, "image_count": image_count, "images_url": f"/images/{folder}", "sample_image_url": f"/images/{folder}/0001.png" if image_count > 0 else None }) return { "total_courses": len(courses), "courses": courses } # Signal handlers to prevent accidental shutdown def handle_shutdown(signum, frame): """Prevent shutdown on SIGTERM/SIGINT""" print(f"\nāš ļø Received signal {signum}. Server will continue running.") print("Use Ctrl+Break or kill -9 to force stop.") # Setup signal handlers for graceful shutdown prevention import signal signal.signal(signal.SIGINT, handle_shutdown) signal.signal(signal.SIGTERM, handle_shutdown) # Server lifecycle events @app.on_event("shutdown") async def shutdown_event(): """Save state on shutdown attempt""" save_tracking_state() print("šŸ’¾ Saved tracking state") print("āš ļø Server shutdown prevented - use Ctrl+Break or kill -9 to force stop") # Prevent shutdown by not returning while True: await asyncio.sleep(1) if __name__ == "__main__": # Start the FastAPI server print("šŸš€ Starting Video Analysis FastAPI Server (Persistent Mode)...") print("API Documentation will be available at: http://localhost:8000/docs") print("API Root endpoint: http://localhost:8000/") print("āš ļø Server will continue running even after processing completes") print("Use Ctrl+Break or kill -9 to force stop") # Start processing in thread instead of blocking processing_thread = threading.Thread(target=main_processing_loop) processing_thread.daemon = False # Make non-daemon so it doesn't exit processing_thread.start() # Configure uvicorn for persistent running config = uvicorn.Config( app=app, host="0.0.0.0", port=8000, log_level="info", reload=False, workers=1, loop="asyncio", timeout_keep_alive=600, # Keep connections alive longer access_log=True ) # Run server with persistent config server = uvicorn.Server(config) server.run()