Spaces:
Sleeping
Sleeping
File size: 2,792 Bytes
066c36c |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 |
from fastapi import FastAPI
from fastapi.responses import JSONResponse
from fastapi import HTTPException
import uvicorn
from demos.cam_service_example.start_frame_capturing import ImageCaptureService
from demos.cam_service_example.stop_frame_capturing import StopImageCaptureServiceExample
from demos.image_load_service_example.start_image_load_example import StartImageLoadExample
from demos.image_load_service_example.stop_image_load_example import StopImageLoadServiceExample
app = FastAPI()
def start_ipcam():
flag_path = "resources/flag"
source = 'rtsp://ishwor:subedi@192.168.1.106:5555/h264_opus.sdp'
image_path_to_save = "images/cam_images"
image_hash_threshold = 5
image_capture_start_example = ImageCaptureService(flag_path, source, image_path_to_save,
image_hash_threshold)
image_capture_start_example.start_service()
def stop_ipcam():
flag_path = "resources/flag"
image_capture_stop_example = StopImageCaptureServiceExample(flag_path)
image_capture_stop_example.stop_load_image_example()
def start_detection():
flag_path = "resources/flag_load_image"
image_dir_path = "images/cam_images"
start_load_image_example = StartImageLoadExample(flag_path, image_dir_path, model_path="resources/model/v1/best.pt")
start_load_image_example.start_service()
def stop_detection():
flag_path = "resources/flag_load_image"
stop_load_image_example = StopImageLoadServiceExample(flag_path)
stop_load_image_example.stop_service()
# FastAPI Endpoints
@app.post("/start_ipcam_server")
def start_ipcam_server_api():
try:
start_ipcam()
return JSONResponse(content={"message": "IP Cam Server started!"}, status_code=200)
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.post("/stop_ipcam_server")
def stop_ipcam_server_api():
try:
stop_ipcam()
return JSONResponse(content={"message": "IP Cam Server stopped!"}, status_code=200)
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.post("/start_detection_service")
def start_detection_service_api():
try:
start_detection()
return JSONResponse(content={"message": "Image loading Server started!"}, status_code=200)
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.post("/stop_detection_service")
def stop_detection_service_api():
try:
stop_detection()
return JSONResponse(content={"message": "Image loading Server stopped!"}, status_code=200)
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
if __name__ == "__main__":
uvicorn.run(app, host="localhost", port=8001)
|