| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960 |
- import uvicorn
- from fastapi import FastAPI, Request
- from logger import logger
- from config import Config
- from uploadData import upload
- from parser import parse_multipart
- from parser import parser_json_data
- from byteimg import byte2pil, pil2byte, draw
- app = FastAPI()
- conf = Config.get("prod")
- url = conf["url"]
- channel = conf["channel"]
- classIndex = conf["classIndex"]
- nvr = conf["nvr"]
- @app.post("/speed")
- async def speed(request: Request):
- multipart_data = await request.body()
- boundary = b'---------------------------7e13971310878'
- result = parse_multipart(multipart_data, boundary)
- car_infos, image_contents = parser_json_data(result)
- if len(car_infos) != 0:
- # with open(f"assert/{index}-tfx.txt", "wb") as f:
- # f.write(multipart_data)
- logger.info(car_infos)
-
- for info, image_content in zip(car_infos, image_contents):
- try:
- videoTime = info["dateTime"]
- speed = info["VehicleInfo"]["vehicleSpeed"]
- plate = info["PlateInfo"]["plate"]
- filename = None
- imagedata = None
- filename = image_content["filename"]
- if filename != "detectionPicture.jpg":
- continue
- imagedata = image_content["content"]
- if imagedata is not None:
- img = byte2pil(imagedata)
- text = f"{plate} : {speed} km/h"
- draw(img, text, (100, 100))
- byte_data = pil2byte(img)
- res = await upload(url, channel, classIndex, nvr, videoTime, filename, byte_data, plate, speed)
- logger.info(res)
- except Exception as error:
- logger.info(error)
- return {"msg": "sucess"}
- if __name__ == "__main__":
- uvicorn.run('server:app', host="0.0.0.0", port=18000)
|
