| 12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364 |
- 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"]
- 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"]
- ip_addr = info["ipAddress"]
- 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)
- if ip_addr not in conf["ip2channel"]:
- logger.info(f"{ip_addr} not in ip2channel")
- return
- channel = conf["ip2channel"][ip_addr]
- res = await upload(url, ip_addr, 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)
|
