init

.gitignore

@@ -0,0 +1,6 @@
+images/
+result/
+__pycache__
+log/
+models/
+assert/

Makefile

@@ -0,0 +1,38 @@
+start:
+	@cd server && \
+	gunicorn -w 2 -b 0.0.0.0:18000 -k uvicorn.workers.UvicornWorker server:app --daemon
+	@sleep 1s
+	@ps -ef | grep gunicorn
+
+stop:
+	@pkill -f gunicorn
+	@sleep 1s
+	@ps -ef | grep gunicorn 
+
+restart: 
+	stop 
+	start
+
+status:
+	@ps -ef | grep gunicorn 
+
+debug:
+	@cd server && \
+	python3 server.py
+
+test:
+	@cd server && python3 test.py
+
+log:
+	@tail -f log/runtime.log
+
+help:
+	@echo "make start   => start server"
+	@echo "make stop    => stop server"
+	@echo "make restart => restart server"
+	@echo "make status  => check server status"
+	@echo "make debug   => python3 server.py"
+	@echo "make log     => print log"
+	@echo "make help    => help"
+
+.PHONY: start stop restart status debug help log

server/byteimg.py

@@ -0,0 +1,19 @@
+from io import BytesIO
+from PIL import Image, ImageDraw, ImageFont
+
+def byte2pil(byte_data):
+    img = Image.open(BytesIO(byte_data))
+    return img
+
+def pil2byte(img):
+    img_bytes = BytesIO()
+    img.save(img_bytes, format='JPEG')
+    img_bytes = img_bytes.getvalue()
+    return img_bytes
+
+def draw(img, text, position):
+    font = ImageFont.truetype("font/SIMHEI.TTF", size=128)
+    # print(font)
+    draw = ImageDraw.Draw(img)
+    draw.text(position, text, fill = (255, 0 ,0), font=font)
+    return img

server/config.py

@@ -0,0 +1,21 @@
+class Config:
+    # 生产环境配置
+    PROD = {
+        'url' : "http://172.19.152.231/open/api/operate/upload",
+        'channel': "64",
+        'classIndex': "68",
+        'nvr': "172.19.152.231"
+    }
+
+    # 测试环境配置
+    TEST = {
+        'url' : "http://36.7.84.146:28801/open/api/operate/upload",
+        'channel': "251",
+        'classIndex': "101",
+        'nvr': "172.16.20.251"
+    }
+
+    @staticmethod
+    def get(environment="test"):
+        # 根据环境选择配置，默认选择测试环境
+        return Config.TEST if environment == "test" else Config.PROD

server/logger.py

@@ -0,0 +1,68 @@
+from functools import wraps
+import os
+import datetime
+import loguru
+
+
+# 单例类的装饰器
+def singleton_class_decorator(cls):
+    """
+    装饰器，单例类的装饰器
+    """
+    _instance = {}
+
+    @wraps(cls)
+    def wrapper_class(*args, **kwargs):
+        if cls not in _instance:
+            _instance[cls] = cls(*args, **kwargs)
+        return _instance[cls]
+
+    return wrapper_class
+
+
+@singleton_class_decorator
+class Logger:
+    def __init__(self):
+        self.logger_add()
+
+    def get_project_path(self, project_path=None):
+        if project_path is None:
+            project_path = os.path.realpath('..')
+        return project_path
+
+    def get_log_path(self):
+        project_path = self.get_project_path()
+        project_log_dir = os.path.join(project_path, 'log')
+        # 统一使用 runtime.log 作为日志文件名
+        project_log_filename = 'runtime.log'  
+        project_log_path = os.path.join(project_log_dir, project_log_filename)
+        return project_log_path
+
+    def logger_add(self):
+        loguru.logger.add(
+            sink=self.get_log_path(),
+            rotation="50 MB",  # 每个日志文件最大10MB
+            retention=5,  # 最多保留5个日志文件
+            compression=None,
+            encoding="utf-8",
+            enqueue=True
+        )
+
+    @property
+    def get_logger(self):
+        return loguru.logger
+
+
+# 实例化日志类
+logger = Logger().get_logger
+
+
+if __name__ == '__main__':
+    logger.debug('调试代码')
+    logger.info('输出信息')
+    logger.success('输出成功')
+    logger.warning('错误警告')
+    logger.error('代码错误')
+    logger.critical('崩溃输出')
+
+    logger.info('----原始测试----')

server/parser.py

@@ -0,0 +1,173 @@
+import email
+from email.message import Message
+from email.policy import default
+import json
+from datetime import datetime
+from lxml import etree
+
+def clean_xml(xml_string):
+    """移除 XML 中的命名空间，包括 xmlns 声明"""
+    parser = etree.XMLParser(remove_blank_text=True)
+    root = etree.XML(xml_string, parser)
+
+    # 遍历所有元素，移除命名空间
+    for elem in root.iter():
+        if '}' in elem.tag:  # 如果有命名空间
+            elem.tag = elem.tag.split('}', 1)[1]  # 移除命名空间
+        if 'version' in elem.attrib:
+            del elem.attrib['version']
+    etree.cleanup_namespaces(root)
+    return etree.tostring(root, pretty_print=True, encoding="unicode")
+
+def parse_xml_to_dict(xml):
+    """
+    将xml文件解析成字典形式，参考tensorflow的recursive_parse_xml_to_dict
+    Args:
+        xml: xml tree obtained by parsing XML file contents using lxml.etree
+
+    Returns:
+        Python dictionary holding XML contents.
+    """
+
+    if len(xml) == 0:  # 遍历到底层，直接返回tag对应的信息
+        return{xml.tag: xml.text}
+
+    result= {}
+    for child in xml:
+        child_result= parse_xml_to_dict(child)  # 递归遍历标签信息
+        if child.tag not in ['pictureInfo', 'ANPR', "PictureURLInfo"]:
+            result[child.tag] = child_result[child.tag]
+        else:
+            if child.tag not in result:  # 因为object可能有多个，所以需要放入列表里
+                result[child.tag] = []
+            result[child.tag].append(child_result[child.tag])
+    return {xml.tag: result}
+
+
+def parse_multipart(data, boundary):
+    # 按 boundary 分割数据
+    parts = data.split(boundary)
+    parsed_data = {}
+    for part in parts:
+        # print("part : ", part)
+        part = part.strip()
+        # print("part strip: ", part)
+        if not part: continue
+        # 获取头部和主体部分
+        headers, _, body = part.partition(b"\r\n\r\n")
+        # print("headers :", headers)
+        if not headers: continue
+        header_dict = email.message_from_bytes(headers, policy=default)
+        # print("header_dict : ", header_dict)
+        # 获取 Content-Disposition
+        content_disposition = header_dict.get("Content-Disposition", "")
+        # print("content_disposition :", content_disposition)
+        if "name=" in content_disposition:
+            name = content_disposition.split('name="')[1].split('"')[0]
+            filename = None
+            if 'filename="' in content_disposition:
+                filename = content_disposition.split('filename="')[1].split('"')[0]
+
+            # 提取 Content-Type
+            content_type = header_dict.get("Content-Type", "").strip()
+
+            # 根据 Content-Type 处理内容
+            if content_type == "application/json":
+                body_content = json.loads(body.decode("utf-8"))
+            elif content_type == "application/xml":
+                body = body.decode('utf-8').replace("</  ", "</").encode()
+                clean_xml_str = clean_xml(body)
+                xml = etree.fromstring(clean_xml_str)
+                body_content = parse_xml_to_dict(xml)  # 直接存储 XML 原始内容
+            elif content_type.startswith("image/"):
+                body_content = body  # 图像以二进制形式存储
+                # with open(filename, "wb") as f:
+                #     f.write(body)
+            else:
+                body_content = body.decode()  # 其他情况
+            
+            # 存储解析的数据
+            parsed_data[name] = {
+                "filename": filename,
+                "content_type": content_type,
+                "content": body_content,
+            }
+    return parsed_data
+
+
+def parser_json_data(data):
+    infos = []
+    images = []
+
+    for _, value in data.items():
+        content_type = value.get("content_type")
+        
+        if not content_type:
+            continue
+        
+        if content_type == "application/xml":
+            # 安全提取嵌套字段
+            content = value.get("content")
+            if content is None:
+                continue  # 跳过内容为空的记录
+
+            event_alert = content.get("EventNotificationAlert")
+            if event_alert is None:
+                continue  # 如果没有 EventNotificationAlert，跳过当前记录
+
+            iso_time = event_alert.get("dateTime")
+            dt = datetime.fromisoformat(iso_time)
+            dateTime = dt.strftime("%Y-%m-%d %H:%M:%S")
+
+            tfs = event_alert.get("TFS")
+            if tfs is None:
+                continue  # 如果没有 TFS，跳过当前记录
+
+            VehicleInfo = tfs.get("VehicleInfo")
+            PlateInfo = tfs.get("PlateInfo")
+            picture_info_list = event_alert.get("PictureURLInfoList", {}).get("PictureURLInfo", [])
+
+            # 对每个字段进行逐一检查，确保有效后再添加到 infos
+            if VehicleInfo is None or PlateInfo is None or not picture_info_list:
+                continue  # 如果任何重要字段为空，跳过当前记录
+
+            infos.append({
+                "dateTime" : dateTime,
+                "pictureInfo": picture_info_list,
+                "VehicleInfo": VehicleInfo,
+                "PlateInfo": PlateInfo
+            })
+
+        elif content_type == "image/jpeg":
+            # 安全提取图片内容
+            filename = value.get("filename")
+            image_content = value.get("content")
+
+            # 确保图片相关字段都有效后才添加
+            if filename and image_content:
+                images.append({
+                    "filename": filename,
+                    "content": image_content
+                })
+
+    return infos, images
+
+    
+if __name__ == "__main__":
+    with open("tfs_0.txt", "r") as f:
+        multipart_data = eval(f.read())
+    # print("===============", multipart_data[:3000])
+    # 示例调用
+    # multipart_data = b'---------------------------7e13971310878\r\n\r\nContent-Disposition: form-data; name="Q3531473496730912851860"; filename="radarVideoDetection.json"\r\n\r\nContent-Type: application/json\r\n\r\nContent-Length: 400\r\n\r\n\r\n\r\n{"ipAddress":"172.19.152.181","protocol":"HTTP","macAddress":"a4:d5:c2:02:96:bf","channelID":1,"dateTime":"2024-12-03T15:30:41.516+08:00","activePostCount":1,"eventType":"radarVideoDetection","eventState":"active","eventDescription":"Radar Video Detection","freezingTimeInfo":{"freezingTimestamp":96730747,"freezingSystemDateTime":"2024-12-03T15:30:41.360"},"Datas":[],"algorithmDataFrames":"966772"}\r\n\r\n---------------------------7e13971310878--\r\n\r\n'
+    # print(multipart_data[0:3000])
+    boundary = b'---------------------------7e13971310878'
+    result = parse_multipart(multipart_data, boundary)
+    print(result)
+    infos, images = parser_json_data(result)
+    print(infos, "======", images)
+
+    # with open("res.xml", "r") as f:
+    #     xml_str = f.read()
+    # clean_xml_str = clean_xml(xml_str.encode())
+    # xml = etree.fromstring(clean_xml_str)
+    # print(parse_xml_to_dict(xml))

server/server.py

@@ -0,0 +1,60 @@
+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)

server/test.py

@@ -0,0 +1,55 @@
+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("prod")
+url = conf["url"]
+channel = conf["channel"]
+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"]
+                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()
+                    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())

server/uploadData.py

@@ -0,0 +1,52 @@
+import aiohttp
+import asyncio
+from config import Config
+
+async def upload(url, channel, classIndex, nvr, videoTime, filename, imgContent, rateLicensePlate, rateSpeed):
+    payload = {
+        'channel': channel,
+        'classIndex': classIndex,
+        'ip': nvr,
+        'videoTime': videoTime,
+        'rateLicensePlate' : rateLicensePlate,
+        'rateSpeed' : rateSpeed
+    }
+
+    # 使用 aiohttp.FormData 构建文件上传的请求体
+    data = aiohttp.FormData()
+    for key, value in payload.items():
+        data.add_field(key, value)
+    
+    # 文件上传，注意这里是 'file' 字段
+    data.add_field('file', imgContent, filename=filename, content_type='image/jpeg')
+
+    # 发送异步 POST 请求
+    async with aiohttp.ClientSession() as session:
+        async with session.post(url, data=data) as response:
+            text = await response.text()
+            return text
+
+if __name__ == "__main__":
+    # for test
+    conf = Config.get("test")
+    url = conf["url"]
+    channel = conf["channel"]
+    classIndex = conf["classIndex"]
+    nvr = conf["nvr"]
+
+    # 测试配置
+    # channel = "64"
+    # classIndex = "68"
+    # nvr = "172.19.152.231"
+    # 生产配置
+    channel = "251"
+    classIndex = "101"
+    nvr = "172.16.20.251"
+    filename = "0-detectionPicture.jpg"
+    with open(filename, "rb") as f:
+        imgContent = f.read()
+    videoTime = "2024-12-05 11:43:33"
+    async def main():
+        response = await upload(url, channel, classIndex, nvr, videoTime, filename, imgContent)
+        print(await response.text())  # 获取响应文本内容
+    asyncio.run(main())