40 lines
1.4 KiB
Python
40 lines
1.4 KiB
Python
import json
|
|
from fastapi import APIRouter, HTTPException, UploadFile
|
|
import cv2
|
|
import fastdeploy as fd
|
|
from fastdeploy.serving.utils import cv2_to_base64
|
|
import aiohttp
|
|
import asyncio
|
|
import io
|
|
from fastapi.responses import StreamingResponse
|
|
import numpy as np
|
|
|
|
router = APIRouter()
|
|
|
|
|
|
async def request(url: str, data, headers):
|
|
async with aiohttp.ClientSession() as session:
|
|
async with session.post(url=url, data=data, headers=headers) as response:
|
|
return await response.json()
|
|
|
|
|
|
@router.post("/get_visuallized_image")
|
|
async def getVisualized_image(image: UploadFile):
|
|
contents = await image.read()
|
|
nparr = np.fromstring(contents, np.uint8)
|
|
im = cv2.imdecode(nparr, cv2.IMREAD_COLOR)
|
|
url = "http://127.0.0.1:8000/fd/yolov5s"
|
|
headers = {"Content-Type": "application/json"}
|
|
data = {"data": {"image": cv2_to_base64(im)}, "parameters": {}}
|
|
try:
|
|
resp = await asyncio.gather(
|
|
*[request(url=url, data=json.dumps(data), headers=headers)]
|
|
)
|
|
r_json = json.loads(resp[0]["result"])
|
|
det_result = fd.vision.utils.json_to_detection(r_json)
|
|
vis_im = fd.vision.vis_detection(im, det_result, score_threshold=0.5)
|
|
_res, im_jpg = cv2.imencode(".jpg", vis_im)
|
|
return StreamingResponse(io.BytesIO(im_jpg.tobytes()), media_type="image/jpeg")
|
|
except:
|
|
raise HTTPException(status_code=500, detail="获取失败")
|