jzw/VWED_server
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
VWED_server/docs/storage_location_management_development.md

604 lines
18 KiB
Markdown
Raw Normal View History

新增 打包exe模块
2025-09-09 10:41:27 +08:00
# 库位管理开发文档
## 概述
库位管理模块是VWED任务系统的核心组件之一，提供对库位（基于动作点分层）的全面管理功能。本文档详细介绍了库位管理的API接口、内部实现、数据模型以及开发指南。
## 架构概述
### 系统架构
```
┌─────────────────┐ ┌─────────────────┐ ┌─────────────────┐
│ API 层 │ │ 服务层 │ │ 数据层 │
│ (Routes) │───▶│ (Services) │───▶│ (Models) │
│ │ │ │ │ │
│ operate_point_ │ │ OperatePoint │ │ OperatePoint │
│ api.py │ │ Service │ │ Layer │
└─────────────────┘ └─────────────────┘ └─────────────────┘
```
### 核心组件
1. **API路由层** (`routes/operate_point_api.py`): FastAPI路由定义
2. **服务层** (`services/operate_point_service.py`): 业务逻辑处理
3. **数据模型层** (`data/models/`): ORM模型定义
4. **请求/响应模型** (`routes/model/operate_point_model.py`): API数据模型
## 数据模型
### 主要数据表
#### OperatePointLayer (库位层表)
```sql
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 (动作点表)
```sql
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 (扩展属性表)
```sql
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 (库位操作日志表)
```sql
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. **库位查询接口** - 获取库位列表、详情、状态
2. **库位操作接口** - 更新库位状态、编辑库位信息
3. **扩展属性接口** - 管理自定义属性定义
4. **操作记录接口** - 查询操作历史记录
### 核心接口实现
#### 1. 获取库位列表 (`GET /api/vwed-operate-point/list`)
**实现流程：**
```python
@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)
```
**服务层实现关键点：**
1. **动态查询构建**：根据请求参数动态构建SQLAlchemy查询
2. **关联查询优化**：只在需要时关联OperatePoint表
3. **自定义排序**：支持层名称和站点名称的智能排序
4. **分页处理**：内存中排序后再分页
5. **数据转换**：将ORM对象转换为响应模型
```python
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`)
**支持的操作类型：**
```python
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" # 清除空托盘状态
```
**实现流程：**
```python
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. 扩展属性管理
扩展属性系统允许用户为库位定义自定义字段，具有以下特点：
1. **动态属性定义**：支持多种数据类型
2. **全局同步**：创建/删除属性会同步到所有库位
3. **验证规则**：支持自定义验证逻辑
4. **UI友好**：支持显示格式和选项配置
**创建扩展属性流程：**
```python
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. 智能排序功能
系统支持对库位名称和站点名称进行智能排序，能够正确处理字母数字混合的名称：
```python
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. 批量操作机制
批量状态更新支持以下特性：
1. **事务一致性**：所有操作在同一事务中执行
2. **部分成功处理**：记录每个库位的操作结果
3. **冲突检测**：检测并报告操作冲突
4. **性能优化**：批量查询和更新
```python
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. 扩展字段管理
扩展字段系统的设计要点：
1. **存储机制**：使用JSONB字段存储扩展属性值
2. **类型系统**：支持string、integer、float、boolean、date等类型
3. **验证系统**：支持必填验证、格式验证、范围验证
4. **UI集成**：提供显示格式、宽度等UI配置
```python
# 扩展字段配置示例
{
"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. 操作日志系统
所有库位操作都会自动记录到操作日志中：
```python
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. 添加新的操作类型
要添加新的库位操作类型，需要：
1. **更新枚举定义**：
```python
class StorageLocationActionEnum(str, Enum):
# 现有操作...
NEW_ACTION = "new_action" # 新操作
```
2. **实现操作逻辑**：
```python
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 "新操作执行成功"
```
3. **更新操作描述**：
```python
def get_action_descriptions():
descriptions = {
# 现有描述...
StorageLocationActionEnum.NEW_ACTION: "执行新操作"
}
return descriptions
```
### 2. 扩展字段类型支持
要添加新的扩展字段类型：
1. **更新类型枚举**：
```python
class ExtendedPropertyTypeEnum(str, Enum):
# 现有类型...
CUSTOM_TYPE = "custom_type"
```
2. **实现验证逻辑**：
```python
def _validate_custom_type_value(value, property_config):
"""验证自定义类型的值"""
# 实现验证逻辑
pass
```
3. **更新UI配置**：
```python
def _get_custom_type_ui_config(property_config):
"""返回自定义类型的UI配置"""
return {
"component": "CustomTypeInput",
"props": {
# UI组件属性
}
}
```
### 3. 性能优化建议
1. **查询优化**：
- 合理使用数据库索引
- 避免N+1查询问题
- 使用批量查询替代循环查询
2. **内存管理**：
- 大量数据查询时使用分页
- 及时释放不需要的对象引用
- 考虑使用数据流处理
3. **缓存策略**：
- 缓存扩展属性配置
- 缓存常用的库位状态查询
- 使用Redis进行分布式缓存
### 4. 错误处理最佳实践
```python
# 统一错误处理模式
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. 测试策略
1. **单元测试**：
```python
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
```
2. **集成测试**：
```python
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
# 创建迁移文件
python scripts/generate_migration.py
# 应用迁移
python scripts/run_migration.py
```
### 2. 监控和日志
关键指标监控：
- API响应时间
- 数据库查询性能
- 错误率统计
- 库位操作频率
日志配置：
```python
# 在utils/logger.py中配置
logger = get_logger("services.operate_point_service")
logger.info("库位状态更新", extra={
"layer_name": layer_name,
"action": action,
"operator": operator
})
```
### 3. 性能调优
数据库优化：
```sql
-- 创建必要的索引
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);
```
## 总结
库位管理模块是一个功能完整、设计良好的系统组件，提供了：
1. **完整的CRUD操作**：支持库位的查询、创建、更新、删除
2. **灵活的状态管理**：支持多种库位状态操作
3. **可扩展的属性系统**：允许用户定义自定义字段
4. **完整的操作审计**：记录所有操作历史
5. **高性能的查询**：支持复杂查询和智能排序
6. **良好的错误处理**：统一的错误处理和响应格式
该模块为AMR调度系统提供了可靠的库位管理基础，支持高并发、高可用的生产环境需求。
Reference in New Issue Copy Permalink