From e592389dc466461fd134d3d9ec6ddbe968b46cde Mon Sep 17 00:00:00 2001 From: "DESKTOP-58UIFAP\\user" Date: Fri, 17 Jul 2026 17:31:12 +0900 Subject: [PATCH] fix: decouple realtime alerts from snapshots --- ai/events/clip_worker.py | 19 +++-- scripts/serve_ai_overlay.py | 23 ------ tests/test_event_clip.py | 103 +++++++++++++++++++++++- tests/test_primary_snapshot_decouple.py | 53 +++++++++++- 4 files changed, 165 insertions(+), 33 deletions(-) diff --git a/ai/events/clip_worker.py b/ai/events/clip_worker.py index 985836b..268be09 100644 --- a/ai/events/clip_worker.py +++ b/ai/events/clip_worker.py @@ -404,6 +404,16 @@ def save_clip_to_mp4(task): raise RuntimeError(f"encoded clip is empty: {output_path}") return output_path +def _stable_event_id(metadata): + if not isinstance(metadata, dict): + return None + event_id = metadata.get("eventId") or metadata.get("event_id") + if event_id is None: + return None + event_id = str(event_id).strip() + return event_id or None + + #ClipWriterWorker는 태스크를 큐에서 꺼내어 관리하는 작업 스레드의 역할만 담당하며, 실제 비디오 파일로 인코딩하여 기록하는 연산은 save_clip_to_mp4 함수가 처리하는 구조 class ClipWriterWorker: def __init__(self, task_queue, uploader=upload_clip, publisher=None, mqtt_event_topic=None, snapshot_uploader=upload_snapshot_jpeg): @@ -435,7 +445,7 @@ def _run(self): try: upload_result = self.uploader(output_path, task.metadata) meta = task.metadata or {} - event_id = meta.get("eventId") or meta.get("event_id") or meta.get("evidenceId") + event_id = _stable_event_id(meta) clip_key = upload_result.get("s3_key") if upload_result else None clip_ok = bool(upload_result and upload_result.get("uploaded") and upload_result.get("url")) duration = len(task.frames) / float(task.fps or 30.0) @@ -485,12 +495,7 @@ def _run(self): if upload_result and upload_result.get("uploaded") and upload_result.get("url") and self.publisher: s3_url = upload_result["url"] meta = task.metadata or {} - # Prefer stable incident eventId from metadata; never use timestamp as id. - event_id = ( - meta.get("eventId") - or meta.get("event_id") - or meta.get("evidenceId") - ) + event_id = _stable_event_id(meta) if not event_id: print( f"[clip-worker] skip MQTT re-publish: missing eventId for camera={task.camera_id}", diff --git a/scripts/serve_ai_overlay.py b/scripts/serve_ai_overlay.py index e84b7c7..4651d65 100644 --- a/scripts/serve_ai_overlay.py +++ b/scripts/serve_ai_overlay.py @@ -38,7 +38,6 @@ update_prediction_counts, update_tracking_summary, lifecycle_payload_kwargs, - snapshot_assist_meta_from_event_payload, ) from ai.inference.rtsp_runtime import ( log_detection_stage, @@ -1061,23 +1060,6 @@ def _process_frame_impl( try: event_publish_result = _publish_event(publisher, payload, topic_settings["event_topic"]) _record_publish_outcome(summary, "events_publish", event_publish_result, attempted=True) - # Single VLM snapshot assist hook (never blocks alert loop) - try: - from ai.snapshot_assist_upload import submit_frame_snapshot_async - assist_meta = snapshot_assist_meta_from_event_payload( - payload, - track_id=track_id, - prediction=track_prediction, - emit_decision=emit_decision, - ) - submit_frame_snapshot_async( - event_id=str(payload.get("eventId") or ""), - camera_login_id=str(stream_id), - frame=getattr(frame_packet, "frame", None), - **assist_meta, - ) - except Exception as snap_exc: - print(f"[snapshot-assist][warn] non-fatal upload schedule failed: {snap_exc}", flush=True) except Exception as exc: _record_publish_outcome(summary, "events_publish", attempted=True) print(f"[ai-worker][error] failed to publish event payload for camera={stream_id}: {exc}", file=sys.stderr, flush=True) @@ -1099,11 +1081,6 @@ def _process_frame_impl( "confidence": payload.get("confidence"), "faint_prob": payload.get("faint_prob") or payload.get("faintProb"), "consecutive_count": payload.get("consecutive_count") or payload.get("consecutiveCount"), - # Primary snapshot key must survive clip re-publish so back can attach Snapshot. - "snapshot_object_key": payload.get("snapshot_object_key") or payload.get("snapshotObjectKey"), - "snapshotObjectKey": payload.get("snapshotObjectKey") or payload.get("snapshot_object_key"), - "snapshot_url": payload.get("snapshot_url") or payload.get("snapshotUrl"), - "snapshotUrl": payload.get("snapshotUrl") or payload.get("snapshot_url"), } clip_triggered = state.clip_buffer.trigger_event( diff --git a/tests/test_event_clip.py b/tests/test_event_clip.py index 5b1a2b1..3e296ce 100644 --- a/tests/test_event_clip.py +++ b/tests/test_event_clip.py @@ -3,10 +3,11 @@ import time import unittest from pathlib import Path +from unittest.mock import patch import numpy as np -from ai.events.clip_worker import ClipWriterWorker, enqueue_event_clip, save_clip_to_mp4 +from ai.events.clip_worker import ClipWriterWorker, _stable_event_id, enqueue_event_clip, save_clip_to_mp4 from ai.events.event_clip import CircularFrameBuffer, EventClipBuffer, EventClipTask @@ -82,9 +83,14 @@ def test_clip_worker_prefers_event_id_not_timestamp(self): "confidence": 0.88, "faint_prob": 0.88, } - event_id = meta.get("eventId") or meta.get("event_id") or meta.get("evidenceId") + event_id = _stable_event_id(meta) self.assertEqual(event_id, "stable-evt-1") self.assertNotEqual(event_id, meta["event_timestamp"]) + + def test_clip_worker_rejects_evidence_id_as_event_id_fallback(self): + self.assertIsNone(_stable_event_id({"evidenceId": "derived-id", "event_timestamp": 1_700_000_000.0})) + self.assertIsNone(_stable_event_id({"eventId": " "})) + self.assertEqual(_stable_event_id({"event_id": " stable-evt-2 "}), "stable-evt-2") def test_clip_worker_passes_snapshot_object_key_from_metadata(self): meta = { "eventId": "stable-evt-snap", @@ -213,6 +219,99 @@ def test_save_clip_to_mp4_keeps_legacy_glob_with_evidence_suffix(self): self.assertTrue(output_path.name.startswith("Fall_cam_01_")) self.assertIn("evidence-cam_01-7-2000", output_path.name) + def test_worker_publishes_clip_and_thumbnail_with_stable_event_id(self): + published = [] + + class Publisher: + def publish_event(self, payload): + published.append(payload) + + with tempfile.TemporaryDirectory() as temp_dir: + output_path = Path(temp_dir) / "clip.mp4" + output_path.write_bytes(b"clip") + task_queue = queue.Queue(maxsize=1) + worker = ClipWriterWorker( + task_queue, + uploader=lambda path, metadata: { + "uploaded": True, + "s3_key": "clips/stable-evt.mp4", + "url": "https://example.test/clips/stable-evt.mp4", + }, + publisher=Publisher(), + snapshot_uploader=lambda jpeg, event_id, metadata: { + "uploaded": True, + "s3_key": f"snapshots/{event_id}.jpg", + }, + ) + task = EventClipTask( + event_type="Fall", + camera_id="cam_01", + frames=[dummy_frame(7)], + fps=10.0, + output_dir=temp_dir, + metadata={ + "eventId": "stable-evt", + "event_timestamp": "2026-07-17T00:00:00Z", + "event_frame_index": 0, + }, + ) + with patch("ai.events.clip_worker.save_clip_to_mp4", return_value=output_path): + worker.start() + self.assertTrue(enqueue_event_clip(task_queue, task)) + task_queue.join() + worker.stop() + + self.assertEqual(len(published), 1) + self.assertEqual(published[0]["eventId"], "stable-evt") + self.assertEqual(published[0]["clip_object_key"], "clips/stable-evt.mp4") + self.assertEqual(published[0]["snapshot_object_key"], "snapshots/stable-evt.jpg") + + def test_worker_snapshot_failure_still_publishes_clip_attachment(self): + published = [] + + class Publisher: + def publish_event(self, payload): + published.append(payload) + + def failing_snapshot_upload(*_args, **_kwargs): + raise RuntimeError("S3 put_object failed") + + with tempfile.TemporaryDirectory() as temp_dir: + output_path = Path(temp_dir) / "clip.mp4" + output_path.write_bytes(b"clip") + task_queue = queue.Queue(maxsize=1) + worker = ClipWriterWorker( + task_queue, + uploader=lambda path, metadata: { + "uploaded": True, + "s3_key": "clips/stable-evt.mp4", + "url": "https://example.test/clips/stable-evt.mp4", + }, + publisher=Publisher(), + snapshot_uploader=failing_snapshot_upload, + ) + task = EventClipTask( + event_type="Fall", + camera_id="cam_01", + frames=[dummy_frame(7)], + fps=10.0, + output_dir=temp_dir, + metadata={ + "eventId": "stable-evt", + "event_timestamp": "2026-07-17T00:00:00Z", + "event_frame_index": 0, + }, + ) + with patch("ai.events.clip_worker.save_clip_to_mp4", return_value=output_path): + worker.start() + self.assertTrue(enqueue_event_clip(task_queue, task)) + task_queue.join() + worker.stop() + + self.assertEqual(len(published), 1) + self.assertEqual(published[0]["eventId"], "stable-evt") + self.assertEqual(published[0]["clip_object_key"], "clips/stable-evt.mp4") + self.assertNotIn("snapshot_object_key", published[0]) def test_worker_failure_does_not_stop_worker_thread(self): with tempfile.TemporaryDirectory() as temp_dir: task_queue = queue.Queue(maxsize=2) diff --git a/tests/test_primary_snapshot_decouple.py b/tests/test_primary_snapshot_decouple.py index 0b851f1..4db7794 100644 --- a/tests/test_primary_snapshot_decouple.py +++ b/tests/test_primary_snapshot_decouple.py @@ -12,9 +12,12 @@ from __future__ import annotations +import ast import os +import time import unittest -from unittest.mock import patch +from pathlib import Path +from unittest.mock import Mock, patch import numpy as np @@ -35,6 +38,54 @@ def _dummy_frame(): return np.zeros((8, 8, 3), dtype="uint8") +class RealtimeEventPathTest(unittest.TestCase): + _SCRIPTS = ( + Path(__file__).parents[1] / "scripts" / "serve_ai_overlay.py", + Path(__file__).parents[1] / "scripts" / "run_rtsp_inference.py", + ) + + def test_realtime_scripts_do_not_call_snapshot_or_vlm_network_hooks(self): + forbidden_calls = { + "attach_primary_snapshot_if_enabled", + "submit_frame_snapshot_async", + "submit_snapshot", + "upload_snapshot_jpeg", + "enqueue_vlm", + } + for script in self._SCRIPTS: + tree = ast.parse(script.read_text(encoding="utf-8"), filename=str(script)) + called_names = set() + for node in ast.walk(tree): + if not isinstance(node, ast.Call): + continue + if isinstance(node.func, ast.Name): + called_names.add(node.func.id) + elif isinstance(node.func, ast.Attribute): + called_names.add(node.func.attr) + self.assertTrue(forbidden_calls.isdisjoint(called_names), f"{script.name}: {forbidden_calls & called_names}") + + def test_slow_s3_put_object_is_not_reached_before_initial_mqtt_publish(self): + from scripts.serve_ai_overlay import _publish_event + + s3_client = Mock() + s3_client.put_object.side_effect = lambda **_kwargs: time.sleep(0.2) + published = [] + + class Publisher: + def enqueue_event(self, payload, topic): + published.append((payload, topic)) + return True + + started = time.perf_counter() + result = _publish_event(Publisher(), {"eventId": "evt-now"}, "safety/events") + elapsed = time.perf_counter() - started + + self.assertTrue(result) + self.assertEqual(published, [({"eventId": "evt-now"}, "safety/events")]) + s3_client.put_object.assert_not_called() + self.assertLess(elapsed, 0.1) + + class StorageBucketConfigTest(unittest.TestCase): def test_legacy_bucket_name_remains_supported(self): with patch.dict(os.environ, {"S3_BUCKET_NAME": "legacy-bucket"}, clear=True):