#!/usr/bin/env python # -*- coding: utf-8 -*- """ 地图数据推送相关模型 用于处理地图数据推送时的动作点和库区信息 """ from typing import Optional, List, Dict, Any from pydantic import BaseModel, Field, validator from enum import Enum class StorageAreaTypeEnum(str, Enum): """库区类型枚举""" GENERAL = "general" DENSE = "dense" class OperatePointLayerData(BaseModel): """动作点分层数据""" layer_name: Optional[str] = Field(None, description="层名称") max_weight: Optional[int] = Field(None, ge=0, description="最大承重(克)") max_volume: Optional[int] = Field(None, ge=0, description="最大体积(立方厘米)") layer_height: Optional[int] = Field(None, ge=0, description="层高(毫米)") description: Optional[str] = Field(None, description="层描述") tags: Optional[str] = Field(None, description="层标签") class OperatePointData(BaseModel): """动作点数据""" station_name: str = Field(..., description="动作站点名称") area_name: Optional[str] = Field(None, description="所属库区名称") max_layers: int = Field(1, ge=1, description="最大层数") position_x: Optional[int] = Field(None, description="X坐标") position_y: Optional[int] = Field(None, description="Y坐标") position_z: Optional[int] = Field(None, description="Z坐标") content: Optional[str] = Field("", description="内容") tags: Optional[str] = Field("", description="标签") description: Optional[str] = Field(None, description="动作点描述") layers: Optional[List[OperatePointLayerData]] = Field(None, description="库位名称") @validator('layers') def validate_layers(cls, v, values): """验证分层数据""" if v is None: return v max_layers = values.get('max_layers', 1) if len(v) > max_layers: raise ValueError(f"分层数量({len(v)})不能超过最大层数({max_layers})") # 检查层名称是否重复 if v: layer_names = [layer.layer_name for layer in v if layer.layer_name] if len(layer_names) != len(set(layer_names)): raise ValueError("层名称不能重复") return v class StorageAreaData(BaseModel): """库区数据""" area_name: str = Field(..., description="库区名称") area_type: StorageAreaTypeEnum = Field(StorageAreaTypeEnum.GENERAL, description="库区类型") description: Optional[str] = Field(None, description="库区描述") tags: Optional[str] = Field(None, description="库区标签") select_logic: int = Field(..., description="选择库位逻辑 1 先进先出 2 先进后出 ") class MapDataPushRequest(BaseModel): """地图数据推送请求""" scene_id: str = Field(..., description="场景ID") storage_areas: List[StorageAreaData] = Field(..., description="库区数据列表") operate_points: List[OperatePointData] = Field(..., description="动作点数据列表") @validator('operate_points') def validate_operate_points(cls, v, values): """验证动作点数据""" if not v: return v # 检查站点名称是否重复 station_names = [point.station_name for point in v] if len(station_names) != len(set(station_names)): raise ValueError("动作站点名称不能重复") return v @validator('storage_areas') def validate_storage_areas(cls, v): """验证库区数据""" if not v: return v # 检查库区名称是否重复 area_names = [area.area_name for area in v] if len(area_names) != len(set(area_names)): raise ValueError("库区名称不能重复") return v class MapDataPushResponse(BaseModel): """地图数据推送响应""" scene_id: str = Field(..., description="场景ID") storage_areas_count: int = Field(..., description="创建的库区数量") operate_points_count: int = Field(..., description="创建的动作点数量") layers_count: int = Field(..., description="创建的分层数量") message: str = Field("推送成功,已覆盖现有数据", description="推送结果说明") class MapDataQueryRequest(BaseModel): """地图数据查询请求""" scene_id: str = Field(..., description="场景ID") area_type: Optional[StorageAreaTypeEnum] = Field(None, description="库区类型筛选") include_layers: bool = Field(True, description="是否包含分层数据") class MapDataQueryResponse(BaseModel): """地图数据查询响应""" scene_id: str = Field(..., description="场景ID") storage_areas: List[Dict[str, Any]] = Field(..., description="库区数据列表") operate_points: List[Dict[str, Any]] = Field(..., description="动作点数据列表") total_capacity: int = Field(0, description="总容量") used_capacity: int = Field(0, description="已使用容量") dense_areas_count: int = Field(0, description="密集库区数量") general_areas_count: int = Field(0, description="一般库区数量") total_layers: int = Field(0, description="总分层数量") occupied_layers: int = Field(0, description="已占用分层数量")