155 lines
6.3 KiB
Python
155 lines
6.3 KiB
Python
#!/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_index: int = Field(..., ge=1, description="层索引(从1开始)")
|
|
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="动作站点名称")
|
|
storage_location_name: str = Field(..., description="库位名称")
|
|
storage_area_id: Optional[str] = Field(None, description="所属库区ID")
|
|
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})")
|
|
|
|
# 检查层索引是否重复
|
|
layer_indices = [layer.layer_index for layer in v]
|
|
if len(layer_indices) != len(set(layer_indices)):
|
|
raise ValueError("层索引不能重复")
|
|
|
|
# 检查层索引是否超出范围
|
|
for layer in v:
|
|
if layer.layer_index > max_layers:
|
|
raise ValueError(f"层索引({layer.layer_index})不能超过最大层数({max_layers})")
|
|
|
|
return v
|
|
|
|
|
|
class StorageAreaData(BaseModel):
|
|
"""库区数据"""
|
|
id: str = Field(..., description="库区ID")
|
|
area_name: str = Field(..., description="库区名称")
|
|
area_code: str = Field(..., description="库区编码")
|
|
area_type: StorageAreaTypeEnum = Field(StorageAreaTypeEnum.GENERAL, description="库区类型")
|
|
description: Optional[str] = Field(None, description="库区描述")
|
|
tags: Optional[str] = Field(None, description="库区标签")
|
|
|
|
|
|
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("动作站点名称不能重复")
|
|
|
|
# 检查库区ID是否存在
|
|
storage_areas = values.get('storage_areas', [])
|
|
storage_area_ids = {area.id for area in storage_areas}
|
|
|
|
for point in v:
|
|
if point.storage_area_id and point.storage_area_id not in storage_area_ids:
|
|
raise ValueError(f"动作点'{point.station_name}'的库区ID'{point.storage_area_id}'不存在")
|
|
|
|
return v
|
|
|
|
@validator('storage_areas')
|
|
def validate_storage_areas(cls, v):
|
|
"""验证库区数据"""
|
|
if not v:
|
|
return v
|
|
|
|
# 检查库区ID是否重复
|
|
area_ids = [area.id for area in v]
|
|
if len(area_ids) != len(set(area_ids)):
|
|
raise ValueError("库区ID不能重复")
|
|
|
|
# 检查库区名称是否重复
|
|
area_names = [area.area_name for area in v]
|
|
if len(area_names) != len(set(area_names)):
|
|
raise ValueError("库区名称不能重复")
|
|
|
|
# 检查库区编码是否重复
|
|
area_codes = [area.area_code for area in v]
|
|
if len(area_codes) != len(set(area_codes)):
|
|
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="已占用分层数量") |