import os import sys import tempfile from pathlib import Path import cv2 import numpy as np os.environ["SKIP_MODEL_LOAD"] = "1" ROOT = Path(__file__).resolve().parents[1] sys.path.insert(0, str(ROOT)) import app # noqa: E402 def write_video(path, frames, fps=12): height, width = frames[0].shape[:2] writer = cv2.VideoWriter(str(path), cv2.VideoWriter_fourcc(*"mp4v"), fps, (width, height)) for frame in frames: writer.write(cv2.cvtColor(frame, cv2.COLOR_RGB2BGR)) writer.release() def solid_frames(count=36, size=96, color=(30, 30, 30)): return [np.full((size, size, 3), color, dtype=np.uint8) for _ in range(count)] def moving_square_frames(count=36, size=96): frames = solid_frames(count=count, size=size) for i in range(18, count): x = 8 + (i - 18) * 3 cv2.rectangle(frames[i], (x, 36), (x + 16, 52), (230, 230, 230), -1) return frames def flicker_frames(count=36, size=96): frames = solid_frames(count=count, size=size, color=(40, 40, 40)) for i in range(18, count): frames[i][:] = (120, 120, 120) return frames def particle_frames(count=36, size=96): rng = np.random.default_rng(123) frames = solid_frames(count=count, size=size, color=(20, 20, 25)) for i in range(16, count): for _ in range(18): x = int(rng.integers(8, size - 8)) y = int((i * 5 + rng.integers(0, size)) % size) cv2.circle(frames[i], (x, y), 2, (220, 220, 240), -1) return frames def analyze_frames(frames, tmp_path, sensitivity=0.55): video_path = tmp_path / "sample.mp4" write_video(video_path, frames) samples = app.read_video_samples(str(video_path), max_duration_seconds=5, sample_fps=4) series = app.compute_temporal_series(samples) events = app.select_events(series, sensitivity=sensitivity, max_events=5) return samples, series, events def test_no_change_returns_no_major_events(): with tempfile.TemporaryDirectory() as tmp: _, _, events = analyze_frames(solid_frames(), Path(tmp), sensitivity=0.8) assert len(events) == 0 def test_single_motion_event_detects_near_known_timestamp(): with tempfile.TemporaryDirectory() as tmp: _, _, events = analyze_frames(moving_square_frames(), Path(tmp), sensitivity=0.45) assert events assert any(1.2 <= event["timestamp"] <= 2.4 for event in events) assert any("Motion" in event["evidence_type"] or "Appearance" in event["evidence_type"] for event in events) def test_lighting_flicker_is_labeled_as_intensity_change(): with tempfile.TemporaryDirectory() as tmp: _, _, events = analyze_frames(flicker_frames(), Path(tmp), sensitivity=0.45) assert events assert any(event["evidence_type"] == "Lighting / exposure change" for event in events) def test_particle_motion_produces_high_flow_or_diff_score(): with tempfile.TemporaryDirectory() as tmp: _, _, events = analyze_frames(particle_frames(), Path(tmp), sensitivity=0.45) assert events assert max(event["score"] for event in events) > 0.3