security-ai-edge/test_leave_post_full_workflow.py

"""
离岗告警完整流程集成测试

测试完整业务流程：
1. 人员上岗确认
2. 人员离开ROI
3. 离岗确认 → OFF_DUTY_COUNTDOWN
4. 倒计时结束 → ALARMED (验证告警无 duration_minutes，有 first_frame_time)
5. 创建告警记录 (验证 duration_ms=None)
6. 人员返回 → CONFIRMING_ON_DUTY
7. 回岗确认 → ON_DUTY (验证发送 resolve 事件)
8. 更新告警记录 (验证 duration_ms 已填充)
"""
import sys
sys.path.insert(0, 'C:/Users/16337/PycharmProjects/ai_edge')

from algorithms import LeavePostAlgorithm
from datetime import datetime, timedelta


def test_leave_post_full_workflow():
    """测试离岗告警完整工作流"""
    print("=" * 80)
    print("Leave Post Alarm Full Workflow Integration Test")
    print("=" * 80)

    algo = LeavePostAlgorithm(
        confirm_on_duty_sec=2,      # 2秒上岗确认
        confirm_off_duty_sec=3,     # 3秒离岗确认
        leave_countdown_sec=5,      # 5秒离岗倒计时
        confirm_return_sec=2,       # 2秒回岗确认
    )

    roi_id = "test_roi_001"
    camera_id = "test_camera_001"
    person_track = [{"matched_rois": [{"roi_id": roi_id}], "class": "person", "bbox": [100, 100, 200, 200]}]

    # 测试起始时间
    t0 = datetime(2026, 2, 13, 14, 0, 0)

    # 存储告警数据（模拟数据库）
    alarm_db = {}

    try:
        print("\n" + "=" * 80)
        print("Stage 1: Person On Duty Confirmation")
        print("=" * 80)
        print("Detecting person for 3 seconds (exceeds confirm_on_duty_sec=2s)")

        for i in range(30):  # 每100ms一帧，共3秒
            t = t0 + timedelta(milliseconds=i*100)
            alerts = algo.process(roi_id, camera_id, person_track, t)
            if alerts:
                print(f"  [!] 意外告警: {alerts}")

        print(f"  State: {algo.state}")
        assert algo.state == "ON_DUTY", f"Expected ON_DUTY, got {algo.state}"
        print("  [OK] Person on duty confirmed")

        print("\n" + "=" * 80)
        print("Stage 2: Person Leaves ROI")
        print("=" * 80)
        print("Sending empty frames, waiting for window expiry (10s) + leave confirmation (3s)")

        t_leave = t0 + timedelta(seconds=3)
        first_frame_time_expected = None

        # 发送空帧让检测窗口清空，然后进入离岗确认
        for i in range(140):  # 每100ms一帧，共14秒
            t = t_leave + timedelta(milliseconds=i*100)
            alerts = algo.process(roi_id, camera_id, [], t)

            if algo.state == "CONFIRMING_OFF_DUTY" and first_frame_time_expected is None:
                print(f"  State change detected: ON_DUTY -> CONFIRMING_OFF_DUTY (time: {t})")

            if algo.state == "OFF_DUTY_COUNTDOWN":
                if first_frame_time_expected is None:
                    first_frame_time_expected = t
                    print(f"  State change detected: CONFIRMING_OFF_DUTY -> OFF_DUTY_COUNTDOWN")
                    print(f"  Leave confirmed, recording leave time: {first_frame_time_expected}")
                break

        print(f"  State: {algo.state}")
        assert algo.state == "OFF_DUTY_COUNTDOWN", f"Expected OFF_DUTY_COUNTDOWN, got {algo.state}"
        print("  [OK] Leave confirmed, countdown started")

        print("\n" + "=" * 80)
        print("Stage 3: Leave Countdown")
        print("=" * 80)
        print("Continuing empty frames, waiting for countdown (5s)")

        t_countdown = t_leave + timedelta(seconds=14)
        alarm_triggered = None

        for i in range(60):  # 每100ms一帧，共6秒 (确保超过倒计时)
            t = t_countdown + timedelta(milliseconds=i*100)
            alerts = algo.process(roi_id, camera_id, [], t)

            if alerts:
                for alert in alerts:
                    if alert.get("alert_type") == "leave_post":
                        alarm_triggered = alert
                        print(f"  [!] Alarm triggered: {t}")
                        print(f"      alert_type: {alert['alert_type']}")
                        print(f"      message: {alert['message']}")
                        print(f"      first_frame_time: {alert.get('first_frame_time')}")
                        print(f"      duration_minutes: {alert.get('duration_minutes', 'NOT_PRESENT')}")
                        break

            if algo.state == "ALARMED":
                print(f"  State: {algo.state}")
                break

        assert algo.state == "ALARMED", f"Expected ALARMED, got {algo.state}"
        assert alarm_triggered is not None, "Alarm should be triggered"
        assert "duration_minutes" not in alarm_triggered, "Alarm should NOT contain duration_minutes"
        assert "first_frame_time" in alarm_triggered, "Alarm should contain first_frame_time"
        print("  [OK] Alarm triggered successfully, NO duration_minutes field")

        print("\n" + "=" * 80)
        print("Stage 4: Create Alarm Record (Simulated DB)")
        print("=" * 80)
        # Simulate saving alarm to database
        import uuid
        alarm_id = f"ALM{datetime.now().strftime('%Y%m%d%H%M%S')}{uuid.uuid4().hex[:4].upper()}"

        # Parse first_frame_time
        first_frame_time_str = alarm_triggered.get("first_frame_time")
        first_frame_time_dt = datetime.strptime(first_frame_time_str, '%Y-%m-%d %H:%M:%S')

        # Simulated alarm record
        alarm_db[alarm_id] = {
            "alarm_id": alarm_id,
            "alarm_type": "leave_post",
            "device_id": camera_id,
            "scene_id": roi_id,
            "event_time": t,  # Alarm trigger time
            "first_frame_time": first_frame_time_dt,  # Leave time
            "last_frame_time": None,  # Not returned yet
            "duration_ms": None,  # Duration unknown
            "alarm_status": "NEW",
            "handle_status": "UNHANDLED",
        }

        # Set alarm_id in algorithm
        algo.set_last_alarm_id(alarm_id)

        alarm = alarm_db[alarm_id]
        print(f"  Alarm ID: {alarm_id}")
        print(f"  event_time: {alarm['event_time']}")
        print(f"  first_frame_time: {alarm['first_frame_time']}")
        print(f"  last_frame_time: {alarm['last_frame_time']} (NULL)")
        print(f"  duration_ms: {alarm['duration_ms']} (NULL)")
        assert alarm["duration_ms"] is None, "duration_ms should be None"
        assert alarm["last_frame_time"] is None, "last_frame_time should be None"
        print("  [OK] Alarm created, duration_ms=NULL, last_frame_time=NULL")

        print("\n" + "=" * 80)
        print("Stage 5: Person Returns to ROI")
        print("=" * 80)
        print("Sending person detection frames, waiting for return confirmation (2s)")

        t_return = t_countdown + timedelta(seconds=7)  # Returned 1s after alarm

        # Send enough person frames to fill detection window and exceed confirm_return_sec
        for i in range(130):  # 100ms per frame, 13s total (way more than needed)
            t = t_return + timedelta(milliseconds=i*100)
            alerts = algo.process(roi_id, camera_id, person_track, t)

            # Check for resolve event
            if alerts:
                for alert in alerts:
                    if alert.get("alert_type") == "alarm_resolve":
                        resolve_event = alert
                        print(f"  [!] Resolve event triggered: {t}")
                        print(f"      alert_type: {alert['alert_type']}")
                        print(f"      resolve_alarm_id: {alert['resolve_alarm_id']}")
                        print(f"      duration_ms: {alert['duration_ms']}")
                        print(f"      last_frame_time: {alert['last_frame_time']}")
                        print(f"      resolve_type: {alert['resolve_type']}")
                        break

            if algo.state == "CONFIRMING_ON_DUTY" and i < 10:
                if i == 0:
                    print(f"  State change detected: ALARMED -> CONFIRMING_ON_DUTY (time: {t})")

            if algo.state == "ON_DUTY":
                print(f"  State change detected: CONFIRMING_ON_DUTY -> ON_DUTY (time: {t})")
                print(f"  State: {algo.state}")
                break

        assert algo.state == "ON_DUTY", f"Expected ON_DUTY, got {algo.state}"
        assert resolve_event is not None, "Resolve event should be triggered"
        assert resolve_event["resolve_alarm_id"] == alarm_id, "resolve_alarm_id mismatch"
        assert resolve_event["duration_ms"] > 0, "duration_ms should be > 0"
        assert resolve_event["resolve_type"] == "person_returned", "resolve_type should be person_returned"
        print("  [OK] Return confirmed, resolve event sent")

        print("\n" + "=" * 80)
        print("Stage 7: Update Alarm Record (Simulated resolve_alarm)")
        print("=" * 80)

        # Simulate resolve_alarm update
        alarm["duration_ms"] = resolve_event["duration_ms"]
        alarm["last_frame_time"] = datetime.strptime(resolve_event["last_frame_time"], '%Y-%m-%d %H:%M:%S')
        alarm["alarm_status"] = "CLOSED"
        alarm["handle_status"] = "DONE"
        alarm["handle_remark"] = "Person returned - auto closed"

        print(f"  Alarm ID: {alarm_id}")
        print(f"  duration_ms: {alarm['duration_ms']} ({alarm['duration_ms']/1000:.1f}s)")
        print(f"  last_frame_time: {alarm['last_frame_time']}")
        print(f"  alarm_status: {alarm['alarm_status']}")
        print(f"  handle_status: {alarm['handle_status']}")
        print(f"  handle_remark: {alarm['handle_remark']}")

        assert alarm["duration_ms"] is not None, "duration_ms should be filled"
        assert alarm["duration_ms"] > 0, "duration_ms should be > 0"
        assert alarm["last_frame_time"] is not None, "last_frame_time should be filled"
        assert alarm["alarm_status"] == "CLOSED", f"alarm_status should be CLOSED, got {alarm['alarm_status']}"
        assert alarm["handle_status"] == "DONE", f"handle_status should be DONE, got {alarm['handle_status']}"
        print("  [OK] Alarm record updated, duration_ms filled, status closed")

        print("\n" + "=" * 80)
        print("Stage 8: Verify Complete Workflow")
        print("=" * 80)

        # Verify duration calculation
        expected_duration_ms = int((datetime.strptime(resolve_event["last_frame_time"], '%Y-%m-%d %H:%M:%S') - first_frame_time_dt).total_seconds() * 1000)
        actual_duration_ms = alarm["duration_ms"]

        print(f"  Expected duration_ms: {expected_duration_ms}ms")
        print(f"  Actual duration_ms: {actual_duration_ms}ms")
        print(f"  Difference: {abs(expected_duration_ms - actual_duration_ms)}ms")

        # Allow 1 second error (1000ms)
        assert abs(expected_duration_ms - actual_duration_ms) <= 1000, "duration_ms calculation error too large"

        print("  [OK] duration_ms calculation correct")

        print("\n" + "=" * 80)
        print("SUCCESS: Full workflow test passed")
        print("=" * 80)
        print("\nWorkflow Summary:")
        print(f"  1. On duty time: {t0}")
        print(f"  2. Leave time: {first_frame_time_dt}")
        print(f"  3. Alarm time: {alarm['event_time']}")
        print(f"  4. Return time: {alarm['last_frame_time']}")
        print(f"  5. Leave duration: {alarm['duration_ms']/1000:.1f}s ({alarm['duration_ms']/60000:.1f}min)")
        print(f"  6. Alarm status: {alarm['alarm_status']}")
        print(f"  7. Handle status: {alarm['handle_status']}")

        print("\n  [OK] Test completed successfully")

    except Exception as e:
        print(f"\n" + "=" * 80)
        print(f"FAILED: Test failed")
        print("=" * 80)
        print(f"Error: {e}")
        import traceback
        traceback.print_exc()
        raise


if __name__ == "__main__":
    test_leave_post_full_workflow()