download_api.py

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()