diff --git a/analysis/config.py b/analysis/config.py index 4d9c525..61651ad 100644 --- a/analysis/config.py +++ b/analysis/config.py @@ -32,7 +32,7 @@ MOTION_SHORT_CLIP_SEC = 2.0 # 短于此时长的视频（约占全量一半 MOTION_PIXEL_RATIO = 0.015 # 一帧里前景像素（不含阴影）占比超过此值才算"有活动" MOTION_MERGE_GAP_SEC = 1.5 # 两段活动间隔小于此秒数则合并为一段 MOTION_MIN_DURATION_SEC = 0.5 # 短于此秒数的活动段视为噪声，丢弃 -MOTION_SAMPLE_STRIDE = 1 # 每隔几帧采样一次（1=逐帧，长视频可调大提速） +MOTION_SAMPLE_STRIDE = 2 # 每隔几帧采样一次（2=隔帧，长视频提速且不影响活动窗口检测精度） # 人物出现次数统计（YOLO11n 检测 + ByteTrack 跟踪）参数 # 只在 motion_filter 找出的活动窗口内跑，空场景直接跳过，省算力。 diff --git a/tests/test_core.py b/tests/test_core.py new file mode 100644 index 0000000..dceae9b --- /dev/null +++ b/tests/test_core.py @@ -0,0 +1,135 @@ +"""基础单元测试：覆盖不依赖外部工具/模型的纯函数。 + +运行：.venv/bin/python -m pytest tests/ -v +或：.venv/bin/python tests/test_core.py +""" +import hashlib +import os +import tempfile +import unittest + +from analysis.hashing import compute_md5 +from analysis.motion_filter import merge_motion_windows +from analysis.fall_detector import _find_fall_windows +from analysis.activity_tagger import _merge_labeled_samples +from analysis.ollama_client import _parse_label + + +class TestHashing(unittest.TestCase): + def test_compute_md5_matches_hashlib(self): + with tempfile.NamedTemporaryFile(delete=False) as f: + f.write(b"hello world" * 1000) + path = f.name + try: + expected = hashlib.md5(b"hello world" * 1000).hexdigest() + self.assertEqual(compute_md5(path), expected) + finally: + os.unlink(path) + + def test_compute_md5_empty_file(self): + with tempfile.NamedTemporaryFile(delete=False) as f: + path = f.name + try: + self.assertEqual(compute_md5(path), hashlib.md5(b"").hexdigest()) + finally: + os.unlink(path) + + +class TestMergeMotionWindows(unittest.TestCase): + def test_single_window(self): + flags = [(0.0, True), (0.5, True), (1.0, True)] + self.assertEqual(merge_motion_windows(flags), [[0.0, 1.0]]) + + def test_merge_nearby_windows(self): + # 两段活动间隔 1.0s < MERGE_GAP 1.5s，应合并 + flags = [(0.0, True), (0.5, True), (2.0, True), (2.5, True)] + result = merge_motion_windows(flags) + self.assertEqual(len(result), 1) + self.assertEqual(result[0], [0.0, 2.5]) + + def test_separate_windows(self): + # 第一段 0-0.5s 活动，中间 1.0-3.5s 无活动（间隔3s > MERGE_GAP 1.5s），第二段 4.0-4.5s 活动 + flags = [(0.0, True), (0.5, True), (1.0, False), (2.0, False), + (3.0, False), (3.5, False), (4.0, True), (4.5, True)] + result = merge_motion_windows(flags) + self.assertEqual(len(result), 2) + + def test_filter_short_windows(self): + # 0.3s < MIN_DURATION 0.5s，应被过滤 + flags = [(0.0, True), (0.3, True)] + self.assertEqual(merge_motion_windows(flags), []) + + def test_empty_input(self): + self.assertEqual(merge_motion_windows([]), []) + + +class TestFallDetection(unittest.TestCase): + def test_upright_to_prone_fast(self): + # 直立(0s) → 1秒后横躺(1s) → 持续到3s，应检测为摔倒 + states = [(0.0, "upright"), (1.0, "prone"), (2.0, "prone"), (3.0, "prone")] + windows = _find_fall_windows(states) + self.assertEqual(len(windows), 1) + self.assertEqual(windows[0], (1.0, 3.0)) + + def test_slow_prone_not_fall(self): + # 直立(0s) → 3秒后才横躺(3s) > TRANSITION_MAX 1.5s，不是摔倒 + states = [(0.0, "upright"), (3.0, "prone"), (4.0, "prone")] + self.assertEqual(_find_fall_windows(states), []) + + def test_prone_without_prior_upright(self): + # 视频开头就是横躺，没有直立前置，不算摔倒 + states = [(0.0, "prone"), (1.0, "prone"), (2.0, "prone")] + self.assertEqual(_find_fall_windows(states), []) + + def test_prone_too_short(self): + # 横躺只持续 0.5s < SUSTAIN_MIN 1.0s，不算摔倒 + states = [(0.0, "upright"), (0.5, "prone"), (1.0, "upright")] + self.assertEqual(_find_fall_windows(states), []) + + def test_unknown_states_ignored(self): + states = [(0.0, "upright"), (0.5, "unknown"), (1.0, "prone"), + (2.0, "prone"), (3.0, "prone")] + windows = _find_fall_windows(states) + self.assertEqual(len(windows), 1) + + +class TestActivityMerge(unittest.TestCase): + def test_merge_same_label(self): + samples = [(0.0, "走动", "raw1"), (2.0, "走动", "raw2")] + segments = _merge_labeled_samples(samples) + self.assertEqual(len(segments), 1) + self.assertEqual(segments[0]["label"], "走动") + self.assertEqual(segments[0]["start"], 0.0) + self.assertEqual(segments[0]["end"], 2.0) + + def test_different_labels_separate(self): + samples = [(0.0, "走动", "raw1"), (5.0, "坐下", "raw2")] + segments = _merge_labeled_samples(samples) + self.assertEqual(len(segments), 2) + + def test_gap_too_large_separates(self): + # 间隔 5.0s > MERGE_GAP 3.0s，即使同标签也分开 + samples = [(0.0, "走动", "raw1"), (5.0, "走动", "raw2")] + segments = _merge_labeled_samples(samples) + self.assertEqual(len(segments), 2) + + +class TestOllamaParse(unittest.TestCase): + def test_exact_label(self): + self.assertEqual(_parse_label("走动"), "走动") + + def test_label_with_punctuation(self): + self.assertEqual(_parse_label("走动。"), "走动") + + def test_label_in_sentence(self): + self.assertEqual(_parse_label("这个人在坐下休息"), "坐下") + + def test_unknown_fallback(self): + self.assertEqual(_parse_label("xyz 不存在的标签"), "其他") + + def test_empty_string(self): + self.assertEqual(_parse_label(""), "其他") + + +if __name__ == "__main__": + unittest.main() diff --git a/webapp/app.py b/webapp/app.py index 3ee14ba..81a7925 100644 --- a/webapp/app.py +++ b/webapp/app.py @@ -21,6 +21,8 @@ def index(): def api_events(): event_type = request.args.get("event_type") capture_date = request.args.get("date") # YYYYMMDD + limit = min(int(request.args.get("limit", 500)), 2000) + offset = max(int(request.args.get("offset", 0)), 0) query = """ SELECT e.id, e.video_id, e.event_type, e.label, e.start_offset_sec, @@ -36,7 +38,8 @@ def api_events(): if capture_date: query += " AND v.capture_date = ?" params.append(capture_date) - query += " ORDER BY e.timestamp_abs DESC, e.id DESC LIMIT 500" + query += " ORDER BY e.timestamp_abs DESC, e.id DESC LIMIT ? OFFSET ?" + params.extend([limit, offset]) with get_connection() as conn: rows = conn.execute(query, params).fetchall()