am-om commited on
Commit
7b9d679
·
1 Parent(s): 06a550b
Files changed (1) hide show
  1. backend.py +0 -66
backend.py DELETED
@@ -1,66 +0,0 @@
1
- import os
2
- import json
3
- import tempfile
4
- from fastapi import FastAPI, UploadFile, File
5
- from fastapi.responses import FileResponse
6
- from ultralytics import YOLO
7
- import supervision as sv
8
-
9
- # --- MODEL AND APP INITIALIZATION ---
10
- # The model is loaded only ONCE when the server starts, making it fast.
11
- MODEL_PATH = "best.pt"
12
- model = YOLO(MODEL_PATH)
13
-
14
- app = FastAPI(title="YOLOv8 Tracking API")
15
-
16
- # --- YOUR EXISTING PROCESSING LOGIC (UNCHANGED) ---
17
- # We just put your code inside this function.
18
- def process_video_logic(input_path, output_path, json_path):
19
- tracker = sv.ByteTrack()
20
- box_annotator = sv.BoxAnnotator(thickness=5)
21
- label_annotator = sv.LabelAnnotator(text_position=sv.Position.TOP_CENTER, text_scale=1, text_thickness=1)
22
- frame_generator = sv.get_video_frames_generator(source_path=input_path)
23
- video_info = sv.VideoInfo.from_video_path(input_path)
24
- results_list = []
25
-
26
- with sv.VideoSink(target_path=output_path, video_info=video_info) as sink:
27
- for frame_number, frame in enumerate(frame_generator):
28
- results = model(frame, verbose=False)[0]
29
- detections = sv.Detections.from_ultralytics(results)
30
- tracked_detections = tracker.update_with_detections(detections=detections)
31
- labels = [f"ID: {det[4]} {model.model.names[int(det[3])]}" for det in tracked_detections]
32
- annotated_frame = box_annotator.annotate(scene=frame.copy(), detections=tracked_detections)
33
- annotated_frame = label_annotator.annotate(scene=annotated_frame, detections=tracked_detections, labels=labels)
34
-
35
- for det in tracked_detections:
36
- bbox, conf, class_id, tracker_id = det[0], det[2], int(det[3]), det[4]
37
- results_list.append({
38
- "frame_number": frame_number,
39
- "track_id": int(tracker_id),
40
- "class": model.model.names[class_id],
41
- "confidence": float(conf),
42
- "bounding_box": [int(coord) for coord in bbox]
43
- })
44
- sink.write_frame(frame=annotated_frame)
45
-
46
- with open(json_path, 'w') as f:
47
- json.dump(results_list, f, indent=4)
48
-
49
- # --- API ENDPOINT ---
50
- @app.post("/track/")
51
- async def track_video_endpoint(video: UploadFile = File(...)):
52
- # Use a temporary directory to handle file operations safely
53
- with tempfile.TemporaryDirectory() as temp_dir:
54
- input_path = os.path.join(temp_dir, video.filename)
55
- output_video_path = os.path.join(temp_dir, "output.mp4")
56
- output_json_path = os.path.join(temp_dir, "results.json")
57
-
58
- # Save the uploaded video file
59
- with open(input_path, "wb") as buffer:
60
- buffer.write(await video.read())
61
-
62
- # Run your existing processing logic
63
- process_video_logic(input_path, output_video_path, output_json_path)
64
-
65
- # Return the processed video as a downloadable file
66
- return FileResponse(output_video_path, media_type="video/mp4", filename="output.mp4")