Add QSC prompt and phase timings

This commit is contained in:
yangyl
2026-06-17 22:52:54 +08:00
parent ef0047af6d
commit 0150c1ab5c
6 changed files with 304 additions and 118 deletions

View File

@@ -1269,6 +1269,20 @@ class CliTests(unittest.TestCase):
self.assertEqual(folder_summary["processed_video_count"], 1)
self.assertEqual(folder_summary["failed_video_count"], 0)
self.assertEqual(folder_summary["event_counts"], {"queue_detected": 1})
phase_timings = json.loads(
(output_dir / "phase_timings.json").read_text(encoding="utf-8")
)
self.assertEqual(phase_timings["schema_version"], "phase-timings-v1")
for phase in (
"source_acquisition_seconds",
"video_probe_seconds",
"frame_sampling_seconds",
"clip_generation_seconds",
"inference_seconds",
"aggregation_seconds",
):
self.assertIn(phase, phase_timings["phases"])
self.assertGreaterEqual(phase_timings["phases"][phase], 0)
if __name__ == "__main__":

View File

@@ -104,6 +104,41 @@ class ResultParserTests(unittest.TestCase):
"2026-06-14 12:31:20",
)
def test_build_clip_result_preserves_qsc_events(self):
result = build_clip_result(
(
'{"Action":"Action_Idle","quality_status":"","error_type":"",'
'"安全隐患":"","人物位置":"员工在操作台边","总结":"画面中有1人",'
'"时间":"2026-06-16 05:00:03","employees":[],"guests":[],'
'"qsc_events":[{"violation_code":"WGSJ0001",'
'"violation_type":"工作状态未戴口罩","is_violation":true,'
'"working_state":"working","reason":"员工在操作台处理食物时未见口罩",'
'"confidence":0.92,"evidence_frame_count":3,'
'"visible_target":"操作台边员工","evidence_checklist":{},'
'"suggested_action":"warning"}]}'
),
{
"video_id": "video-abc",
"clip_id": "video-abc_c000001",
"clip_start_seconds": 0.0,
"clip_end_seconds": 10.0,
"clip_start_timecode": "00:00:00",
"clip_end_timecode": "00:00:10",
"frame_times": [],
},
{"path": "/videos/a.mp4"},
{
"schema": {"version": "local-batch-v1"},
"runtime": {"timezone": "Asia/Shanghai"},
},
processing={},
)
self.assertEqual(result["status"], "ok")
self.assertEqual(len(result["qsc_events"]), 1)
self.assertEqual(result["qsc_events"][0]["violation_code"], "WGSJ0001")
self.assertEqual(result["qsc_events"][0]["suggested_action"], "warning")
def test_build_clip_result_records_parse_failure_without_crashing(self):
result = build_clip_result(
"not json",
@@ -126,6 +161,7 @@ class ResultParserTests(unittest.TestCase):
self.assertEqual(result["status"], "parse_failed")
self.assertEqual(result["events"], [])
self.assertEqual(result["qsc_events"], [])
self.assertEqual(result["monitoring_timeline"]["screen_time"], "")
self.assertEqual(result["raw_response"], "not json")
self.assertIn("JSON", result["error"])