Spaces:
Sleeping
Sleeping
File size: 3,432 Bytes
c640bc9 |
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 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 |
from fastapi import FastAPI
from fastapi.responses import JSONResponse
from fastapi import HTTPException
import uvicorn
import base64
import io
from PIL import Image
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
from demos.single_image_inference import single_image_inference
from pydantic import BaseModel
app = FastAPI()
# Helper functions
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))
class ImageInput(BaseModel):
base64_image: str
@app.post("/single_image_detection")
def single_image_detection_api(image_input: ImageInput):
try:
image_data = base64.b64decode(image_input.base64_image)
image = Image.open(io.BytesIO(image_data))
image = single_image_inference(image)
encoded_img = base64.b64encode(image)
return encoded_img
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
if __name__ == "__main__":
uvicorn.run(app, host="localhost", port=8001)
|