18 KiB
18 KiB
库位管理开发文档
概述
库位管理模块是VWED任务系统的核心组件之一，提供对库位（基于动作点分层）的全面管理功能。本文档详细介绍了库位管理的API接口、内部实现、数据模型以及开发指南。
架构概述
系统架构
┌─────────────────┐ ┌─────────────────┐ ┌─────────────────┐
│ API 层 │ │ 服务层 │ │ 数据层 │
│ (Routes) │───▶│ (Services) │───▶│ (Models) │
│ │ │ │ │ │
│ operate_point_ │ │ OperatePoint │ │ OperatePoint │
│ api.py │ │ Service │ │ Layer │
└─────────────────┘ └─────────────────┘ └─────────────────┘
核心组件
- API路由层 (
routes/operate_point_api.py): FastAPI路由定义 - 服务层 (
services/operate_point_service.py): 业务逻辑处理 - 数据模型层 (
data/models/): ORM模型定义 - 请求/响应模型 (
routes/model/operate_point_model.py): API数据模型
数据模型
主要数据表
OperatePointLayer (库位层表)
CREATE TABLE operate_point_layer (
id VARCHAR(255) PRIMARY KEY, -- 库位ID
operate_point_id VARCHAR(255), -- 动作点ID (外键)
layer_index INTEGER, -- 层索引
layer_name VARCHAR(255), -- 库位名称
is_occupied BOOLEAN DEFAULT FALSE, -- 是否占用
is_locked BOOLEAN DEFAULT FALSE, -- 是否锁定
is_disabled BOOLEAN DEFAULT FALSE, -- 是否禁用
is_empty_tray BOOLEAN DEFAULT FALSE, -- 是否空托盘
locked_by VARCHAR(255), -- 锁定者
goods_content TEXT, -- 货物内容
goods_weight INTEGER, -- 货物重量(克)
goods_volume INTEGER, -- 货物体积(立方厘米)
max_weight INTEGER, -- 最大承重(克)
max_volume INTEGER, -- 最大体积(立方厘米)
layer_height INTEGER, -- 层高(毫米)
tags TEXT, -- 标签
description TEXT, -- 描述
config_json JSONB, -- 扩展字段配置
created_at TIMESTAMP, -- 创建时间
updated_at TIMESTAMP, -- 更新时间
is_deleted BOOLEAN DEFAULT FALSE -- 软删除标记
);
OperatePoint (动作点表)
CREATE TABLE operate_point (
id VARCHAR(255) PRIMARY KEY, -- 动作点ID
station_name VARCHAR(255), -- 站点名称
scene_id VARCHAR(255), -- 场景ID
storage_area_id VARCHAR(255), -- 库区ID
max_layers INTEGER, -- 最大层数
current_layers INTEGER, -- 当前层数
position_x DECIMAL(10,3), -- X坐标
position_y DECIMAL(10,3), -- Y坐标
position_z DECIMAL(10,3), -- Z坐标
description TEXT, -- 描述
created_at TIMESTAMP, -- 创建时间
updated_at TIMESTAMP, -- 更新时间
is_deleted BOOLEAN DEFAULT FALSE -- 软删除标记
);
ExtendedProperty (扩展属性表)
CREATE TABLE extended_property (
id VARCHAR(255) PRIMARY KEY, -- 属性ID
property_key VARCHAR(255) UNIQUE, -- 属性键名
property_name VARCHAR(255), -- 属性名称
property_type VARCHAR(50), -- 属性类型
is_required BOOLEAN DEFAULT FALSE, -- 是否必填
is_enabled BOOLEAN DEFAULT TRUE, -- 是否启用
default_value TEXT, -- 默认值
description TEXT, -- 描述
category VARCHAR(100), -- 分类
validation_rules JSONB, -- 验证规则
options JSONB, -- 选项(select类型使用)
created_at TIMESTAMP, -- 创建时间
updated_at TIMESTAMP, -- 更新时间
is_deleted BOOLEAN DEFAULT FALSE -- 软删除标记
);
StorageLocationLog (库位操作日志表)
CREATE TABLE storage_location_log (
id VARCHAR(255) PRIMARY KEY, -- 日志ID
operation_time TIMESTAMP, -- 操作时间
operator VARCHAR(255), -- 操作人
operation_type VARCHAR(100), -- 操作类型
affected_storage_locations JSONB, -- 受影响的库位列表
description TEXT, -- 操作描述
request_data JSONB, -- 请求数据
created_at TIMESTAMP -- 创建时间
);
API接口详解
接口分类
库位管理API主要分为以下几类：
- 库位查询接口 - 获取库位列表、详情、状态
- 库位操作接口 - 更新库位状态、编辑库位信息
- 扩展属性接口 - 管理自定义属性定义
- 操作记录接口 - 查询操作历史记录
核心接口实现
1. 获取库位列表 (GET /api/vwed-operate-point/list)
实现流程：
@router.get("/list")
async def get_storage_location_list(
# 查询参数
scene_id: Optional[str] = None,
storage_area_id: Optional[str] = None,
# ... 其他参数
db: Session = Depends(get_db)
):
# 1. 参数验证和构建请求对象
request = StorageLocationListRequest(...)
# 2. 调用服务层方法
result = OperatePointService.get_storage_location_list(db, request)
# 3. 返回格式化响应
return api_response(message="查询成功", data=result)
服务层实现关键点：
- 动态查询构建：根据请求参数动态构建SQLAlchemy查询
- 关联查询优化：只在需要时关联OperatePoint表
- 自定义排序：支持层名称和站点名称的智能排序
- 分页处理：内存中排序后再分页
- 数据转换：将ORM对象转换为响应模型
def get_storage_location_list(db: Session, request: StorageLocationListRequest):
# 构建基础查询
query = db.query(OperatePointLayer).filter(
OperatePointLayer.is_deleted == False
)
# 条件过滤
if request.scene_id or request.storage_area_id:
query = query.join(OperatePoint)
# 应用自定义排序
storage_locations = query.all()
if request.layer_name_sort is not None:
storage_locations = _apply_custom_sorting(
storage_locations, request.layer_name_sort, request.station_name_sort
)
# 分页和数据转换
# ...
2. 更新库位状态 (PUT /api/vwed-operate-point/status)
支持的操作类型：
class StorageLocationActionEnum(str, Enum):
OCCUPY = "occupy" # 占用库位
RELEASE = "release" # 释放库位
LOCK = "lock" # 锁定库位
UNLOCK = "unlock" # 解锁库位
ENABLE = "enable" # 启用库位
DISABLE = "disable" # 禁用库位
SET_EMPTY_TRAY = "set_empty_tray" # 设置为空托盘
CLEAR_EMPTY_TRAY = "clear_empty_tray" # 清除空托盘状态
实现流程：
def update_storage_location_status(db: Session, request: StorageLocationStatusUpdateRequest):
# 1. 查找库位
storage_location = db.query(OperatePointLayer).filter(
OperatePointLayer.layer_name == request.layer_name,
OperatePointLayer.is_deleted == False
).first()
# 2. 验证操作合法性
if not storage_location:
raise ValueError("库位不存在")
# 3. 执行状态更新
old_status = _get_storage_location_status(storage_location)
_apply_action(storage_location, request.action, request.locked_by)
# 4. 记录操作日志
_log_operation(request.layer_name, request.action, request.operator, request.reason)
# 5. 提交事务
db.commit()
return StatusUpdateResult(...)
3. 扩展属性管理
扩展属性系统允许用户为库位定义自定义字段，具有以下特点：
- 动态属性定义：支持多种数据类型
- 全局同步：创建/删除属性会同步到所有库位
- 验证规则：支持自定义验证逻辑
- UI友好：支持显示格式和选项配置
创建扩展属性流程：
def create_extended_property(db: Session, request: ExtendedPropertyCreateRequest):
# 1. 验证属性名唯一性
existing = db.query(ExtendedProperty).filter(
ExtendedProperty.property_key == request.property_key
).first()
if existing:
raise ValueError("属性键名已存在")
# 2. 创建属性定义
property_obj = ExtendedProperty(...)
db.add(property_obj)
# 3. 同步到所有库位
layers = db.query(OperatePointLayer).filter(
OperatePointLayer.is_deleted == False
).all()
for layer in layers:
config = layer.config_json or {}
config[request.property_key] = request.default_value
layer.config_json = config
# 4. 提交事务
db.commit()
return CreateResult(affected_layers_count=len(layers))
核心功能详解
1. 智能排序功能
系统支持对库位名称和站点名称进行智能排序，能够正确处理字母数字混合的名称：
def _apply_custom_sorting(storage_locations, layer_name_sort, station_name_sort):
"""
智能排序算法：
1. 解析名称中的字母和数字部分
2. 先按字母部分排序
3. 字母相同时按数字部分逐位比较
示例：AP1, AP2, AP10, AP11, B1, B2
"""
import re
def _parse_alphanumeric_name(name):
parts = re.findall(r'([A-Za-z]+)|(\d+)', name)
letters = ""
numbers = []
for letter_part, number_part in parts:
if letter_part:
letters += letter_part
elif number_part:
numbers.append(int(number_part))
return (letters.upper(), numbers)
# 实现排序比较逻辑
# ...
2. 批量操作机制
批量状态更新支持以下特性：
- 事务一致性：所有操作在同一事务中执行
- 部分成功处理：记录每个库位的操作结果
- 冲突检测：检测并报告操作冲突
- 性能优化：批量查询和更新
def batch_update_storage_location_status(db: Session, request: BatchStorageLocationStatusUpdateRequest):
results = []
# 批量查询所有库位
storage_locations = db.query(OperatePointLayer).filter(
OperatePointLayer.layer_name.in_(request.layer_names),
OperatePointLayer.is_deleted == False
).all()
# 创建名称到对象的映射
location_map = {sl.layer_name: sl for sl in storage_locations}
# 处理每个库位
for layer_name in request.layer_names:
try:
if layer_name not in location_map:
results.append(OperationResult(
layer_name=layer_name,
success=False,
message="库位不存在"
))
continue
# 执行操作
storage_location = location_map[layer_name]
result = _apply_single_action(storage_location, request.action, request.locked_by)
results.append(result)
except Exception as e:
results.append(OperationResult(
layer_name=layer_name,
success=False,
message=str(e)
))
# 提交事务
db.commit()
return BatchUpdateResult(results=results)
3. 扩展字段管理
扩展字段系统的设计要点：
- 存储机制：使用JSONB字段存储扩展属性值
- 类型系统：支持string、integer、float、boolean、date等类型
- 验证系统：支持必填验证、格式验证、范围验证
- UI集成：提供显示格式、宽度等UI配置
# 扩展字段配置示例
{
"temperature": {
"value": 25.5,
"type": "float",
"display_format": "{{value}}°C",
"validation": {
"min": -50,
"max": 100
}
},
"product_category": {
"value": "electronics",
"type": "select",
"options": [
{"value": "electronics", "label": "电子产品"},
{"value": "machinery", "label": "机械配件"}
]
}
}
4. 操作日志系统
所有库位操作都会自动记录到操作日志中：
def _log_storage_location_operation(
layer_names: List[str],
operation_type: str,
operator: str,
description: str,
request_data: Dict = None
):
"""记录库位操作日志"""
log_entry = StorageLocationLog(
id=generate_uuid(),
operation_time=datetime.now(),
operator=operator,
operation_type=operation_type,
affected_storage_locations=layer_names,
description=description,
request_data=request_data or {}
)
db.add(log_entry)
# 注意：不在此处提交事务，由调用方统一提交
开发指南
1. 添加新的操作类型
要添加新的库位操作类型，需要：
- 更新枚举定义：
class StorageLocationActionEnum(str, Enum):
# 现有操作...
NEW_ACTION = "new_action" # 新操作
- 实现操作逻辑：
def _apply_new_action(storage_location: OperatePointLayer, **kwargs):
"""实现新操作的具体逻辑"""
# 验证前置条件
if not _can_perform_new_action(storage_location):
raise ValueError("无法执行新操作")
# 执行状态更新
storage_location.some_field = new_value
storage_location.updated_at = datetime.now()
return "新操作执行成功"
- 更新操作描述：
def get_action_descriptions():
descriptions = {
# 现有描述...
StorageLocationActionEnum.NEW_ACTION: "执行新操作"
}
return descriptions
2. 扩展字段类型支持
要添加新的扩展字段类型：
- 更新类型枚举：
class ExtendedPropertyTypeEnum(str, Enum):
# 现有类型...
CUSTOM_TYPE = "custom_type"
- 实现验证逻辑：
def _validate_custom_type_value(value, property_config):
"""验证自定义类型的值"""
# 实现验证逻辑
pass
- 更新UI配置：
def _get_custom_type_ui_config(property_config):
"""返回自定义类型的UI配置"""
return {
"component": "CustomTypeInput",
"props": {
# UI组件属性
}
}
3. 性能优化建议
-
查询优化：
- 合理使用数据库索引
- 避免N+1查询问题
- 使用批量查询替代循环查询
-
内存管理：
- 大量数据查询时使用分页
- 及时释放不需要的对象引用
- 考虑使用数据流处理
-
缓存策略：
- 缓存扩展属性配置
- 缓存常用的库位状态查询
- 使用Redis进行分布式缓存
4. 错误处理最佳实践
# 统一错误处理模式
try:
# 业务逻辑
result = perform_operation()
return success_response(result)
except ValueError as e:
# 业务逻辑错误（客户端错误）
logger.warning(f"业务逻辑错误: {str(e)}")
return error_response(str(e), 400)
except PermissionError as e:
# 权限错误
logger.warning(f"权限不足: {str(e)}")
return error_response("权限不足", 403)
except Exception as e:
# 系统错误（服务器错误）
logger.error(f"系统错误: {str(e)}", exc_info=True)
return error_response("系统内部错误", 500)
5. 测试策略
- 单元测试：
def test_update_storage_location_status():
# 准备测试数据
storage_location = create_test_storage_location()
request = StorageLocationStatusUpdateRequest(
layer_name="test-layer",
action="occupy"
)
# 执行测试
result = OperatePointService.update_storage_location_status(db, request)
# 验证结果
assert result.success == True
assert storage_location.is_occupied == True
- 集成测试：
def test_storage_location_api_integration():
# 测试完整的API调用流程
response = client.put("/api/vwed-operate-point/status", json={
"layer_name": "test-layer",
"action": "occupy"
})
assert response.status_code == 200
assert response.json()["code"] == 200
部署和维护
1. 数据库迁移
当需要修改数据表结构时：
# 创建迁移文件
python scripts/generate_migration.py
# 应用迁移
python scripts/run_migration.py
2. 监控和日志
关键指标监控：
- API响应时间
- 数据库查询性能
- 错误率统计
- 库位操作频率
日志配置：
# 在utils/logger.py中配置
logger = get_logger("services.operate_point_service")
logger.info("库位状态更新", extra={
"layer_name": layer_name,
"action": action,
"operator": operator
})
3. 性能调优
数据库优化：
-- 创建必要的索引
CREATE INDEX idx_operate_point_layer_name ON operate_point_layer(layer_name);
CREATE INDEX idx_operate_point_layer_status ON operate_point_layer(is_occupied, is_locked, is_disabled);
CREATE INDEX idx_storage_location_log_time ON storage_location_log(operation_time);
总结
库位管理模块是一个功能完整、设计良好的系统组件，提供了：
- 完整的CRUD操作：支持库位的查询、创建、更新、删除
- 灵活的状态管理：支持多种库位状态操作
- 可扩展的属性系统：允许用户定义自定义字段
- 完整的操作审计：记录所有操作历史
- 高性能的查询：支持复杂查询和智能排序
- 良好的错误处理：统一的错误处理和响应格式
该模块为AMR调度系统提供了可靠的库位管理基础，支持高并发、高可用的生产环境需求。