perf: webapp事件接口加分页、motion默认隔帧采样、新增20个单元测试
- /api/events 支持 limit/offset 分页参数,默认500上限2000 - MOTION_SAMPLE_STRIDE 默认从1改为2,长视频提速一倍 - 新增 tests/test_core.py:覆盖MD5、活动窗口合并、摔倒检测、活动标签合并、Ollama解析等20个纯函数测试
This commit is contained in:
@@ -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()
|
||||
Reference in New Issue
Block a user