From 683791d1c98ba9cc582a92541e95e4fbde9c8bd0 Mon Sep 17 00:00:00 2001 From: 16337 <1633794139@qq.com> Date: Fri, 13 Feb 2026 09:50:53 +0800 Subject: [PATCH] fix(service): Alarm creation without duration - aligns with ai_edge changes Changes: 1. Modified create_from_mqtt to parse first_frame_time from MQTT data 2. Removed duration_minutes processing logic 3. Set duration_ms=None and last_frame_time=None on alarm creation 4. Updated _determine_alarm_level to handle duration_ms=None (returns level 2 for leave_post) This ensures alarms are created with status=NEW and no duration/end time, which will be populated later when the alarm is resolved. Test: test_alarm_create_no_duration.py validates the new behavior. Related: Task 2 of alarm status management fix Co-Authored-By: Claude Sonnet 4.5 --- app/services/alarm_event_service.py | 31 +++++++++++------ test_alarm_create_no_duration.py | 53 +++++++++++++++++++++++++++++ 2 files changed, 74 insertions(+), 10 deletions(-) create mode 100644 test_alarm_create_no_duration.py diff --git a/app/services/alarm_event_service.py b/app/services/alarm_event_service.py index 7cbbd02..34d4b05 100644 --- a/app/services/alarm_event_service.py +++ b/app/services/alarm_event_service.py @@ -28,9 +28,14 @@ def _determine_alarm_level(alarm_type: str, confidence: float, duration_ms: Opti if alarm_type == "intrusion": return 3 # 严重 elif alarm_type == "leave_post": - if duration_ms and duration_ms > 30 * 60 * 1000: + # 告警触发时 duration_ms 为 None，设置为一般级别 + if duration_ms is None: + return 2 # 一般级别（刚触发，持续时长未知） + + # 根据持续时长判断级别 + if duration_ms > 30 * 60 * 1000: return 3 # 严重 - elif duration_ms and duration_ms > 10 * 60 * 1000: + elif duration_ms > 10 * 60 * 1000: return 2 # 一般 return 1 # 提醒 elif confidence and confidence > 0.9: @@ -63,6 +68,15 @@ class AlarmEventService: else: event_time = datetime.now(timezone.utc) + # 解析 first_frame_time（告警首次触发时间） + first_frame_str = mqtt_data.get("first_frame_time") + first_frame_time = None + if first_frame_str: + try: + first_frame_time = datetime.fromisoformat(first_frame_str.replace("Z", "+00:00")) + except ValueError: + pass + # 置信度保持 float 0-1 confidence = mqtt_data.get("confidence") if confidence is not None: @@ -71,14 +85,9 @@ class AlarmEventService: if confidence > 1: confidence = confidence / 100.0 - # duration_minutes → duration_ms - duration_minutes = mqtt_data.get("duration_minutes") - duration_ms = None - if duration_minutes is not None: - duration_ms = int(float(duration_minutes) * 60 * 1000) - alarm_type = mqtt_data.get("alert_type", "unknown") - alarm_level = _determine_alarm_level(alarm_type, confidence or 0, duration_ms) + # 告警创建时不传递 duration_ms，传递 None + alarm_level = _determine_alarm_level(alarm_type, confidence or 0, None) alarm = AlarmEvent( alarm_id=alarm_id, @@ -87,7 +96,9 @@ class AlarmEventService: device_id=mqtt_data.get("camera_id", "unknown"), scene_id=mqtt_data.get("roi_id"), event_time=event_time, - duration_ms=duration_ms, + first_frame_time=first_frame_time, + duration_ms=None, + last_frame_time=None, alarm_level=alarm_level, confidence_score=confidence, alarm_status="NEW", diff --git a/test_alarm_create_no_duration.py b/test_alarm_create_no_duration.py new file mode 100644 index 0000000..bee751f --- /dev/null +++ b/test_alarm_create_no_duration.py @@ -0,0 +1,53 @@ +""" +Test alarm creation without duration +Verify: +1. Alarm creation with duration_ms as None +2. last_frame_time should be None +3. first_frame_time should have value +4. alarm_status should be NEW +5. handle_status should be UNHANDLED +""" +import sys +sys.path.insert(0, 'C:/Users/16337/PycharmProjects/service') + +from app.services.alarm_event_service import AlarmEventService + +def test_create_alarm_no_duration(): + """Test creating alarm without duration""" + service = AlarmEventService() + + mqtt_data = { + "alert_type": "leave_post", + "camera_id": "cam001", + "roi_id": "roi001", + "timestamp": "2026-02-12 14:30:00", + "first_frame_time": "2026-02-12 14:23:00", + "message": "Leave post alarm", + "confidence": 0.95, + # No longer sending duration_minutes + } + + print("Creating alarm test...") + alarm = service.create_from_mqtt(mqtt_data) + + # Verify + assert alarm is not None, "Alarm creation failed" + assert alarm.alarm_status == "NEW", f"Status should be NEW, but got {alarm.alarm_status}" + assert alarm.handle_status == "UNHANDLED", f"Handle status should be UNHANDLED, but got {alarm.handle_status}" + assert alarm.duration_ms is None, f"duration_ms should be None, but got {alarm.duration_ms}" + assert alarm.last_frame_time is None, f"last_frame_time should be None, but got {alarm.last_frame_time}" + assert alarm.first_frame_time is not None, "first_frame_time should have value" + assert alarm.alarm_level == 2, f"alarm_level should be 2 (general level), but got {alarm.alarm_level}" + + print("[PASS] Test passed: Alarm created without duration") + print(f" alarm_id: {alarm.alarm_id}") + print(f" alarm_type: {alarm.alarm_type}") + print(f" alarm_status: {alarm.alarm_status}") + print(f" alarm_level: {alarm.alarm_level}") + print(f" duration_ms: {alarm.duration_ms}") + print(f" first_frame_time: {alarm.first_frame_time}") + print(f" last_frame_time: {alarm.last_frame_time}") + print(f" event_time: {alarm.event_time}") + +if __name__ == "__main__": + test_create_alarm_no_duration()