目录

1. 项目概述
2. 技术栈
3. 目录结构
4. 核心模块
5. API 接口文档
6. 数据库设计
7. 部署指南
8. 开发指南

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.数据库设计

当前实现

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

存储方案

0. 文件存储
1. 图像/视频: 直接存储在文件系统
2. 路径: images/upload/, images/detected/, videos/upload/, videos/detected/
3. 元数据存储
4. 模型信息: models/models_info.json
5. 日志: logs/YYYY.MM.DD.log
6. 优点
7. 简单直接，无需数据库配置
8. 适合小规模应用
9. 易于备份和迁移
10. 缺点
11. 不支持复杂查询
12. 并发性能有限
13. 缺少事务支持

未来扩展（可选）

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

推荐方案: SQLite / PostgreSQL

表结构设计:

-- 图像表CREATE TABLE images (    id INTEGER PRIMARY KEY AUTOINCREMENT,    filename VARCHAR(255) NOT NULL,    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));-- 检测记录表CREATE TABLE 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));-- 检测目标表CREATE TABLE 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));-- 视频表CREATE TABLE videos (    id INTEGER PRIMARY KEY AUTOINCREMENT,    filename VARCHAR(255) NOT NULL,    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');-- 模型表CREATE TABLE models (    id INTEGER PRIMARY KEY AUTOINCREMENT,    filename VARCHAR(255) NOT NULL,    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 backendpip install -r requirements.txt

2. 启动服务

开发模式

python main.py

或使用 uvicorn：

uvicorn main:app --host 0.0.0.0 --port 10077 --reload

生产模式

uvicorn main:app --host 0.0.0.0 --port 10077 --workers 4

参数说明:

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

8.开发指南

1. 添加新的 API 路由

步骤 1: 创建路由文件

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

"""新功能路由"""from fastapi import APIRouter, Requestfrom utils.response import success_response, error_responsefrom utils.logger import loggerrouter = APIRouter()@router.get("/list")async def get_list():    """获取列表"""    try:        data = []  # 获取数据逻辑        return success_response(data, "获取成功")    except Exception as e:        logger.error(f"获取列表失败: {str(e)}")        return error_response(f"获取失败: {str(e)}")@router.post("/create")async def create_item(request: Request):    """创建项目"""    try:        # 创建逻辑        return success_response({"id": 1}, "创建成功")    except Exception as e:        logger.error(f"创建失败: {str(e)}")        return error_response(f"创建失败: {str(e)}")

步骤 2: 注册路由

在 main.py 中导入并注册：

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

2. 使用模型进行检测

@router.post("/detect")async def detect(request: Request):    # 获取模型管理器    model_manager = request.app.state.model_manager    model = model_manager.get_current_model()        if model is None:        return error_response("模型未加载")        # 读取图像    image = cv2.imread("path/to/image.jpg")        # 执行检测    results = model(image)        # 解析结果    detections = []    if results and len(results) > 0:        for box in results[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)        return success_response({"detections": detections})

3. 文件上传处理

from fastapi import File, UploadFile@router.post("/upload")async def upload_file(file: UploadFile = File(...)):    # 读取文件内容    content = await file.read()        # 验证文件大小    if len(content) > MAX_SIZE:        return error_response("文件过大")        # 保存文件    file_path = Path("uploads") / file.filename    with open(file_path, "wb") as f:        f.write(content)        return success_response({"filename": file.filename})

4. 错误处理最佳实践

from fastapi import HTTPException@router.get("/item/{item_id}")async def get_item(item_id: int):    try:        # 业务逻辑        item = find_item(item_id)                if item is None:            # 使用 HTTPException 返回标准 HTTP 错误            raise HTTPException(status_code=404, detail="项目不存在")                return success_response(item)            except HTTPException:        # 重新抛出 HTTP 异常        raise    except Exception as e:        # 记录错误日志        logger.error(f"获取项目失败: {str(e)}", exc_info=True)        # 返回通用错误        return error_response("服务器内部错误", 500)

5. 日志记录

from utils.logger import logger# 不同级别的日志logger.debug("调试信息")logger.info("普通信息")logger.warning("警告信息")logger.error("错误信息")logger.critical("严重错误")# 记录异常堆栈try:    risky_operation()except Exception as e:    logger.error(f"操作失败: {str(e)}", exc_info=True)

6. 异步操作

import asyncio@router.post("/batch-process")async def batch_process(files: List[str]):    """批量处理文件"""        async def process_file(filename: str):        # 异步处理单个文件        await asyncio.sleep(1)  # 模拟耗时操作        return {"filename": filename, "status": "processed"}        # 并发处理多个文件    tasks = [process_file(f) for f in files]    results = await asyncio.gather(*tasks)        return success_response(results)

7. 依赖注入

from fastapi import Dependsdef 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, validatorclass 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. 模型加载失败

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

解决方案:

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

2. 文件上传失败

问题: 上传大文件时失败

解决方案:

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

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

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

解决方案:

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

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

4. CORS 错误

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

解决方案:

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

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

5. 内存占用过高

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

解决方案:

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

6. 端口被占用

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

解决方案:

Windows:

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

Linux/Mac:

lsof -i :10077kill -9 

或修改端口号：

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

10.性能指标

1. API 响应时间

2. 并发能力

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

3. 资源占用

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

4. 优化建议

0. 使用 GPU 加速
1. 启用多进程
2. uvicorn main:app --workers 4
3. 定期清理临时文件
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 magicdef 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 Pathdef 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 Limiterfrom slowapi.util import get_remote_addresslimiter = 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, HTTPAuthorizationCredentialssecurity = 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 jsondef 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 timefrom functools import wrapsdef 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_performanceasync 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 checksdef 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 接口、部署方案和开发指南。

核心特点

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

技术亮点

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

未来规划

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