blong12/API_Arduino_video_stream
0
1# Import necessary modules2import json3import os4import uuid5from fastapi import APIRouter, HTTPException6from fastapi import File, UploadFile7from app.base_model import DataResponse, VideoData8 9from app.data_loader import UserDataManager10from .config import *11 12router = APIRouter()13 14 15@router.get("/", response_model=DataResponse)16def status():17 return {18 "status": "ok",19 "author": "Blong12",20 "data": {"message": "Welcome to my FastAPI application!"},21 "message": "Success"22 }23 24 25@router.get("/api/list_video", response_model=DataResponse)26def list_video():27 video_directory = "src/video"28 if os.path.exists(video_directory ):29 video_files = [file for file in os.listdir(video_directory) if file.endswith(".json")] 30 return {31 "status": "ok",32 "author": "Blong12",33 "data": video_files,34 "message": "Video deleted successfully",35 }36 else:37 return {38 "status": "error",39 "author": "Blong12",40 "data": None,41 "message": "Video not found",42 }43 44 45@router.get("/api/video_stream", response_model=DataResponse)46def video_stream(video_id: str, frame: int):# -> dict[str, Any]:47 video_filename = f"src/video/{video_id}.json"48 if os.path.exists(video_filename):49 with open(video_filename, "r") as json_file:50 video_data_dict = json.load(json_file)51 print(video_data_dict)52 response_data = {53 # "video_id": video_data_dict.video_id,54 "video_info": {55 "frames": video_data_dict['upload_data']['Frames'],56 "width": video_data_dict['upload_data']['Width'],57 "height": video_data_dict['upload_data']['Height'],58 },59 "frame":frame,60 "count_frame": len(video_data_dict['upload_data']['Frame']),61 "frame_data": video_data_dict['upload_data']['Frame'][frame],62 }63 return {64 "status": "ok",65 "author": "Blong12",66 "data": response_data,67 "message": "Video deleted successfully",68 }69 else:70 return {71 "status": "error",72 "author": "Blong12",73 "data": None,74 "message": "Video not found",75 }76 77 78@router.post("/api/add_video", response_model=DataResponse)79async def add_video(video_file: UploadFile = File(...)):80 video_data = VideoData.from_file(video_file)81 video_data_dict = video_data.to_dict()82 json_filename = f"src/video/{video_data_dict['video_id']}.json"83 with open(json_filename, "w") as json_file:84 json.dump(video_data.to_dict(), json_file)85 86 return {87 "status": "ok",88 "author": "Blong12",89 "data": video_data_dict,90 "message": "Video added successfully"91 }92 93 94@router.put("/api/replace_video/{video_id}", response_model=DataResponse)95async def replace_video(video_id: str, video_file: UploadFile = File(...)):96 video_data = VideoData.from_file(video_file)97 video_data_dict = video_data.to_dict()98 json_filename = f"src/video/{uuid.uuid4()}.json"99 with open(json_filename, "w") as json_file:100 json.dump(video_data.to_dict(), json_file)101 102 return {103 "status": "ok",104 "author": "Blong12",105 "data": video_data.to_dict(),106 "message": "Video replaced successfully",107 }108 109 110@router.delete("/api/delete_video/{video_id}", response_model=DataResponse)111def delete_video(video_id: str):112 video_filename = f"src/video/{video_id}.json"113 if os.path.exists(video_filename):114 os.remove(video_filename) 115 return {116 "status": "ok",117 "author": "Blong12",118 "data": None,119 "message": "Video deleted successfully",120 }121 else:122 return {123 "status": "error",124 "author": "Blong12",125 "data": None,126 "message": "Video not found",127 }128 129 130 