model_train_dm/docs/边缘智能终端模型同步功能设计.md

边缘智能终端模型同步功能设计

一、概述

1.1 系统架构

┌──────────────────────────────────────────────────────────────────┐
│                        云端服务器 (Cloud)                         │
│  ┌──────────┐  ┌────────────┐  ┌───────────┐  ┌──────────────┐  │
│  │ 认证服务  │  │ 心跳服务    │  │ 版本管理   │  │ 模型下载服务  │  │
│  │ (Token)  │  │ (Heartbeat)│  │ (Version) │  │ (Download)   │  │
│  └────┬─────┘  └─────┬──────┘  └─────┬─────┘  └──────┬───────┘  │
│       │              │               │               │           │
│  ┌────┴──────────────┴───────────────┴───────────────┴──────┐   │
│  │                    REST API + WebSocket                   │   │
│  └──────────────────────────┬───────────────────────────────┘   │
│                             │                                    │
│  ┌──────────────────────────┴───────────────────────────────┐   │
│  │  Django REST Framework + Channels (ASGI)                  │   │
│  │  MinIO 对象存储 (archived-model / trained-model)           │   │
│  └──────────────────────────────────────────────────────────┘   │
└──────────────────────────────────────────────────────────────────┘
                              │
                    HTTPS / WSS（公网/内网）
                              │
┌─────────────────────────────┴────────────────────────────────────┐
│                    边缘智能终端 (Edge Device)                       │
│  ┌──────────┐  ┌────────────┐  ┌───────────┐  ┌──────────────┐  │
│  │ Token管理 │  │ 心跳上报    │  │ 版本检查   │  │ 模型下载更新  │  │
│  │ (每日刷新)│  │ (30s/次)   │  │ (定时轮询) │  │ (差量更新)   │  │
│  └──────────┘  └────────────┘  └───────────┘  └──────────────┘  │
│                                                                   │
│  ┌──────────────────────────────────────────────────────────┐   │
│  │  Edge Runtime (Python 模拟程序 / 边缘盒子固件)             │   │
│  │  本地模型目录: /opt/edge/models/                          │   │
│  │  本地配置目录: /opt/edge/config/                          │   │
│  └──────────────────────────────────────────────────────────┘   │
└──────────────────────────────────────────────────────────────────┘

1.2 核心数据模型

基于已有数据表 ai_smart_terminal，核心字段如下：

字段	类型	说明
stcd	CharField(40)	终端唯一编码，边缘设备与云端一一对应
stnm	CharField(255)	终端名称
model_id	CharField(40)	已部署模型ID（关联 AiAlgorithmModels）
model_name	CharField(200)	已部署模型文件名
model_version	CharField(64)	云端当前发布的模型版本
model_path	CharField(600)	模型文件 MinIO 存储路径
model_publish_time	DateTimeField	模型发布时间
edge_model_version	CharField(64)	边缘端当前运行的模型版本
edge_update_time	DateTimeField	边缘端最后一次模型更新时间
last_heartbeat_time	DateTimeField	边缘端最后心跳时间
online_status	SmallIntegerField	在线状态：0=离线 1=在线
sync_check_interval	IntegerField	同步检查间隔（单位：分钟）

---

二、云端服务功能设计

2.1 设备接入认证（Token 生成）

接口: POST /server/terminal/auth/

功能说明:

• 边缘设备使用设备编码（stcd）向云端申请认证 Token
• Token 有效期 1 天（24小时），基于 JWT 机制
• Token 中携带 stcd ，区别于用户 Token

请求参数:

{
  "stcd": "DEVICE001" 
}

响应示例:

{
  "code": 200,
  "data": {
    "access_token": "eyJhbGciOiJIUzI1NiIs...",
    "token_type": "bearer",
    "expires_in": 86400,
    "stcd": "DEVICE001"   
  }
}

认证中间件:

• 所有边缘设备 API 使用 EdgeDeviceAuthentication 认证类
• 验证 Token 中的 stcd 与请求参数/URL 中的 stcd 一致
• Token 过期返回 401，设备端触发重新认证流程

JWT Token 生成逻辑:

from rest_framework_simplejwt.tokens import AccessToken
from datetime import timedelta

def generate_device_token(stcd: str) -> str:
    token = AccessToken()
    token['stcd'] = stcd   
    token['type'] = 'device'
    token.set_exp(lifetime=timedelta(days=1))
    return str(token)

2.2 心跳时间更新

接口: PUT /server/terminal/heartbeat/（已有，需增强认证）

功能说明:

• 边缘设备定时调用（建议 30 秒/次）
• 更新 last_heartbeat_time 为当前时间
• 更新 online_status 为 1（在线）
• 若设备首次接入，同时设置 dtin=1 和 dtin_tm

请求参数:

{
  "stcd": "DEVICE001"
}

请求头:

Authorization: Bearer <device_token>

响应示例:

{
  "code": 200,
  "data": {
    "stcd": "DEVICE001",
    "last_heartbeat_time": "2026-07-25T15:30:00Z",
    "online_status": 1
  }
}

2.3 离线状态检测（定时任务）

实现方式: Django Management Command + Celery Beat / APScheduler / 系统 Cron

执行频率: 每 30 秒执行一次

检测逻辑:

1. 查询所有 online_status=1 且 last_heartbeat_time 不为空的终端
2. 计算 now() - last_heartbeat_time
3. 若差值 > 5 分钟（300秒），则将 online_status 更新为 0
4. 通过 WebSocket 向前端管理页面推送离线通知

核心代码逻辑:

# management/commands/check_terminal_offline.py
from django.core.management.base import BaseCommand
from django.utils.timezone import now
from datetime import timedelta
from apps.core.models import AiSmartTerminal
from asgiref.sync import async_to_sync
from channels.layers import get_channel_layer


class Command(BaseCommand):
    help = '检测超时心跳的终端并标记为离线'

    def handle(self, *args, **options):
        threshold = now() - timedelta(minutes=5)
        offline_terminals = AiSmartTerminal.objects.filter(
            online_status=1,
            last_heartbeat_time__lt=threshold,
        )
        count = offline_terminals.update(online_status=0)

        if count > 0:
            # WebSocket 推送离线通知到前端
            channel_layer = get_channel_layer()
            offline_list = list(
                offline_terminals.values('stcd', 'stnm')
            )
            async_to_sync(channel_layer.group_send)(
                "terminal_status",
                {
                    "type": "terminal.status",
                    "data": {
                        "event": "offline",
                        "count": count,
                        "terminals": offline_list,
                    }
                }
            )
        self.stdout.write(f"标记 {count} 个终端为离线")

WebSocket 前端订阅:

• 前端管理页面连接 WebSocket: ws://host/ws/terminal/status/
• 收到 online_status=0 消息时，实时更新终端列表中的在线状态标识
• 可触达前端消息提示（如 ElNotification）

2.4 模型版本检查

接口: GET /server/terminal/checkVersion/

功能说明:

• 边缘设备定期调用，检查云端是否有新版本模型
• 对比 edge_model_version（边缘当前版本）与 model_version（云端发布版本）
• 返回是否需要更新

请求参数:

GET /server/terminal/checkVersion/?stcd=DEVICE001&edge_version=v1.0

响应示例（有新版本）:

{
  "code": 200,
  "data": {
    "having_newmodel": 1,
    "model_name": "model-001.onnx",
    "model_version": "v2.0",  
    "model_publish_time": "2026-07-25T12:00:00Z",
    "sync_check_interval": 30
  }
}

响应示例（已是最新）:

{
  "code": 200,
  "data": {
    "having_newmodel": 0,
    "model_version": "v1.0",
    "sync_check_interval": 30
  }
}

版本对比逻辑:

• 若 edge_model_version 为空（首次同步），having_newmodel=1
• 若 edge_model_version != model_version，having_newmodel=1
• 若 edge_model_version == model_version，having_newmodel=0
• 返回 sync_check_interval 供边缘端动态调整检查频率

2.5 模型下载

接口: GET /server/terminal/downloadModel/（已有，需增强）

功能说明:

• 边缘设备下载最新 ONNX 模型文件
• 支持断点续传（通过 Range 请求头）
• 返回文件流，Content-Disposition 包含文件名
• 下载完成后，边缘端调用回调接口更新 edge_model_version 和 edge_update_time

请求参数:

GET /server/terminal/downloadModel/?stcd=DEVICE001&format=onnx

请求头:

Authorization: Bearer <device_token>
Range: bytes=0-1048576   （可选，断点续传）

响应:

HTTP/1.1 200 OK
Content-Type: application/octet-stream
Content-Disposition: attachment; filename="model-001.onnx"
Content-Length: 10485760

下载完成回调接口: PUT /server/terminal/updateModelVersion/

{
  "stcd": "DEVICE001",
  "edge_model_version": "v2.0"
}

---

三、边缘设备端设计（模拟程序）

3.1 目录结构

edge_simulator/
├── config/
│   ├── config.json          # 配置文件（设备编码、云端地址等）
│   └── token.json           # 缓存的认证 Token
├── models/                   # 模型存储目录
│   ├── current/              # 当前运行模型
│   └── backup/               # 历史版本备份
├── logs/
│   └── edge.log
├── main.py                   # 主入口，调度各模块
├── auth.py                   # Token 管理与刷新
├── heartbeat.py              # 心跳上报模块
├── version_check.py          # 版本检查与下载
└── requirements.txt

3.2 配置文件 (config/config.json)

{
  "stcd": "DEVICE001", 
  "cloud_base_url": "http://192.168.1.60:8000",
  "heartbeat_interval_sec": 30,
  "version_check_interval_min": null,
  "model_dir": "./models/current",
  "backup_dir": "./models/backup",
  "token_file": "./config/token.json",
  "log_level": "INFO",
  "retry_count": 3,
  "retry_delay_sec": 5
}

3.3 模块详细设计

3.3.1 Token 管理模块 (auth.py)

功能:

• 每日定时获取新 Token（每天 00:00:00 执行）
• 首次启动时立即获取 Token
• 缓存 Token 到本地文件，记录过期时间
• 每次 API 调用前检查 Token 是否有效

流程:

开始
  │
  ├── 检查本地 token.json 是否存在
  │     ├── 存在 → 检查是否过期
  │     │     ├── 未过期 → 直接使用
  │     │     └── 已过期 → 调用认证接口获取新 Token
  │     └── 不存在 → 调用认证接口获取新 Token
  │
  └── 保存 Token 到 token.json
       └── 记录 expires_at 时间戳

token.json 格式:

{
  "access_token": "eyJhbGciOiJIUzI1NiIs...",
  "token_type": "bearer",
  "expires_at": "2026-07-26T00:00:00Z",
  "stcd": "DEVICE001"
}

3.3.2 心跳上报模块 (heartbeat.py)

功能:

• 定时 30 秒调用云端心跳接口
• 携带设备 Token 进行认证
• 失败重试 3 次（间隔 2 秒）
• 记录心跳日志

流程:

定时触发 (每30秒)
  │
  ├── 获取本地 Token
  ├── PUT /server/terminal/heartbeat/
  │      Headers: Authorization: Bearer <token>
  │      Body: {"stcd": "DEVICE001"}
  │
  ├── 响应码 = 200 → 成功，记录日志
  ├── 响应码 = 401 → Token 过期，触发 Token 刷新 → 重试
  ├── 响应码 = 其他 / 超时 → 重试（最多3次）
  │
  └── 3次失败 → 记录错误日志，等待下次触发

3.3.3 版本检查与模型下载模块 (version_check.py)

功能:

• 根据 sync_check_interval 定时检查云端模型版本
• 若有新版本，下载模型文件到本地（支持断点续传）
• 验证下载文件的 MD5/SHA256 完整性
• 备份旧版本模型
• 更新 edge_model_version 到云端

流程:

定时触发 (根据 sync_check_interval，默认从云端获取)
  │
  ├── 读取本地当前模型版本 (从本地 version.txt)
  ├── GET /server/terminal/checkVersion/
  │      ?stcd=DEVICE001
  │      &edge_version=v1.0
  │
  ├── 响应 having_newmodel = 0 → 无更新，等待下次检查
  │
  ├── 响应 having_newmodel = 1
  │     │
  │     ├── 更新本地 sync_check_interval
  │     │
  │     ├── 备份旧模型: models/current/ → models/backup/model-{version}/
  │     │
  │     ├── GET /server/terminal/downloadModel/
  │     │      ?stcd=DEVICE001
  │     │      &format=onnx
  │     │      支持断点续传 (Range 请求头)
  │     │
  │     ├── 下载完成 → 验证文件完整性
  │     │     验证方式: 响应头 X-File-SHA256 与实际文件对比
  │     │
  │     ├── 保存到 models/current/{model_filename}
  │     ├── 更新本地 version.txt
  │     │
  │     └── PUT /server/terminal/updateModelVersion/
  │            {"stcd": "DEVICE001", "edge_model_version": "v2.0"}
  │            → 更新 edge_model_version 和 edge_update_time
  │
  └── 记录更新日志

3.3.4 主调度程序 (main.py)

"""
边缘智能终端模拟程序 - 主入口
"""
import json
import logging
import os
import signal
import sys
import threading
import time
from datetime import datetime

import requests


class EdgeDeviceSimulator:
    """边缘智能终端模拟器"""

    def __init__(self, config_path: str = "./config/config.json"):
        self.config = self._load_config(config_path)
        self.stcd = self.config["stcd"]
        self.secret = self.config["secret"]
        self.base_url = self.config["cloud_base_url"]
        self.running = True

        self._setup_logging()
        self._setup_signal_handlers()

        self.token = None
        self.token_refreshed_at = None

        self.heartbeat_interval = self.config.get("heartbeat_interval_sec", 30)
        self.version_check_interval = None  # 从云端获取
        self.last_version_check = None

    # ── 配置与日志 ──────────────────────────────

    def _load_config(self, path: str) -> dict:
        with open(path, "r", encoding="utf-8") as f:
            return json.load(f)

    def _setup_logging(self):
        logging.basicConfig(
            level=getattr(logging, self.config.get("log_level", "INFO")),
            format="%(asctime)s [%(levelname)s] %(message)s",
            handlers=[
                logging.FileHandler("./logs/edge.log", encoding="utf-8"),
                logging.StreamHandler(sys.stdout),
            ],
        )
        self.logger = logging.getLogger("EdgeDevice")

    def _setup_signal_handlers(self):
        signal.signal(signal.SIGINT, self._graceful_shutdown)
        signal.signal(signal.SIGTERM, self._graceful_shutdown)

    def _graceful_shutdown(self, signum, frame):
        self.logger.info("收到退出信号，正在安全关闭...")
        self.running = False

    # ── Token 管理 ───────────────────────────────

    def _load_token_from_file(self) -> dict | None:
        token_file = self.config["token_file"]
        if not os.path.exists(token_file):
            return None
        with open(token_file, "r", encoding="utf-8") as f:
            data = json.load(f)
        expires_at = datetime.fromisoformat(data["expires_at"])
        if datetime.now() >= expires_at:
            self.logger.info("本地 Token 已过期")
            return None
        return data

    def _save_token_to_file(self, token_data: dict):
        with open(self.config["token_file"], "w", encoding="utf-8") as f:
            json.dump(token_data, f, indent=2)

    def ensure_token(self) -> str:
        """确保 Token 有效，过期则重新获取"""
        token_data = self._load_token_from_file()
        if token_data:
            return token_data["access_token"]

        self.logger.info("获取新的设备 Token...")
        resp = requests.post(
            f"{self.base_url}/server/terminal/auth/",
            json={"stcd": self.stcd, "secret": self.secret},
            timeout=10,
        )
        resp.raise_for_status()
        result = resp.json()
        token_data = result["data"]
        self._save_token_to_file(token_data)
        self.logger.info(f"Token 获取成功, 有效期至 {token_data['expires_at']}")
        return token_data["access_token"]

    # ── 心跳上报 ──────────────────────────────────

    def _send_heartbeat(self):
        """发送一次心跳"""
        try:
            token = self.ensure_token()
            resp = requests.put(
                f"{self.base_url}/server/terminal/heartbeat/",
                json={"stcd": self.stcd},
                headers={"Authorization": f"Bearer {token}"},
                timeout=5,
            )
            if resp.status_code == 200:
                self.logger.debug("心跳上报成功")
                return True
            elif resp.status_code == 401:
                self.logger.warning("Token 认证失败，清除本地 Token 缓存")
                os.remove(self.config["token_file"])
            else:
                self.logger.warning(f"心跳上报失败: {resp.status_code}")
        except requests.RequestException as e:
            self.logger.error(f"心跳上报异常: {e}")
        return False

    def _heartbeat_loop(self):
        """心跳上报定时循环"""
        self.logger.info(f"心跳上报模块启动, 间隔 {self.heartbeat_interval} 秒")
        while self.running:
            self._send_heartbeat()
            time.sleep(self.heartbeat_interval)

    # ── 版本检查与模型更新 ─────────────────────────

    def _check_version(self) -> dict | None:
        """检查云端模型版本"""
        local_version = self._get_local_version()
        try:
            token = self.ensure_token()
            resp = requests.get(
                f"{self.base_url}/server/terminal/checkVersion/",
                params={"stcd": self.stcd, "edge_version": local_version or ""},
                headers={"Authorization": f"Bearer {token}"},
                timeout=10,
            )
            resp.raise_for_status()
            return resp.json()["data"]
        except requests.RequestException as e:
            self.logger.error(f"版本检查失败: {e}")
            return None

    def _get_local_version(self) -> str | None:
        """获取本地当前运行的模型版本"""
        version_file = os.path.join(self.config["model_dir"], "version.txt")
        if os.path.exists(version_file):
            with open(version_file, "r") as f:
                return f.read().strip()
        return None

    def _backup_current_model(self):
        """备份当前模型"""
        model_dir = self.config["model_dir"]
        backup_dir = self.config["backup_dir"]
        local_version = self._get_local_version()
        if not os.path.exists(model_dir) or not local_version:
            return
        version_backup = os.path.join(backup_dir, f"model-{local_version}")
        os.makedirs(version_backup, exist_ok=True)
        for f in os.listdir(model_dir):
            src = os.path.join(model_dir, f)
            dst = os.path.join(version_backup, f)
            if os.path.isfile(src):
                import shutil
                shutil.copy2(src, dst)
        self.logger.info(f"旧模型已备份到 {version_backup}")

    def _download_model(self, model_name: str) -> str | None:
        """下载模型文件，支持断点续传"""
        model_dir = self.config["model_dir"]
        os.makedirs(model_dir, exist_ok=True)
        save_path = os.path.join(model_dir, model_name)

        try:
            token = self.ensure_token()
            url = f"{self.base_url}/server/terminal/downloadModel/"
            params = {"stcd": self.stcd, "format": "onnx"}
            headers = {"Authorization": f"Bearer {token}"}

            self.logger.info(f"开始下载模型: {model_name}")
            resp = requests.get(
                url, params=params, headers=headers,
                stream=True, timeout=300,
            )
            resp.raise_for_status()

            total_size = int(resp.headers.get("Content-Length", 0))
            downloaded = 0
            with open(save_path, "wb") as f:
                for chunk in resp.iter_content(chunk_size=8192):
                    f.write(chunk)
                    downloaded += len(chunk)
                    if total_size:
                        progress = downloaded / total_size * 100
                        if downloaded % (1024 * 1024 * 10) < 8192:  # 每10MB打印
                            self.logger.info(
                                f"下载进度: {progress:.1f}% "
                                f"({downloaded}/{total_size})"
                            )

            self.logger.info(f"模型下载完成, 保存至 {save_path}")
            return save_path
        except Exception as e:
            self.logger.error(f"模型下载失败: {e}")
            return None

    def _update_edge_version(self, new_version: str):
        """通知云端已更新到新版本"""
        try:
            token = self.ensure_token()
            resp = requests.put(
                f"{self.base_url}/server/terminal/updateModelVersion/",
                json={"stcd": self.stcd, "edge_model_version": new_version},
                headers={"Authorization": f"Bearer {token}"},
                timeout=10,
            )
            if resp.status_code == 200:
                self.logger.info(f"边缘模型版本已更新: {new_version}")
                return True
        except Exception as e:
            self.logger.error(f"云端版本更新通知失败: {e}")
        return False

    def _version_check_loop(self):
        """版本检查与更新定时循环"""
        self.logger.info("版本检查模块启动")
        while self.running:
            result = self._check_version()
            if result and result.get("having_newmodel") == 1:
                model_name = result["model_name"]
                new_version = result["model_version"]
                self.logger.info(f"发现新版本: {new_version}, 模型: {model_name}")

                # 备份旧模型
                self._backup_current_model()

                # 下载新模型
                saved = self._download_model(model_name)
                if saved:
                    # 更新本地版本记录
                    version_file = os.path.join(
                        self.config["model_dir"], "version.txt"
                    )
                    with open(version_file, "w") as f:
                        f.write(new_version)

                    # 通知云端
                    self._update_edge_version(new_version)

                # 更新检查间隔
                if result.get("sync_check_interval"):
                    self.version_check_interval = result["sync_check_interval"]

            # 等待下次检查
            interval = self.version_check_interval or 60  # 默认60分钟
            self.logger.debug(f"下次版本检查: {interval} 分钟后")
            time.sleep(interval * 60)

    # ── 主入口 ────────────────────────────────────

    def run(self):
        """启动边缘模拟器"""
        self.logger.info(f"边缘智能终端启动, stcd={self.stcd}")
        self.logger.info(f"云端地址: {self.base_url}")

        # 启动心跳线程
        heartbeat_thread = threading.Thread(
            target=self._heartbeat_loop, daemon=True, name="Heartbeat"
        )
        heartbeat_thread.start()

        # 启动版本检查线程
        version_thread = threading.Thread(
            target=self._version_check_loop, daemon=True, name="VersionCheck"
        )
        version_thread.start()

        # 主线程等待
        try:
            while self.running:
                time.sleep(1)
        except KeyboardInterrupt:
            pass

        self.logger.info("边缘智能终端已安全退出")


if __name__ == "__main__":
    simulator = EdgeDeviceSimulator()
    simulator.run()

---

四、新增/修改 API 接口汇总

接口路径	方法	认证	说明	状态
/server/terminal/auth/	POST	无（设备密钥认证）	设备认证，获取 JWT Token	新增
/server/terminal/heartbeat/	PUT	设备 Token	更新心跳时间（已有，需增强认证）	修改
/server/terminal/checkVersion/	GET	设备 Token	检查云端模型版本	新增
/server/terminal/downloadModel/	GET	无历史遗留	下载模型文件（已有，需增强）	修改
/server/terminal/updateModelVersion/	PUT	设备 Token	更新边缘端模型版本	新增
ws://host/ws/terminal/status/	WebSocket	用户 Token	推送终端在线状态变更	新增

---

五、安全性设计

5.1 认证体系

┌─────────────────────────────────────┐
│         双层 Token 体系              │
├─────────────────────────────────────┤
│ 用户 Token (user)                    │
│ ├── 有效期: 60分钟                   │
│ ├── 用途: 前端管理页面操作            │
│ └── sub: user                       │
│                                     │
│ 设备 Token (device)                  │
│ ├── 有效期: 24小时                   │
│ ├── 用途: 边缘设备 API 调用           │
│ ├── sub: edge_device                │
│ └── stcd: 设备编码                   │
└─────────────────────────────────────┘

5.2 安全措施

1. 设备预置密钥: 首次激活时下发唯一密钥，用于初始 Token 申请
2. Token 绑定设备: Token 中携带 stcd，服务端验证请求中的 stcd 与 Token 一致
3. HTTPS 传输: 生产环境强制 HTTPS，防止中间人攻击
4. 请求频率限制: 心跳接口限制 10次/分钟/设备，防止恶意刷接口
5. 模型下载签名: 可扩展为预签名 URL，限制下载有效期

---

六、部署与运维

6.1 云端定时任务配置

方式一: Django Management Command + 系统 Cron

# 每30秒执行一次离线检测
* * * * * /path/to/python manage.py check_terminal_offline
* * * * * sleep 30 && /path/to/python manage.py check_terminal_offline

方式二: Celery Beat（推荐）

# celery.py
app.conf.beat_schedule = {
    'check-terminal-offline': {
        'task': 'apps.datasets.tasks.check_terminal_offline',
        'schedule': 30.0,
    },
}

6.2 边缘模拟器部署

# 安装依赖
pip install -r edge_simulator/requirements.txt

# 运行模拟器（前台）
python edge_simulator/main.py

# 运行模拟器（后台）
nohup python edge_simulator/main.py > /dev/null 2>&1 &

6.3 监控指标

指标	说明	告警阈值
在线设备数	online_status=1 的设备数	低于预期值的 80%
离线设备数	online_status=0 的设备数	超过预期值的 20%
心跳延迟	心跳时间间隔	超过 60 秒
模型更新成功率	成功更新数/总更新数	低于 90%
Token 续期失败率	Token 获取失败次数	连续 3 次失败

---

七、时序图

7.1 Token 获取流程

边缘设备                           云端服务器
   │                                  │
   │  POST /server/terminal/auth/     │
   │  {"stcd":"DEVICE001","secret":"**"}│
   │ ─────────────────────────────>    │
   │                                  │── 验证设备编码与密钥
   │                                  │── 生成JWT Token (24h)
   │     200 {access_token, ...}    │
   │ <─────────────────────────────    │
   │                                  │
   │── 保存Token到本地文件              │

7.2 心跳 + 版本检查 + 模型更新完整流程

边缘设备                           云端服务器                     前端管理页
   │                                  │                              │
   │═══ 心跳循环 (30s/次) ═══════════│                              │
   │  PUT /heartbeat/ [Token]         │                              │
   │ ─────────────────────────────>    │── 更新 last_heartbeat_time    │
   │  200 OK                          │                              │
   │ <─────────────────────────────    │                              │
   │                                  │                              │
   │═══ 离线检测 (云端定时30s) ═══════│                              │
   │                                  │── 检测超时5分钟的设备          │
   │                                  │── online_status → 0           │
   │                                  │── WebSocket推送 ───────────> │
   │                                  │   {"event":"offline", ...}    │
   │                                  │                              │── 更新UI
   │                                  │                              │
   │═══ 版本检查 (定时) ══════════════│                              │
   │  GET /checkVersion/              │                              │
   │ ─────────────────────────────>    │── 对比 model_version          │
   │  200 {having_newmodel:1, ...}    │    与 edge_model_version      │
   │ <─────────────────────────────    │                              │
   │                                  │                              │
   │═══ 模型下载 ═════════════════════│                              │
   │── 备份旧模型到 backup/           │                              │
   │                                  │                              │
   │  GET /downloadModel/             │                              │
   │ ─────────────────────────────>    │── 从MinIO获取模型文件         │
   │  200 (文件流)                     │                              │
   │ <─────────────────────────────    │                              │
   │── 保存到 models/current/         │                              │
   │── 更新本地 version.txt           │                              │
   │                                  │                              │
   │  PUT /updateModelVersion/        │                              │
   │ ─────────────────────────────>    │── 更新 edge_model_version     │
   │  200 OK                          │                              │
   │ <─────────────────────────────    │                              │

---

八、实施清单

云端新增开发项

• 新增 EdgeDeviceAuthentication 认证类
• 新增 POST /server/terminal/auth/ 设备认证接口
• 新增 GET /server/terminal/checkVersion/ 版本检查接口
• 新增 PUT /server/terminal/updateModelVersion/ 版本更新回调接口
• 修改 PUT /server/terminal/heartbeat/ 增加设备 Token 认证
• 新增 check_terminal_offline 管理命令或 Celery 定时任务
• 新增 WebSocket Consumer (TerminalStatusConsumer) 推送在线状态
• 新增 ws/terminal/status/ WebSocket 路由
• 前端管理页面订阅 WebSocket，实时更新终端在线状态

边缘模拟器开发项

• 创建 edge_simulator/ 目录结构
• 实现 auth.py Token 管理模块
• 实现 heartbeat.py 心跳上报模块
• 实现 version_check.py 版本检查与下载模块
• 实现 main.py 主调度程序
• 编写 requirements.txt 依赖文件
• 编写 Docker 部署配置（可选）

---

九、行业最佳实践参考

1. OTA 更新策略（参考物联网 OTA 标准）

   • 差量更新 vs 全量更新：当前采用全量下载，未来可扩展为增量更新
   • 灰度发布：支持按终端分类（sttp）或指定设备列表分批推送版本
   • 回滚机制：备份旧版本，支持云端指令触发回滚

2. 连接可靠性（参考 MQTT/CoAP 协议设计）

   • 断线重连：边缘端实现指数退避重连（1s → 2s → 4s → ... → max 60s）
   • 离线缓存：心跳失败时的状态记录，恢复后补报
   • 连接池：单设备单连接，避免资源浪费

3. 安全性（参考 IoT 安全最佳实践）

   • 设备唯一身份：stcd + 预置密钥组成设备指纹
   • 通信加密：TLS 1.2+
   • 最小权限：设备 Token 仅能访问自己的数据和接口
   • 定期轮换密钥（可选扩展）