2025-08-06 14:45:07 +08:00

164 lines
5.6 KiB
Python

#!/usr/bin/env python3
"""
初始化示例数据脚本
"""
import sys
import os
sys.path.append(os.path.dirname(os.path.abspath(__file__)))
from core.database import SessionLocal
from models.algorithm import Algorithm
from models.device import Device
from models.event import Event
from datetime import datetime, timedelta
import json
def init_sample_data():
"""初始化示例数据"""
db = SessionLocal()
try:
# 创建示例设备
devices = [
{
"name": "港口A区主监控",
"device_type": "camera",
"location": "港口A区",
"ip_address": "192.168.1.100",
"port": 554,
"username": "admin",
"password": "admin123",
"rtsp_url": "rtsp://192.168.1.100:554/stream1",
"status": "online",
"resolution": "1920x1080",
"fps": 25,
"algorithm_id": 1,
"is_enabled": True,
"latitude": 22.3193,
"longitude": 114.1694,
"description": "港口A区主要监控摄像头",
"manufacturer": "Hikvision",
"model": "DS-2CD2T47G1-L",
"serial_number": "HK123456789"
},
{
"name": "港口B区监控",
"device_type": "camera",
"location": "港口B区",
"ip_address": "192.168.1.101",
"port": 554,
"username": "admin",
"password": "admin123",
"rtsp_url": "rtsp://192.168.1.101:554/stream1",
"status": "online",
"resolution": "1920x1080",
"fps": 25,
"algorithm_id": 2,
"is_enabled": True,
"latitude": 22.3195,
"longitude": 114.1696,
"description": "港口B区监控摄像头",
"manufacturer": "Dahua",
"model": "IPC-HFW4431R-ZE",
"serial_number": "DH987654321"
},
{
"name": "边检站门禁",
"device_type": "gate",
"location": "边检站入口",
"ip_address": "192.168.1.102",
"port": 80,
"username": "admin",
"password": "admin123",
"status": "online",
"is_enabled": True,
"latitude": 22.3190,
"longitude": 114.1690,
"description": "边检站入口门禁系统",
"manufacturer": "Suprema",
"model": "BioStation 2",
"serial_number": "SP123456789"
}
]
for dev_data in devices:
device = Device(**dev_data)
db.add(device)
db.commit()
print("✅ 设备数据初始化完成")
# 创建示例事件
events = [
{
"event_type": "person_detection",
"device_id": 1,
"algorithm_id": 1,
"severity": "medium",
"status": "pending",
"confidence": 0.95,
"bbox": json.dumps([100, 150, 80, 160]),
"image_path": "/events/images/person_001.jpg",
"description": "检测到人员进入监控区域",
"location": "港口A区",
"detected_objects": json.dumps([{"type": "person", "confidence": 0.95}]),
"processing_time": 15.5,
"is_alert": True,
"alert_sent": True
},
{
"event_type": "vehicle_detection",
"device_id": 2,
"algorithm_id": 2,
"severity": "low",
"status": "resolved",
"confidence": 0.92,
"bbox": json.dumps([200, 100, 120, 80]),
"image_path": "/events/images/vehicle_001.jpg",
"description": "检测到车辆通过",
"location": "港口B区",
"detected_objects": json.dumps([{"type": "car", "confidence": 0.92}]),
"processing_time": 18.2,
"is_alert": False,
"alert_sent": False,
"operator_id": 1,
"resolution_notes": "正常车辆通行",
"resolved_at": datetime.utcnow() - timedelta(hours=2)
},
{
"event_type": "intrusion",
"device_id": 1,
"algorithm_id": 1,
"severity": "high",
"status": "processing",
"confidence": 0.88,
"bbox": json.dumps([300, 200, 60, 120]),
"image_path": "/events/images/intrusion_001.jpg",
"description": "检测到可疑人员入侵",
"location": "港口A区",
"detected_objects": json.dumps([{"type": "person", "confidence": 0.88}]),
"processing_time": 16.0,
"is_alert": True,
"alert_sent": True
}
]
for evt_data in events:
event = Event(**evt_data)
db.add(event)
db.commit()
print("✅ 事件数据初始化完成")
print("\n🎉 所有示例数据初始化完成!")
except Exception as e:
print(f"❌ 初始化数据时出错: {e}")
db.rollback()
finally:
db.close()
if __name__ == "__main__":
init_sample_data()