131 lines
3.9 KiB
Python
131 lines
3.9 KiB
Python
"""
|
||
边缘端运行测试脚本
|
||
添加测试摄像头和ROI配置，验证系统正常运行
|
||
"""
|
||
|
||
import sys
|
||
import os
|
||
|
||
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||
|
||
from config.database import get_sqlite_manager
|
||
from datetime import datetime
|
||
import random
|
||
|
||
|
||
def setup_test_data():
|
||
"""设置测试数据"""
|
||
db = get_sqlite_manager()
|
||
|
||
print("=" * 60)
|
||
print("边缘端运行测试 - 数据准备")
|
||
print("=" * 60)
|
||
|
||
camera_id = "test_camera_01"
|
||
rtsp_url = "rtsp://admin:admin@172.16.8.35/cam/realmonitor?channel=6&subtype=1"
|
||
|
||
print(f"\n1. 添加摄像头配置")
|
||
print(f" camera_id: {camera_id}")
|
||
print(f" rtsp_url: {rtsp_url}")
|
||
|
||
result = db.save_camera_config(
|
||
camera_id=camera_id,
|
||
rtsp_url=rtsp_url,
|
||
camera_name="测试摄像头-车间入口",
|
||
location="车间入口通道",
|
||
enabled=True,
|
||
status=True
|
||
)
|
||
print(f" 结果: {'成功' if result else '失败'}")
|
||
|
||
print(f"\n2. 添加ROI配置（随机划分区域）")
|
||
|
||
roi_configs = [
|
||
{
|
||
"roi_id": f"{camera_id}_roi_01",
|
||
"name": "离岗检测区域",
|
||
"roi_type": "polygon",
|
||
"coordinates": [[100, 50], [300, 50], [300, 200], [100, 200]],
|
||
"algorithm_type": "leave_post",
|
||
"target_class": "person",
|
||
"confirm_on_duty_sec": 10,
|
||
"confirm_leave_sec": 30,
|
||
"cooldown_sec": 60,
|
||
"working_hours": [{"start": "08:00", "end": "18:00"}],
|
||
},
|
||
{
|
||
"roi_id": f"{camera_id}_roi_02",
|
||
"name": "入侵检测区域",
|
||
"roi_type": "polygon",
|
||
"coordinates": [[350, 50], [550, 50], [550, 200], [350, 200]],
|
||
"algorithm_type": "intrusion",
|
||
"target_class": "person",
|
||
"alert_threshold": 3,
|
||
"alert_cooldown": 60,
|
||
"confirm_on_duty_sec": 10,
|
||
"confirm_leave_sec": 10,
|
||
"cooldown_sec": 60,
|
||
"working_hours": None,
|
||
},
|
||
]
|
||
|
||
for roi in roi_configs:
|
||
print(f"\n ROI: {roi['name']}")
|
||
print(f" - roi_id: {roi['roi_id']}")
|
||
print(f" - algorithm_type: {roi['algorithm_type']}")
|
||
print(f" - coordinates: {roi['coordinates']}")
|
||
|
||
result = db.save_roi_config(
|
||
roi_id=roi["roi_id"],
|
||
camera_id=camera_id,
|
||
roi_type=roi["roi_type"],
|
||
coordinates=roi["coordinates"],
|
||
algorithm_type=roi["algorithm_type"],
|
||
target_class=roi["target_class"],
|
||
confirm_on_duty_sec=roi["confirm_on_duty_sec"],
|
||
confirm_leave_sec=roi["confirm_leave_sec"],
|
||
cooldown_sec=roi["cooldown_sec"],
|
||
working_hours=str(roi["working_hours"]),
|
||
)
|
||
print(f" 结果: {'成功' if result else '失败'}")
|
||
|
||
print("\n" + "=" * 60)
|
||
print("测试数据准备完成")
|
||
print("=" * 60)
|
||
|
||
return camera_id, roi_configs
|
||
|
||
|
||
def verify_data():
|
||
"""验证数据"""
|
||
db = get_sqlite_manager()
|
||
|
||
print("\n" + "=" * 60)
|
||
print("验证数据库中的配置")
|
||
print("=" * 60)
|
||
|
||
cameras = db.get_all_camera_configs()
|
||
print(f"\n摄像头数量: {len(cameras)}")
|
||
for cam in cameras:
|
||
print(f" - {cam['camera_id']}: {cam['camera_name']} ({cam['rtsp_url'][:50]}...)")
|
||
|
||
rois = db.get_all_roi_configs()
|
||
print(f"\nROI数量: {len(rois)}")
|
||
for roi in rois:
|
||
print(f" - {roi['roi_id']}: {roi['name']} ({roi['algorithm_type']})")
|
||
|
||
return len(cameras) > 0 and len(rois) > 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
print("\n" + "#" * 60)
|
||
print("# 边缘端运行测试 - 数据准备")
|
||
print("#" * 60)
|
||
|
||
setup_test_data()
|
||
verify_data()
|
||
|
||
print("\n" + "#" * 60)
|
||
print("# 测试数据准备完成，请运行 main.py 进行推理测试")
|
||
print("#" * 60)
|