增加数量超限报警功能

middleware/entity/up_osd_info_push.py

@@ -1,7 +1,11 @@
 from dataclasses import dataclass
-import json
+
-from typing import Optional
 import asyncio
+from typing import Union, Optional
+from typing import Any
+
+
+
 
 
 @dataclass
@@ -25,9 +29,23 @@ class OSDInfo:
     wind_speed: float
 
 
+@dataclass
+class OSDInfo_v1:
+    attitude_head: float
+    latitude: float
+    longitude: float
+    height: float
+    speed_x: float
+    speed_y: float
+    speed_z: float
+    gimbal_pitch: float
+    gimbal_roll: float
+    gimbal_yaw: float
+
+
 @dataclass
 class OSDMessage:
-    data: OSDInfo
+    data: Any  # 可以是两种类型之一
     method: str
     seq: int
     timestamp: int
@@ -36,19 +54,47 @@ class OSDMessage:
 def parse_osd_message(json_str: Optional[str]) -> Optional[OSDMessage]:
     if not json_str:
         return None
-
-    try:
     data = json_str
+    try:
         osd_info = OSDInfo(**data["data"])
+        data_seq=data["seq"]
+    except (TypeError, KeyError) as e:
+        # 如果OSDInfo格式失败，尝试使用OSDInfo_v1格式
+        try:
+            osd_info = OSDInfo_v1(**data["data"])
+            data_seq = 0
+        except Exception as e2:
+            print(f"Error parsing OSD message with both formats: {e2}")
+            return None
     return OSDMessage(
         data=osd_info,
         method=data["method"],
-            seq=data["seq"],
+        seq=data_seq,
         timestamp=data["timestamp"]
     )
-    except Exception as e:
+    # try:
-        print(f"Error parsing OSD message: {e}")
+    #     data=json_str
-        return None
+    #     # osd_info = OSDInfo(**data["data"])
+    #     # if osd_info is None:
+    #     #     osd_info = OSDInfo_v1(**data["data"]) #适配东西湖区一代飞机的格式
+    #     try:
+    #         osd_info = OSDInfo(**data["data"])
+    #     except (TypeError, KeyError) as e:
+    #         # 如果OSDInfo格式失败，尝试使用OSDInfo_v1格式
+    #         try:
+    #             osd_info = OSDInfo_v1(**data["data"])
+    #         except Exception as e2:
+    #             print(f"Error parsing OSD message with both formats: {e2}")
+    #             return None
+    #     return OSDMessage(
+    #         data=osd_info,
+    #         method=data["method"],
+    #         seq=data["seq"],
+    #         timestamp=data["timestamp"]
+    #     )
+    # except Exception as e:
+    #     print(f"Error parsing OSD message: {e}")
+    #     return None
 
 
 async def main():

yolo/cv_multi_model_back_video.py

@@ -2681,7 +2681,8 @@ async def write_results_to_rtmp(task_id: str, output_url: str = None, input_fps:
                                     label = f"{en_name}:{confidence:.2f}:{track_id}"
                                     label_name = f"{en_name}"
                                     # 计算文本位置
-                                    text_size = cv2.getTextSize(label, cv2.FONT_HERSHEY_SIMPLEX, fontScale=8, thickness=4)[
+                                    text_size = \
+                                    cv2.getTextSize(label, cv2.FONT_HERSHEY_SIMPLEX, fontScale=8, thickness=4)[
                                         0]
                                     text_width, text_height = text_size[0], text_size[1]
                                     text_x = x1
@@ -2722,7 +2723,8 @@ async def write_results_to_rtmp(task_id: str, output_url: str = None, input_fps:
                                     label = f"{en_name}:{confidence:.2f}:{track_id}"
                                     label_name = f"{en_name}"
                                     # 计算文本位置
-                                    text_size = cv2.getTextSize(label, cv2.FONT_HERSHEY_SIMPLEX, fontScale=8, thickness=4)[
+                                    text_size = \
+                                    cv2.getTextSize(label, cv2.FONT_HERSHEY_SIMPLEX, fontScale=8, thickness=4)[
                                         0]
                                     text_width, text_height = text_size[0], text_size[1]
                                     text_x = x1
@@ -3118,7 +3120,8 @@ async def cal_des_invade(loop, invade_executor, task_id: str, mqtt, mqtt_publish
                          list_points: list[list[any]], camera_para: Camera_Para, model_count: int,
                          cancel_flag: asyncio.Event = None, invade_switch: int = 0, invade_queue: asyncio.Queue = None,
                          event_queue: asyncio.Queue = None,
-                         device_height: float = float(200), repeat_dis: float = -1, repeat_time: float = -1):
+                         device_height: float = float(200), repeat_dis: float = -1, repeat_time: float = -1,
+                         high_count_warn: float = -1):
     # loop = asyncio.get_running_loop()
     # upload_executor = ThreadPoolExecutor(max_workers=Config.UPLOAD_WORKERS)
     pic_count_hongxian = 0
@@ -3129,7 +3132,13 @@ async def cal_des_invade(loop, invade_executor, task_id: str, mqtt, mqtt_publish
     report_interval = 8
     target_location_back = []  # 本地缓存，用作位置重复计算
     current_time_second = int(time.time())
+    high_count_warn_status = False
+    high_count_warn_num = 0
     while not cancel_flag.is_set():
+        # 在循环开始处初始化变量
+        high_count_warn_status = False
+        high_count_warn_num = 0
+
         # 检查队列长度，避免堆积
         if invade_queue.qsize() > Config.PROCESSED_QUEUE_SIZE // 2:
             print(f"警告：invade_queue 积压（当前长度={invade_queue.qsize()}），清空队列")
@@ -3191,7 +3200,8 @@ async def cal_des_invade(loop, invade_executor, task_id: str, mqtt, mqtt_publish
         height = air_alti.height
         cam_longitude = air_alti.longitude
         cam_latitude = air_alti.latitude
-
+        high_count_warn_status = False  # 侵限情况下的超数报警，针对东西湖
+        high_count_warn_num = 0  # 侵限情况下的超数报警，针对东西湖
         try:
             current_time = time.time()
             h = device_height
@@ -3244,7 +3254,13 @@ async def cal_des_invade(loop, invade_executor, task_id: str, mqtt, mqtt_publish
             show_des = 0
             str_loca = ""
             des_location_result = []
+
             if repeat_dis > 0:  # ai_model_list repeat_dis 字段大于零，才启用去重
+                if target_location_back is not None:  # 触发计数报警
+                    if 0 < cls_count < high_count_warn:
+                        high_count_warn_num = cls_count
+                        high_count_warn_status = True
+
                 if len(target_location_back) > 0:  # 当前逻辑并不严谨，只是比较了第一个位置信息
                     des1_back = target_location_back[0]
                     des1_back_longitude = des1_back[0]
@@ -3336,6 +3352,11 @@ async def cal_des_invade(loop, invade_executor, task_id: str, mqtt, mqtt_publish
                             "box_count": [message_point],  # 特殊处理
                             "location_results": location_results  # 增加经纬度信息
                         }],
+                        "high_count_warn": {
+                            "high_count_warn_status": high_count_warn_status,
+                            "high_count_warn_num": high_count_warn_num,
+                            "high_count_warn": high_count_warn
+                        },
                         "osd_location": {
                             "longitude": cam_longitude,
                             "latitude": cam_latitude
@@ -3484,7 +3505,8 @@ async def send_frame_to_s3_mq(loop, upload_executor, task_id, mqtt, mqtt_topic,
             high_count_warn_status = False
             high_count_warn_num = 0
 
-            if target_point is not None and 0 < high_count_warn < len(target_point):  # 触发计数报警
+            if target_point is not None:  # 触发计数报警
+                if 0 < high_count_warn < len(target_point):
                     high_count_warn_num = len(target_point)
                     high_count_warn_status = True
 
@@ -4251,7 +4273,7 @@ async def start_rtmp_processing(video_url: str, task_id: str, model_configs: Lis
                         invade_queue,
                         event_queue,
                         device_height,
-                        repeat_dis, repeat_time
+                        repeat_dis, repeat_time, high_count_warn
                     ),
                     name="cal_des_invade"
                 )
@@ -4552,7 +4574,7 @@ async def start_video_processing(minio_path: str, task_id: str, model_configs: L
                     invade_queue,
                     event_queue,
                     device_height,
-                    repeat_dis, repeat_time
+                    repeat_dis, repeat_time, -1
                 ),
                 name="cal_des_invade"
             )
@@ -4562,7 +4584,8 @@ async def start_video_processing(minio_path: str, task_id: str, model_configs: L
         for _ in range(2):
             upload_task = asyncio.create_task(
                 send_frame_to_s3_mq(task_id, mqtt, mqtt_topic,
-                                    cancel_flag, cv_frame_queue, event_queue, device_height, repeat_dis, repeat_time),
+                                    cancel_flag, cv_frame_queue, event_queue, device_height, repeat_dis, repeat_time,
+                                    -1),
                 name=f"send_frame_to_s3_mq_{_}"
             )
             upload_tasks.append(upload_task)

yolo_api.py

@@ -718,9 +718,14 @@ async def run_back_Multi_Detect_async(request, request_json, stop_event: asyncio
         invade = request_json.content_body.invade
         invade_file = invade["invade_file"]
         camera_para_url = invade["camera_para_url"]
-        invade_switch = 0
+
-        if invade["invade_switch"] is not None:
+        if high_count_warn is None:
+            high_count_warn=0
+
+        if "invade_switch" in invade:
             invade_switch = invade["invade_switch"]
+        else:
+            invade_switch = 0  # 或其他默认值
         # dao.get_mqtt_config_by_orgcode(org_code,)
         str_request = str(request) + "&" + str(request.socket)  # 待测试，看看公网能不能捕获到请求端ip
         dao.insert_request_log(task_id, sn, org_code, str(request.body), str_request)
@@ -954,9 +959,10 @@ async def run_back_Video_Multi_Detect_async(request, request_json):
         invade = request_json.content_body.invade
         invade_file = invade["invade_file"]
         camera_para_url = invade["camera_para_url"]
-        invade_switch = 0
+        if "invade_switch" in invade:
-        if invade["invade_switch"] is not None:
             invade_switch = invade["invade_switch"]
+        else:
+            invade_switch = 0  # 或其他默认值
         await start_video_processing(minio_file_path, task_id, model_configs, mqtt_pub_ip, mqtt_pub_port,
                                      mqtt_pub_topic, push_url,
                                      invade_enable,invade_switch, invade_file, camera_para_url, device_height, repeat_dis,