行人摔倒检测系统 – 后端文档-夜雨聆风

行人摔倒检测系统 – 后端文档

1.项目概述

行人摔倒检测系统后端基于 FastAPI 框架开发，使用 YOLO11 深度学习模型进行目标检测。系统提供 RESTful API 接口，支持图像和视频的上传、检测、下载等功能。

5.API 接口文档

1. 图像管理 API (routers/images.py)

1.1 上传图像

接口: POST /api/images/upload/{filename}

响应:

{
"code": 200,
"message": "上传成功",
"data": {
"filename": "20240129_103000_test.jpg",
"url": "/static/images/upload/20240129_103000_test.jpg"
  }
}

1.2 获取图像列表

接口: GET /api/images/list

1.3 图像检测

接口: POST /api/images/detect/{filename}

响应:

{
"code": 200,
"message": "检测成功",
"data": {
"filename": "20240129_103000_test.jpg",
"detections": [
      {
"class_name": "person",
"confidence": 0.95,
"bbox": [100, 200, 300, 400]
      }
    ]
  }
}

1.4 下载图像

接口: GET /api/images/download/{filename}

响应: 文件流（Blob）

1.5 批量下载图像

接口: POST /api/images/download/batch

响应: ZIP 文件流

1.6 删除图像

接口: DELETE /api/images/delete/{filename}

1.7 批量删除图像

接口: DELETE /api/images/delete/batch

2. 视频管理 API (routers/videos.py)

2.1 上传视频

接口: POST /api/videos/upload/{filename}

2.2 视频检测

接口: POST /api/videos/detect

3. 系统监控 API (routers/systems.py)

3.1 获取系统状态

接口: GET /api/systems/status

响应:

{
"code": 200,
"message": "获取成功",
"data": {
"cpu": {
"percent": 45.2,
"count": 8
    },
"memory": {
"total": 17179869184,
"used": 8589934592,
"available": 8589934592,
"percent": 50.0
    },
"disk": {
"total": 1000000000000,
"used": 500000000000,
"free": 500000000000,
"percent": 50.0
    },
"network": {
"bytes_sent": 1000000,
"bytes_recv": 2000000
    },
"uptime": {
"days": 5,
"hours": 12,
"minutes": 30,
"boot_time": "2024-01-20 10:30:00"
    },
"system": {
"system": "Windows",
"release": "10",
"version": "10.0.19041",
"machine": "AMD64"
    }
  }
}

4. 日志管理 API (routers/logs.py)

4.1 获取日志列表

接口: GET /api/logs/list

响应:

{
"code": 200,
"message": "获取成功",
"data": [
    {
"filename": "2024.01.29.log",
"size": 102400,
"create_time": "2024-01-29 00:00:00",
"modify_time": "2024-01-29 15:30:00"
    }
  ]
}

4.2 预览日志

接口: GET /api/logs/preview/{filename}?lines=200

响应:

4.3 下载日志

接口: GET /api/logs/download/{filename}

响应: 文件流

4.4 删除日志

接口: DELETE /api/logs/{filename}

5. 模型管理 API (routers/models.py)

5.1 获取模型列表

接口: GET /api/models/list

响应:

{
"code": 200,
"message": "获取成功",
"data": [
    {
"filename": "best.pt",
"size": 6291456,
"upload_time": "2024-01-29 10:00:00",
"is_current": true
    }
  ]
}

5.2 获取当前模型

接口: GET /api/models/current

5.3 上传模型

接口: POST /api/models/upload/{filename}

5.4 切换模型

接口: POST /api/models/switch/{filename}

5.5 下载模型

接口: GET /api/models/download/{filename}

5.6 删除模型

接口: DELETE /api/models/delete/{filename}

6. 数据可视化 API (routers/dashboard.py)

6.1 获取统计数据

接口: GET /api/dashboard/stats

6.2 获取数据分布

接口: GET /api/dashboard/data-distribution

6.3 获取检测趋势

接口: GET /api/dashboard/detection-trend

6.数据库设计

当前实现

系统目前采用文件系统存储，不使用传统数据库。

存储方案

文件存储

图像/视频: 直接存储在文件系统
路径: images/upload/, images/detected/, videos/upload/, videos/detected/

元数据存储

模型信息: models/models_info.json
日志: logs/YYYY.MM.DD.log

优点

简单直接，无需数据库配置
适合小规模应用
易于备份和迁移

缺点

不支持复杂查询
并发性能有限
缺少事务支持

未来扩展（可选）

如需支持大规模应用，可考虑引入数据库：

推荐方案: SQLite / PostgreSQL

表结构设计:

-- 图像表
CREATETABLE images (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    filename VARCHAR(255) NOTNULL,
    original_name VARCHAR(255),
    file_path VARCHAR(500),
    file_size INTEGER,
    upload_time DATETIME DEFAULT CURRENT_TIMESTAMP,
    status VARCHAR(50) DEFAULT 'uploaded',
    INDEX idx_filename (filename),
    INDEX idx_upload_time (upload_time)
);

-- 检测记录表
CREATETABLE detections (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    image_id INTEGER,
    video_id INTEGER,
    detect_time DATETIME DEFAULT CURRENT_TIMESTAMP,
    model_name VARCHAR(100),
    result_path VARCHAR(500),
    detection_count INTEGER DEFAULT 0,
    FOREIGN KEY (image_id) REFERENCES images(id),
    FOREIGN KEY (video_id) REFERENCES videos(id)
);

-- 检测目标表
CREATETABLE detection_targets (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    detection_id INTEGER,
    class_name VARCHAR(50),
    confidence FLOAT,
    bbox_x1 FLOAT,
    bbox_y1 FLOAT,
    bbox_x2 FLOAT,
    bbox_y2 FLOAT,
    FOREIGN KEY (detection_id) REFERENCES detections(id)
);

-- 视频表
CREATETABLE videos (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    filename VARCHAR(255) NOTNULL,
    original_name VARCHAR(255),
    file_path VARCHAR(500),
    file_size INTEGER,
    duration FLOAT,
    fps INTEGER,
    width INTEGER,
    height INTEGER,
    upload_time DATETIME DEFAULT CURRENT_TIMESTAMP,
    status VARCHAR(50) DEFAULT 'uploaded'
);

-- 模型表
CREATETABLE models (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    filename VARCHAR(255) NOTNULL,
    file_path VARCHAR(500),
    file_size INTEGER,
    upload_time DATETIME DEFAULT CURRENT_TIMESTAMP,
    is_active BOOLEAN DEFAULT FALSE,
    description TEXT
);

7.部署指南

1. 环境准备

系统要求

Python 3.9+
4GB+ RAM
10GB+ 磁盘空间
CUDA（可选，用于 GPU 加速）

安装 Python 依赖

cd backend
pip install -r requirements.txt

2. 启动服务

开发模式

python main.py

或使用 uvicorn：

uvicorn main:app --host0.0.0.0 --port10077--reload

生产模式

uvicorn main:app --host0.0.0.0 --port10077--workers4

参数说明:

--host: 监听地址（0.0.0.0 表示所有网络接口）
--port: 端口号
--reload: 开发模式自动重载
--workers: 工作进程数（生产环境）

8.开发指南

1. 添加新的 API 路由

步骤 1: 创建路由文件

在 routers/ 目录下创建新文件，例如 new_feature.py:

"""
新功能路由
"""
fromfastapiimportAPIRouter, Request
fromutils.responseimportsuccess_response, error_response
fromutils.loggerimportlogger

router = APIRouter()

@router.get("/list")
asyncdefget_list():
"""获取列表"""
try:
data = []  # 获取数据逻辑
returnsuccess_response(data, "获取成功")
exceptExceptionase:
logger.error(f"获取列表失败: {str(e)}")
returnerror_response(f"获取失败: {str(e)}")

@router.post("/create")
asyncdefcreate_item(request: Request):
"""创建项目"""
try:
# 创建逻辑
returnsuccess_response({"id": 1}, "创建成功")
exceptExceptionase:
logger.error(f"创建失败: {str(e)}")
returnerror_response(f"创建失败: {str(e)}")

步骤 2: 注册路由

在 main.py 中导入并注册：

fromroutersimportnew_feature

app.include_router(
new_feature.router, 
prefix="/api/new-feature", 
tags=["新功能"]
)

2. 使用模型进行检测

@router.post("/detect")
asyncdefdetect(request: Request):
# 获取模型管理器
model_manager = request.app.state.model_manager
model = model_manager.get_current_model()

ifmodelisNone:
returnerror_response("模型未加载")

# 读取图像
image = cv2.imread("path/to/image.jpg")

# 执行检测
results = model(image)

# 解析结果
detections = []
ifresultsandlen(results) >0:
forboxinresults[0].boxes:
detections.append({
"class_name": results[0].names[int(box.cls)],
"confidence": float(box.conf),
"bbox": box.xyxy[0].tolist()
            })

# 保存标注图像
annotated_image = results[0].plot()
cv2.imwrite("output.jpg", annotated_image)

returnsuccess_response({"detections": detections})

3. 文件上传处理

fromfastapiimportFile, UploadFile

@router.post("/upload")
asyncdefupload_file(file: UploadFile = File(...)):
# 读取文件内容
content = awaitfile.read()

# 验证文件大小
iflen(content) >MAX_SIZE:
returnerror_response("文件过大")

# 保存文件
file_path = Path("uploads") /file.filename
withopen(file_path, "wb") asf:
f.write(content)

returnsuccess_response({"filename": file.filename})

4. 错误处理最佳实践

fromfastapiimportHTTPException

@router.get("/item/{item_id}")
asyncdefget_item(item_id: int):
try:
# 业务逻辑
item = find_item(item_id)

ifitemisNone:
# 使用 HTTPException 返回标准 HTTP 错误
raiseHTTPException(status_code=404, detail="项目不存在")

returnsuccess_response(item)

exceptHTTPException:
# 重新抛出 HTTP 异常
raise
exceptExceptionase:
# 记录错误日志
logger.error(f"获取项目失败: {str(e)}", exc_info=True)
# 返回通用错误
returnerror_response("服务器内部错误", 500)

5. 日志记录

fromutils.loggerimportlogger

# 不同级别的日志
logger.debug("调试信息")
logger.info("普通信息")
logger.warning("警告信息")
logger.error("错误信息")
logger.critical("严重错误")

# 记录异常堆栈
try:
risky_operation()
exceptExceptionase:
logger.error(f"操作失败: {str(e)}", exc_info=True)

6. 异步操作

importasyncio

@router.post("/batch-process")
asyncdefbatch_process(files: List[str]):
"""批量处理文件"""

asyncdefprocess_file(filename: str):
# 异步处理单个文件
awaitasyncio.sleep(1)  # 模拟耗时操作
return {"filename": filename, "status": "processed"}

# 并发处理多个文件
tasks = [process_file(f) forfinfiles]
results = awaitasyncio.gather(*tasks)

returnsuccess_response(results)

7. 依赖注入

from fastapi import Depends  def get_current_user(token: str = Header(...)):     """获取当前用户（示例）"""     # 验证 token     user = verify_token(token)     if not user:         raise HTTPException(status_code=401, detail="未授权")     return user  @router.get("/profile") async def get_profile(user = Depends(get_current_user)):     """获取用户资料"""     return success_response(user)

8. 请求验证

from pydantic import BaseModel, Field, validator  class CreateItemRequest(BaseModel):     name: str = Field(..., min_length=1, max_length=100)     price: float = Field(..., gt=0)     quantity: int = Field(default=1, ge=1)          @validator('name')     def name_must_not_be_empty(cls, v):         if not v.strip():             raise ValueError('名称不能为空')         return v  @router.post("/items") async def create_item(item: CreateItemRequest):     """创建项目"""     return success_response({         "name": item.name,         "price": item.price,         "quantity": item.quantity     })

9.常见问题

1. 模型加载失败

问题: 启动时提示”模型加载失败”

解决方案:

检查模型文件是否存在: trained/weights/best.pt
确认模型文件格式正确（.pt 格式）
检查文件权限
查看日志文件获取详细错误信息

2. 文件上传失败

问题: 上传大文件时失败

解决方案:

检查文件大小限制配置
检查磁盘空间是否充足
确认文件格式是否支持

3. 视频检测结果无法播放

问题: 检测后的视频在浏览器中无法播放

解决方案:

确认使用 H.264 编码器（avc1）
检查 OpenCV 是否正确安装
尝试使用不同的编码器
查看后端日志确认编码器选择

# 检查可用的编码器 fourcc_options = [     cv2.VideoWriter_fourcc(*'avc1'),  # H.264     cv2.VideoWriter_fourcc(*'H264'),     cv2.VideoWriter_fourcc(*'X264'),     cv2.VideoWriter_fourcc(*'mp4v'), ]

4. CORS 错误

问题: 前端请求被 CORS 策略阻止

解决方案:

确认 CORS 中间件已正确配置
检查 allow_origins 设置
开发环境可以使用 ["*"]，生产环境应指定具体域名

app.add_middleware(     CORSMiddleware,     allow_origins=["http://localhost:3000"],  # 生产环境     allow_credentials=True,     allow_methods=["*"],     allow_headers=["*"], )

5. 内存占用过高

问题: 长时间运行后内存占用过高

解决方案:

及时释放大对象（图像、视频）
使用生成器处理大文件
限制并发请求数
定期重启服务

6. 端口被占用

问题: 启动时提示端口 10077 已被占用

解决方案:

Windows:

netstat -ano | findstr :10077 taskkill /PID <PID> /F

Linux/Mac:

lsof -i :10077 kill -9 <PID>

或修改端口号：

uvicorn.run("main:app", host="0.0.0.0", port=10078)

10.性能指标

1. API 响应时间

接口	平均响应时间	说明
图像上传	100-500ms	取决于文件大小
图像检测	200-800ms	取决于图像分辨率
视频检测	5-60s	取决于视频长度和分辨率
获取列表	10-50ms	取决于文件数量
系统监控	50-100ms	实时数据采集

2. 并发能力

单进程: 支持 50-100 并发请求
多进程: 支持 200-500 并发请求（4 workers）
建议: 生产环境使用 4-8 个 worker 进程

3. 资源占用

内存: 基础 500MB，每个检测任务额外 200-500MB
CPU: 检测时 CPU 使用率 60-90%
磁盘: 日志约 10MB/天，检测结果根据使用量

4. 优化建议

使用 GPU 加速
启用多进程
```
uvicorn main:app --workers 4
```

定期清理临时文件

# 定时任务清理 7 天前的文件 import schedule  def cleanup_old_files():     cutoff = datetime.now() - timedelta(days=7)     for file in Path("images/upload").glob("*"):         if datetime.fromtimestamp(file.stat().st_mtime) < cutoff:             file.unlink()  schedule.every().day.at("02:00").do(cleanup_old_files)

11.安全建议

1. 文件上传安全

import magic  def validate_file(file_content: bytes, allowed_types: set):     """验证文件类型"""     mime = magic.from_buffer(file_content, mime=True)     if mime not in allowed_types:         raise ValueError(f"不支持的文件类型: {mime}")

2. 路径遍历防护

from pathlib import Path  def safe_join(base_dir: Path, filename: str) -> Path:     """安全的路径拼接"""     full_path = (base_dir / filename).resolve()     if not str(full_path).startswith(str(base_dir.resolve())):         raise ValueError("非法的文件路径")     return full_path

3. 限流

from slowapi import Limiter from slowapi.util import get_remote_address  limiter = Limiter(key_func=get_remote_address)  @app.get("/api/images/list") @limiter.limit("10/minute") async def list_images(request: Request):     pass

4. 认证和授权（可选）

from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials  security = HTTPBearer()  @app.get("/api/protected") async def protected_route(credentials: HTTPAuthorizationCredentials = Depends(security)):     token = credentials.credentials     # 验证 token     if not verify_token(token):         raise HTTPException(status_code=401, detail="无效的令牌")     return {"message": "访问成功"}

11.监控和日志

1. 日志级别

import logging  # 设置不同环境的日志级别 if ENV == "development":     logger.setLevel(logging.DEBUG) elif ENV == "production":     logger.setLevel(logging.INFO)

2. 结构化日志

import json  def log_request(request: Request, response_time: float):     """记录请求日志"""     log_data = {         "timestamp": datetime.now().isoformat(),         "method": request.method,         "path": request.url.path,         "client_ip": request.client.host,         "response_time": response_time,         "user_agent": request.headers.get("user-agent")     }     logger.info(json.dumps(log_data))

3. 性能监控

import time from functools import wraps  def monitor_performance(func):     """性能监控装饰器"""     @wraps(func)     async def wrapper(*args, **kwargs):         start_time = time.time()         result = await func(*args, **kwargs)         elapsed_time = time.time() - start_time                  logger.info(f"{func.__name__} 执行时间: {elapsed_time:.2f}s")         return result     return wrapper  @router.post("/detect") @monitor_performance async def detect_image(filename: str):     # 检测逻辑     pass

4. 健康检查

@app.get("/health") async def health_check():     """健康检查端点"""     checks = {         "status": "healthy",         "timestamp": datetime.now().isoformat(),         "checks": {             "database": check_database(),             "model": check_model(),             "disk_space": check_disk_space()         }     }     return checks  def check_model():     """检查模型状态"""     try:         model = app.state.model_manager.get_current_model()         return "ok" if model else "error"     except:         return "error"  def check_disk_space():     """检查磁盘空间"""     disk = psutil.disk_usage('/')     if disk.percent > 90:         return "warning"     return "ok"

12.API 文档

自动生成的文档

FastAPI 自动生成交互式 API 文档：

Swagger UI: http://localhost:10077/docs
ReDoc: http://localhost:10077/redoc
OpenAPI JSON: http://localhost:10077/openapi.json

13.总结

本文档详细介绍了行人摔倒检测系统后端的架构、API 接口、部署方案和开发指南。

核心特点

FastAPI 框架: 现代化、高性能的 Python Web 框架
YOLO11 模型: 先进的目标检测算法
RESTful API: 标准化的接口设计
模块化架构: 清晰的代码组织
完善的日志: 便于调试和监控

技术亮点

异步处理提高并发性能
统一的响应格式
完善的错误处理
自动生成 API 文档
灵活的模型管理
实时系统监控

未来规划

添加数据库支持
实现用户认证系统
支持分布式部署
添加消息队列
实现实时推送
性能优化和缓存