o

app.py

@@ -1,124 +0,0 @@
-import gradio as gr
-
-
-
-import os
-import json
-from ultralytics import YOLO
-import supervision  as sv
-
-
-# --- 1. CONFIGURATION ---
-
-# Folder where the script is
-BASE_DIR = os.path.dirname(os.path.abspath(__file__))
-
-# Go up one level to Om_Singh
-PROJECT_DIR = os.path.dirname(BASE_DIR)  # parent folder of app/
-
-# Paths for outputs in video_and_json folder
-VIDEO_DIR = os.path.join(PROJECT_DIR, "video_and_json")  # Om_Singh/video_and_json
-os.makedirs(VIDEO_DIR, exist_ok=True)  # ensure folder exists
-
-MODEL_PATH = os.path.join(BASE_DIR, "best.pt")  # model stays in app/
-OUTPUT_VIDEO_PATH = os.path.join(VIDEO_DIR, "output.mp4")
-OUTPUT_JSON_PATH  = os.path.join(VIDEO_DIR, "result.json")
-
-def process_video(INPUT_VIDEO_PATH, OUTPUT_VIDEO_PATH, OUTPUT_JSON_PATH):
-    print("Loading model...")
-    model = YOLO(MODEL_PATH)
-
-    print("Initializing tracker and annotators...")
-    tracker = sv.ByteTrack()
-    box_annotator = sv.BoxAnnotator(thickness=5)
-    label_annotator = sv.LabelAnnotator(text_position=sv.Position.TOP_CENTER, text_scale = 1, text_thickness= 1)
-
-    sv.LabelAnnotator()
-
-    frame_generator = sv.get_video_frames_generator(source_path=INPUT_VIDEO_PATH)
-    video_info = sv.VideoInfo.from_video_path(INPUT_VIDEO_PATH)
-
-    results_list = []
-
-    with sv.VideoSink(target_path=OUTPUT_VIDEO_PATH, video_info=video_info) as sink:
-        print("Processing video frames...")
-        for frame_number, frame in enumerate(frame_generator):
-            # Run YOLO prediction
-            results = model(frame)[0]
-            detections = sv.Detections.from_ultralytics(results)
-
-            # Update tracker
-            tracked_detections = tracker.update_with_detections(detections=detections)
-
-            # Prepare labels
-            labels = [
-                f"ID: {det[4]} {model.model.names[int(det[3])]}"
-                for det in tracked_detections
-            ]
-
-            # Annotate frame
-            annotated_frame = box_annotator.annotate(scene=frame.copy(), detections=tracked_detections)
-            annotated_frame = label_annotator.annotate(scene=annotated_frame, detections=tracked_detections, labels=labels)
-
-            # Save tracking info
-            for det in tracked_detections:
-                bbox = det[0]
-                conf = det[2]
-                class_id = int(det[3])
-                tracker_id = det[4]
-
-                results_list.append({
-                    "frame_number": frame_number,
-                    "track_id": int(tracker_id),
-                    "class": model.model.names[class_id],
-                    "confidence": float(conf),
-                    "bounding_box": [int(coord) for coord in bbox]
-                })
-
-            # Write annotated frame
-            sink.write_frame(frame=annotated_frame)
-
-            if frame_number % 30 == 0:
-                print(f"Processed frame {frame_number}...")
-
-    print("Video processing complete. Saving results...")
-    with open(OUTPUT_JSON_PATH, 'w') as f:
-        json.dump(results_list, f, indent=4)
-
-    print("--- All tasks finished successfully! ---")
-
-
-
-# --- Main processing function ---
-def process(input_video):
-    output_video = "output.mp4"
-    output_json = "result.json"
-    
-    # During processing: red text
-    status_html = "<p style='color:red; font-weight:bold;'>Processing...</p>"
-    
-    # Run video processing
-    process_video(input_video, output_video, output_json)
-    
-    # After processing: green text
-    status_html = "<p style='color:limegreen; font-weight:bold;'>Processing complete!</p>"
-    return status_html, output_video, output_json
-
-# --- Gradio UI ---
-with gr.Blocks() as demo:
-    gr.Markdown("<h1 style='text-align:center;'>Vehicle and Pedestrian Tracker</h1>")
-
-    input_video = gr.Video(label="Upload Video")
-    start_btn = gr.Button("Start Tracking")
-    status_display = gr.HTML("")  # Initially empty
-    output_video = gr.Video(label="Processed Video")
-    output_json = gr.File(label="Download JSON Output")
-
-    start_btn.click(
-        fn=process,
-        inputs=input_video,
-        outputs=[status_display, output_video, output_json]
-    )
-
-if __name__ == "__main__":
-    demo.launch(share = True)

backend.py

@@ -0,0 +1,66 @@
+import os
+import json
+import tempfile
+from fastapi import FastAPI, UploadFile, File
+from fastapi.responses import FileResponse
+from ultralytics import YOLO
+import supervision as sv
+
+# --- MODEL AND APP INITIALIZATION ---
+# The model is loaded only ONCE when the server starts, making it fast.
+MODEL_PATH = "best.pt"
+model = YOLO(MODEL_PATH)
+
+app = FastAPI(title="YOLOv8 Tracking API")
+
+# --- YOUR EXISTING PROCESSING LOGIC (UNCHANGED) ---
+# We just put your code inside this function.
+def process_video_logic(input_path, output_path, json_path):
+    tracker = sv.ByteTrack()
+    box_annotator = sv.BoxAnnotator(thickness=5)
+    label_annotator = sv.LabelAnnotator(text_position=sv.Position.TOP_CENTER, text_scale=1, text_thickness=1)
+    frame_generator = sv.get_video_frames_generator(source_path=input_path)
+    video_info = sv.VideoInfo.from_video_path(input_path)
+    results_list = []
+
+    with sv.VideoSink(target_path=output_path, video_info=video_info) as sink:
+        for frame_number, frame in enumerate(frame_generator):
+            results = model(frame, verbose=False)[0]
+            detections = sv.Detections.from_ultralytics(results)
+            tracked_detections = tracker.update_with_detections(detections=detections)
+            labels = [f"ID: {det[4]} {model.model.names[int(det[3])]}" for det in tracked_detections]
+            annotated_frame = box_annotator.annotate(scene=frame.copy(), detections=tracked_detections)
+            annotated_frame = label_annotator.annotate(scene=annotated_frame, detections=tracked_detections, labels=labels)
+            
+            for det in tracked_detections:
+                bbox, conf, class_id, tracker_id = det[0], det[2], int(det[3]), det[4]
+                results_list.append({
+                    "frame_number": frame_number,
+                    "track_id": int(tracker_id),
+                    "class": model.model.names[class_id],
+                    "confidence": float(conf),
+                    "bounding_box": [int(coord) for coord in bbox]
+                })
+            sink.write_frame(frame=annotated_frame)
+
+    with open(json_path, 'w') as f:
+        json.dump(results_list, f, indent=4)
+
+# --- API ENDPOINT ---
+@app.post("/track/")
+async def track_video_endpoint(video: UploadFile = File(...)):
+    # Use a temporary directory to handle file operations safely
+    with tempfile.TemporaryDirectory() as temp_dir:
+        input_path = os.path.join(temp_dir, video.filename)
+        output_video_path = os.path.join(temp_dir, "output.mp4")
+        output_json_path = os.path.join(temp_dir, "results.json")
+
+        # Save the uploaded video file
+        with open(input_path, "wb") as buffer:
+            buffer.write(await video.read())
+
+        # Run your existing processing logic
+        process_video_logic(input_path, output_video_path, output_json_path)
+
+        # Return the processed video as a downloadable file
+        return FileResponse(output_video_path, media_type="video/mp4", filename="output.mp4")

gradio_ui.py

@@ -0,0 +1,56 @@
+import gradio as gr
+import requests
+import os
+
+# --- IMPORTANT: UPDATE THIS URL ---
+# After deploying your backend_app.py, paste its public URL here.
+BACKEND_URL = "https://your-backend-space-name.hf.space/track/"
+
+def track_video_via_api(input_video_path):
+    if input_video_path is None:
+        return None, "Please upload a video first."
+    
+    status = "Sending video to the backend for processing..."
+    yield status, None, None # Update status immediately
+
+    try:
+        with open(input_video_path, "rb") as video_file:
+            files = {'video': (os.path.basename(input_video_path), video_file, 'video/mp4')}
+            
+            # Send the video to the backend API and stream the response
+            with requests.post(BACKEND_URL, files=files, stream=True, timeout=300) as response:
+                response.raise_for_status()
+                
+                # Save the processed video that the backend sends back
+                processed_video_path = "processed_video.mp4"
+                with open(processed_video_path, "wb") as f:
+                    for chunk in response.iter_content(chunk_size=8192):
+                        f.write(chunk)
+                
+                # For this Gradio app, we will just show the video.
+                # You could modify the backend to return JSON as well.
+                status = "Processing complete!"
+                yield status, processed_video_path, None
+
+    except requests.exceptions.RequestException as e:
+        status = f"API Error: {e}"
+        yield status, None, None
+
+# --- GRADIO UI (from your original app.py) ---
+with gr.Blocks() as demo:
+    gr.Markdown("<h1 style='text-align:center;'>Vehicle and Pedestrian Tracker</h1>")
+
+    input_video = gr.Video(label="Upload Video")
+    start_btn = gr.Button("Start Tracking")
+    status_display = gr.HTML("")  # Initially empty
+    output_video = gr.Video(label="Processed Video")
+    output_json = gr.File(label="Download JSON Output") # Note: This is not hooked up in this version
+
+    start_btn.click(
+        fn=track_video_via_api,
+        inputs=input_video,
+        outputs=[status_display, output_video, output_json]
+    )
+
+if __name__ == "__main__":
+    demo.launch(share = True)