model_train_dm/backend/apps/cloud_terminal/smart_terminal.py

462 lines
18 KiB
Python
Raw Permalink Normal View History

2026-07-27 17:51:49 +08:00
"""智能终端(云端孪生镜像)— API 视图"""
import uuid
from datetime import datetime, timedelta
from django.conf import settings
from django.db.models import Q
from django.http import FileResponse, JsonResponse, StreamingHttpResponse
from django.shortcuts import get_object_or_404
from django.utils.timezone import now
from minio import Minio
from rest_framework import status
from rest_framework.authentication import BaseAuthentication
from rest_framework.decorators import api_view, authentication_classes, permission_classes
from rest_framework.exceptions import AuthenticationFailed
from rest_framework.pagination import PageNumberPagination
from rest_framework.permissions import BasePermission, IsAuthenticated
from rest_framework.response import Response
from rest_framework_simplejwt.exceptions import InvalidToken, TokenError
from rest_framework_simplejwt.tokens import AccessToken
from apps.common.serializers import AiSmartTerminalSerializer
from apps.core.models import AiSmartTerminal, AiAlgorithmModels
# ─── 设备认证 ──────────────────────────────────────────────────────────
class EdgeDeviceAuthentication(BaseAuthentication):
"""边缘设备 JWT Token 认证类
验证 Token 中的 stcd 与请求中的 stcd 一致确保设备只能操作自己的数据
"""
keyword = "Bearer"
def authenticate(self, request):
auth_header = request.META.get("HTTP_AUTHORIZATION", "")
if not auth_header.startswith(self.keyword):
return None
raw_token = auth_header[len(self.keyword):].strip()
try:
token = AccessToken(raw_token)
except (InvalidToken, TokenError) as e:
raise AuthenticationFailed(f"无效的认证令牌: {str(e)}")
token_type = token.get("type") or token.get("sub")
if token_type != "device":
raise AuthenticationFailed("此令牌不是设备认证令牌")
token_stcd = token.get("stcd")
if not token_stcd:
raise AuthenticationFailed("令牌中缺少设备编码(stcd)")
# 验证请求中的 stcd 与 Token 中的一致
request_stcd = (
request.GET.get("stcd")
or (request.data.get("stcd") if hasattr(request, "data") else None)
)
if request_stcd and token_stcd != request_stcd:
raise AuthenticationFailed("令牌中的设备编码与请求不匹配")
terminal = AiSmartTerminal.objects.filter(stcd=token_stcd).first()
if terminal is None:
raise AuthenticationFailed(f"设备未注册: {token_stcd}")
return (terminal, token)
class EdgeDevicePermission(BasePermission):
"""边缘设备请求权限 — 仅允许通过 EdgeDeviceAuthentication 认证的请求"""
def has_permission(self, request, view):
return bool(request.user and isinstance(request.user, AiSmartTerminal))
def _minio_client():
return Minio(
settings.MINIO_ENDPOINT,
access_key=settings.MINIO_ACCESS_KEY,
secret_key=settings.MINIO_SECRET_KEY,
secure=settings.MINIO_SECURE,
)
# ─── 设备认证 — Token 生成 ────────────────────────────────────────────
@api_view(["POST"])
def terminal_device_auth(request):
"""设备接入认证 — 生成边缘设备 JWT Token有效期24小时"""
stcd = str(request.data.get("stcd", "")).strip()
if not stcd:
return Response({"error": "缺少必要参数stcd"}, status=status.HTTP_400_BAD_REQUEST)
terminal = AiSmartTerminal.objects.filter(stcd=stcd).first()
if terminal is None:
return Response({"error": f"未找到终端:{stcd}"}, status=status.HTTP_404_NOT_FOUND)
token = AccessToken()
token["stcd"] = stcd
token["type"] = "device"
token.set_exp(lifetime=timedelta(days=1))
return Response({
"code": 200,
"data": {
"access_token": str(token),
"token_type": "bearer",
"expires_in": 86400,
"stcd": stcd,
"stnm": terminal.stnm or "",
}
})
@api_view(["POST"])
@permission_classes([IsAuthenticated])
def create_smart_terminal(request):
"""创建智能终端设备"""
data = request.data.copy()
stcd = data.get("stcd")
if stcd and AiSmartTerminal.objects.filter(stcd=stcd).exists():
return Response({"error": f"终端编码 '{stcd}' 已存在"}, status=status.HTTP_400_BAD_REQUEST)
terminal_id = uuid.uuid4().hex
data["id"] = terminal_id
data["record_user"] = request.user.username
serializer = AiSmartTerminalSerializer(data=data)
if serializer.is_valid():
serializer.save()
return Response(serializer.data, status=status.HTTP_201_CREATED)
return Response({"error": "无效数据", "details": serializer.errors}, status=status.HTTP_400_BAD_REQUEST)
@api_view(["PUT"])
@permission_classes([IsAuthenticated])
def update_smart_terminal(request):
"""修改智能设备信息"""
terminal_id = request.data.get("id")
terminal = get_object_or_404(AiSmartTerminal, id=terminal_id)
# 不允许通过此接口修改模型发布字段和边缘同步字段
_protected_fields = {
"model_id", "model_version", "model_path",
"model_publish_time", "model_publish_user",
"edge_model_version", "edge_update_time",
"last_heartbeat_time", "online_status",
}
update_data = {k: v for k, v in request.data.items() if k not in _protected_fields}
serializer = AiSmartTerminalSerializer(instance=terminal, data=update_data, partial=True)
if serializer.is_valid():
serializer.save()
return Response(serializer.data)
return Response({"error": "无效数据", "details": serializer.errors}, status=status.HTTP_400_BAD_REQUEST)
@api_view(["PUT"])
@permission_classes([IsAuthenticated])
def publish_terminal_model(request):
"""发布更新设备算法模型 — 将训练好的模型发布到一个或多个云端孪生设备
支持传 id单个 ids列表 {"ids":["id1","id2"],"model_id":"...","model_version":"v1","model_path":"..."}
"""
# 归一化:支持 id单个或 ids列表
raw_ids = request.data.get("ids", [])
single_id = request.data.get("id")
if single_id:
raw_ids = [single_id]
if isinstance(raw_ids, str):
import json as _json
try:
raw_ids = _json.loads(raw_ids)
except Exception:
raw_ids = [raw_ids]
ids = [tid for tid in raw_ids if tid] if isinstance(raw_ids, list) else []
model_id = request.data.get("model_id")
# 查找算法模型
ai_model = get_object_or_404(AiAlgorithmModels, id=model_id)
model_version = ai_model.model_version or ""
model_path = ai_model.onnx_bucketpath or ai_model.model_bucketpath
if not model_path:
return Response({"error": "模型文件未存储在MinIO中"}, status=status.HTTP_400_BAD_REQUEST)
model_name = model_path.split("/")[-1]
if not ids:
return Response({"error": "缺少必要参数id 或 ids"}, status=status.HTTP_400_BAD_REQUEST)
if not all([model_id, model_version, model_path]):
return Response({"error": "缺少必要参数model_id, model_version, model_path"}, status=status.HTTP_400_BAD_REQUEST)
terminals = AiSmartTerminal.objects.filter(id__in=ids)
if not terminals.exists():
return Response({"error": "未找到匹配的终端设备"}, status=status.HTTP_404_NOT_FOUND)
publish_time = now()
publish_user = request.user.username
terminals.update(
model_id=model_id,
model_version=model_version,
model_path=model_path,
model_name=model_name,
model_publish_time=publish_time,
model_publish_user=publish_user,
)
updated = AiSmartTerminal.objects.filter(id__in=ids)
serializer = AiSmartTerminalSerializer(updated, many=True)
return Response({
"published_count": len(ids),
"terminals": serializer.data,
})
@api_view(["GET"])
def search_smart_terminals(request):
"""条件查询智能设备列表(无需登录,边缘端也要调用)"""
filter_kwargs = {}
stcd = request.GET.get("stcd", "")
stnm = request.GET.get("stnm", "")
sttp = request.GET.get("sttp", "")
model_name = request.GET.get("model_name", "")
online_status = request.GET.get("online_status", "")
usfl = request.GET.get("usfl", "")
is_deleted = request.GET.get("is_deleted", "0")
page_size = request.GET.get("page_size") or 10
if stcd:
filter_kwargs["stcd__icontains"] = stcd
if stnm:
filter_kwargs["stnm__icontains"] = stnm
if sttp:
filter_kwargs["sttp"] = sttp
if model_name:
filter_kwargs["model_name__icontains"] = model_name
if online_status:
filter_kwargs["online_status"] = int(online_status)
if usfl:
filter_kwargs["usfl"] = int(usfl)
filter_kwargs["is_deleted"] = is_deleted
queryset = AiSmartTerminal.objects.filter(**filter_kwargs).order_by("order_index", "stcd")
paginator = PageNumberPagination()
paginator.page_size = page_size
page = paginator.paginate_queryset(queryset, request)
serializer = AiSmartTerminalSerializer(page, many=True)
return paginator.get_paginated_response(serializer.data)
@api_view(["PUT"])
@authentication_classes([EdgeDeviceAuthentication])
@permission_classes([EdgeDevicePermission])
def update_terminal_heartbeat(request):
"""更新设备在线状态(边缘上报心跳)— 使用设备 Token 认证"""
terminal = request.user # EdgeDeviceAuthentication 已将 terminal 注入为 request.user
if not isinstance(terminal, AiSmartTerminal):
return Response({"error": "设备认证失败"}, status=status.HTTP_401_UNAUTHORIZED)
update_fields = ["last_heartbeat_time", "online_status"]
was_offline = terminal.online_status == 0
terminal.last_heartbeat_time = now()
terminal.online_status = 1
# 首次接入:标记数据已接入
if not terminal.dtin:
terminal.dtin = 1
terminal.dtin_tm = now()
update_fields.extend(["dtin", "dtin_tm"])
terminal.save(update_fields=update_fields)
# 终端从离线恢复上线时,通过 WebSocket 推送通知
if was_offline:
_push_terminal_online_notification(terminal)
return Response({
"code": 200,
"data": {
"stcd": terminal.stcd,
"online_status": terminal.online_status,
"last_heartbeat_time": terminal.last_heartbeat_time,
}
})
def _push_terminal_online_notification(terminal):
"""通过 WebSocket 向前端推送终端上线通知"""
try:
from asgiref.sync import async_to_sync
from channels.layers import get_channel_layer
channel_layer = get_channel_layer()
if channel_layer is not None:
async_to_sync(channel_layer.group_send)(
"terminal_status",
{
"type": "terminal.status",
"data": {
"event": "online",
"count": 1,
"terminals": [
{"stcd": terminal.stcd, "stnm": terminal.stnm}
],
},
},
)
except Exception:
pass
# ─── 模型版本检查 ──────────────────────────────────────────────────────
@api_view(["GET"])
@authentication_classes([EdgeDeviceAuthentication])
@permission_classes([EdgeDevicePermission])
def check_terminal_model_version(request):
"""检查云端模型版本 — 边缘端定时轮询是否有新版本模型"""
terminal = request.user
if not isinstance(terminal, AiSmartTerminal):
return Response({"error": "设备认证失败"}, status=status.HTTP_401_UNAUTHORIZED)
if not terminal.model_id:
return Response({
"code": 200,
"data": {
"having_newmodel": 0,
"model_version": "",
"sync_check_interval": terminal.sync_check_interval or 30,
}
})
ai_model = AiAlgorithmModels.objects.filter(id=terminal.model_id).first()
if ai_model is None:
return Response({
"code": 200,
"data": {
"having_newmodel": 0,
"model_version": terminal.model_version or "",
"sync_check_interval": terminal.sync_check_interval or 30,
}
})
edge_version = request.GET.get("edge_version", "")
cloud_version = ai_model.model_version or ""
having_newmodel = 0
if not edge_version or edge_version != cloud_version:
having_newmodel = 1
result = {
"code": 200,
"data": {
"having_newmodel": having_newmodel,
"sync_check_interval": terminal.sync_check_interval or 30,
}
}
if having_newmodel:
result["data"].update({
"model_name": terminal.model_name or "",
"model_version": cloud_version,
"model_path": terminal.model_path or "",
"model_publish_time": terminal.model_publish_time.isoformat() if terminal.model_publish_time else "",
})
else:
result["data"]["model_version"] = cloud_version
return Response(result)
@api_view(["PUT"])
@authentication_classes([EdgeDeviceAuthentication])
@permission_classes([EdgeDevicePermission])
def update_terminal_model_version(request):
"""更新边缘端模型版本 — 边缘端下载新模型后回调,更新 edge_model_version 和 edge_update_time"""
terminal = request.user
if not isinstance(terminal, AiSmartTerminal):
return Response({"error": "设备认证失败"}, status=status.HTTP_401_UNAUTHORIZED)
edge_model_version = str(request.data.get("edge_model_version", "")).strip()
if not edge_model_version:
return Response({"error": "缺少必要参数edge_model_version"}, status=status.HTTP_400_BAD_REQUEST)
terminal.edge_model_version = edge_model_version
terminal.edge_update_time = now()
terminal.save(update_fields=["edge_model_version", "edge_update_time"])
return Response({
"code": 200,
"data": {
"stcd": terminal.stcd,
"edge_model_version": terminal.edge_model_version,
"edge_update_time": terminal.edge_update_time,
}
})
# ─── 模型下载(流式) ──────────────────────────────────────────────────
def _stream_object(response, chunk_size: int = 1024 * 1024):
try:
while True:
chunk = response.read(chunk_size)
if not chunk:
break
yield chunk
finally:
response.close()
response.release_conn()
def download_terminal_model(request):
"""下载最新版本设备算法模型 — 边缘端在校验版本差异后,从此接口下载模型文件"""
stcd = str(request.GET.get("stcd") or "").strip()
model_format = request.GET.get("model_format") or request.GET.get("format", "onnx")
if not stcd:
return JsonResponse({"error": "缺少必要参数stcd"}, status=status.HTTP_400_BAD_REQUEST)
terminal = AiSmartTerminal.objects.filter(stcd=stcd).first()
if terminal is None:
return JsonResponse({"error": f"未找到终端:{stcd}"}, status=status.HTTP_404_NOT_FOUND)
if not terminal.model_id:
return JsonResponse({"error": "该终端尚未发布模型"}, status=status.HTTP_404_NOT_FOUND)
try:
from apps.common.minio_client import minio_storage
ai_model = AiAlgorithmModels.objects.filter(id=terminal.model_id).first()
if ai_model is None:
return JsonResponse(
{"error": f"终端已绑定的模型不存在:{terminal.model_id}"},
status=status.HTTP_404_NOT_FOUND,
)
if model_format == "onnx":
model_path = ai_model.onnx_bucketpath
if not model_path:
return JsonResponse({"error": "该模型尚未转换为ONNX格式"}, status=status.HTTP_404_NOT_FOUND)
else:
model_path = ai_model.model_bucketpath
if not model_path:
return JsonResponse({"error": "该模型没有原始模型文件"}, status=status.HTTP_404_NOT_FOUND)
parts = model_path.split('/')
if len(parts) < 2:
return JsonResponse({"error": f"模型存储路径无效:{model_path}"}, status=status.HTTP_400_BAD_REQUEST)
bucket_name = parts[0]
object_name = '/'.join(parts[1:])
if not minio_storage.object_exists(bucket_name, object_name):
return JsonResponse(
{"error": f"MinIO中未找到模型文件{bucket_name}/{object_name}"},
status=status.HTTP_404_NOT_FOUND,
)
response = minio_storage.get_object(bucket_name, object_name)
content_type = "application/octet-stream"
filename = object_name.rsplit("/", 1)[-1]
stream = StreamingHttpResponse(_stream_object(response), content_type=content_type)
stream["Content-Disposition"] = f'attachment; filename="{filename}"'
return stream
except Exception as e:
return JsonResponse({"error": f"模型文件下载失败:{str(e)}"}, status=status.HTTP_500_INTERNAL_SERVER_ERROR)
@api_view(["DELETE"])
@permission_classes([IsAuthenticated])
def delete_smart_terminal(request):
"""删除智能设备"""
terminal_id = request.GET.get("id")
if not terminal_id:
return Response({"error": "缺少必要参数id"}, status=status.HTTP_400_BAD_REQUEST)
terminal = get_object_or_404(AiSmartTerminal, id=terminal_id)
terminal.delete()
return Response({"message": "删除成功", "id": terminal_id})