功能:新增车辆违停和车辆拥堵检测算法

- IllegalParkingAlgorithm: 5状态机(IDLE→CONFIRMING_VEHICLE→PARKED_COUNTDOWN→ALARMED→CONFIRMING_CLEAR)
  禁停区域检测,15秒确认+5分钟倒计时,滑动窗口抗抖动,支持car/truck/bus/motorcycle
- VehicleCongestionAlgorithm: 4状态机(NORMAL→CONFIRMING_CONGESTION→CONGESTED→CONFIRMING_CLEAR)
  车辆计数≥阈值+持续60秒触发,滑动窗口平均值判断
- AlgorithmManager: 新增default_params、register_algorithm、get_status支持两种新算法
- main.py: 泛化alarm_id回填和first_frame_time提取,ext_data新增vehicle_count字段

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
2026-03-16 16:54:47 +08:00
parent 9a1ac16f19
commit ea992c6daa
2 changed files with 536 additions and 28 deletions

32
main.py
View File

@@ -763,12 +763,16 @@ class EdgeInferenceService:
self._camera_alert_cooldown[dedup_key] = now
self._performance_stats["total_alerts_generated"] += 1
# 获取算法的离岗开始时间
leave_start_time = None
if alert_type == "leave_post":
algo = self._algorithm_manager.algorithms.get(roi_id, {}).get(f"{roi_id}_{bind.bind_id}", {}).get("leave_post")
if algo and hasattr(algo, '_leave_start_time') and algo._leave_start_time:
leave_start_time = algo._leave_start_time.isoformat()
# 获取算法的事件开始时间(泛化:支持所有算法类型)
first_frame_time = None
algo = self._algorithm_manager.algorithms.get(roi_id, {}).get(f"{roi_id}_{bind.bind_id}", {}).get(alert_type)
if algo:
# 各算法使用不同的内部变量名存储开始时间
for attr in ('_leave_start_time', '_parking_start_time', '_congestion_start_time', '_intrusion_start_time'):
val = getattr(algo, attr, None)
if val:
first_frame_time = val.isoformat()
break
from core.result_reporter import AlarmInfo, generate_alarm_id
alarm_info = AlarmInfo(
@@ -788,20 +792,16 @@ class EdgeInferenceService:
"bind_id": bind.bind_id,
"message": alert.get("message", ""),
"edge_node_id": self._settings.mqtt.device_id,
"first_frame_time": leave_start_time,
"first_frame_time": first_frame_time,
"vehicle_count": alert.get("vehicle_count"),
},
)
self._reporter.report_alarm(alarm_info, screenshot=frame.image)
# 回填 alarm_id 到算法实例(用于后续 resolve 追踪)
if alert_type == "leave_post":
algo = self._algorithm_manager.algorithms.get(roi_id, {}).get(f"{roi_id}_{bind.bind_id}", {}).get("leave_post")
if algo and hasattr(algo, 'set_last_alarm_id'):
algo.set_last_alarm_id(alarm_info.alarm_id)
elif alert_type == "intrusion":
algo = self._algorithm_manager.algorithms.get(roi_id, {}).get(f"{roi_id}_{bind.bind_id}", {}).get("intrusion")
if algo and hasattr(algo, 'set_last_alarm_id'):
algo.set_last_alarm_id(alarm_info.alarm_id)
# 回填 alarm_id 到算法实例(用于后续 resolve 追踪,泛化支持所有算法类型
algo = self._algorithm_manager.algorithms.get(roi_id, {}).get(f"{roi_id}_{bind.bind_id}", {}).get(alert_type)
if algo and hasattr(algo, 'set_last_alarm_id'):
algo.set_last_alarm_id(alarm_info.alarm_id)
self._logger.info(
f"告警已生成: type={alert_type}, "