jnhg_lidarSpeed/server/test.py

import asyncio
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

conf = Config.get("test")
url = conf["url"]
classIndex = conf["classIndex"]
nvr = conf["nvr"]

print(url)
async def local_test():
    with open("../assert/isapi_data/tfs_0.txt", "r") as f:
        multipart_data = eval(f.read())
    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)
                    # img.show()
                    if ip_addr not in conf["ip2channle"]:
                        logger.info(f"{ip_addr} not in ip2channle")
                        return
                    channel = conf["ip2channle"][ip_addr]
                    res = await upload(url, channel, classIndex, nvr, videoTime, filename, imagedata, plate, speed)
                    logger.info(res)
            except Exception as error:
                logger.info(error)
    return {"msg": "sucess"}


if __name__ == "__main__":
    asyncio.run(local_test())