279 lines
9.7 KiB
Python
279 lines
9.7 KiB
Python
"""
|
||
MCP for Bexell — 试产排程评估及优化系统
|
||
使用 MCPServer 框架，4 个工具：设备产能查询、排产占用查询、品号负荷查询、试产产能评估
|
||
HTTP 模式，挂载在 /bexell/mcp 路径
|
||
"""
|
||
|
||
import json
|
||
from datetime import datetime
|
||
from typing import Dict, List
|
||
|
||
from sqlalchemy import func, text
|
||
from mcp.server import MCPServer
|
||
|
||
from auth import get_auth
|
||
from .db import get_db
|
||
from .models import EquipmentCapacity, ProductEquipmentLoad, ProductEquipmentSchedule
|
||
|
||
|
||
# 鉴权：独立 Bearer Token，API Key 来自 BEXELL_MCP_AUTH_API_KEY 环境变量
|
||
_auth, _token_verifier = get_auth(service="BEXELL")
|
||
|
||
bexell_server = MCPServer(
|
||
name="bexell-scheduling",
|
||
description="试产排程评估及优化系统",
|
||
version="1.0.0",
|
||
auth=_auth,
|
||
token_verifier=_token_verifier,
|
||
)
|
||
|
||
|
||
# ============================================================
|
||
# 工具 1: get_equipment_by_type — 按设备类型查询设备产能信息
|
||
# ============================================================
|
||
|
||
@bexell_server.tool()
|
||
async def get_equipment_by_type(equipment_type: str) -> List[Dict]:
|
||
"""
|
||
根据设备类型获取该类型所有设备的基础产能信息。
|
||
|
||
Args:
|
||
equipment_type: 设备类型，如"冲压"、"表面"、"检测"
|
||
|
||
Returns:
|
||
设备列表，每个设备包含：
|
||
- equipment_code: 设备编号
|
||
- equipment_name: 设备名称
|
||
- equipment_type: 设备类型
|
||
- max_capacity: 极限产能(H)
|
||
- standard_capacity: 标准产能(H)
|
||
- min_startup: 最小开机产能(H)
|
||
- line_change_cost: 换线成本系数
|
||
"""
|
||
db = next(get_db())
|
||
try:
|
||
rows = db.query(EquipmentCapacity).filter(
|
||
EquipmentCapacity.equipment_type == equipment_type
|
||
).order_by(EquipmentCapacity.equipment_code).all()
|
||
|
||
return [{
|
||
"equipment_code": r.equipment_code,
|
||
"equipment_name": r.equipment_name,
|
||
"equipment_type": r.equipment_type,
|
||
"max_capacity": float(r.max_capacity),
|
||
"standard_capacity": float(r.standard_capacity),
|
||
"min_startup": float(r.min_startup),
|
||
"line_change_cost": float(r.line_change_cost),
|
||
} for r in rows]
|
||
finally:
|
||
db.close()
|
||
|
||
|
||
# ============================================================
|
||
# 工具 2: get_schedule_occupied — 查询设备排产占用产能合计
|
||
# ============================================================
|
||
|
||
@bexell_server.tool()
|
||
async def get_schedule_occupied(equipment_type: str) -> List[Dict]:
|
||
"""
|
||
获取指定类型全部设备的已占用产能汇总。
|
||
|
||
Args:
|
||
equipment_type: 设备类型，如"冲压"、"表面"、"检测"
|
||
|
||
Returns:
|
||
设备占用列表，每个设备包含：
|
||
- equipment_code: 设备编号
|
||
- equipment_name: 设备名称
|
||
- total_occupied_capacity: 占用产能合计(H)
|
||
"""
|
||
db = next(get_db())
|
||
try:
|
||
rows = db.query(
|
||
ProductEquipmentSchedule.equipment_code,
|
||
ProductEquipmentSchedule.equipment_name,
|
||
func.coalesce(
|
||
func.sum(ProductEquipmentSchedule.occupied_capacity), 0
|
||
).label("total_occupied")
|
||
).filter(
|
||
ProductEquipmentSchedule.equipment_type == equipment_type
|
||
).group_by(
|
||
ProductEquipmentSchedule.equipment_code,
|
||
ProductEquipmentSchedule.equipment_name
|
||
).order_by(
|
||
ProductEquipmentSchedule.equipment_code
|
||
).all()
|
||
|
||
return [{
|
||
"equipment_code": r.equipment_code,
|
||
"equipment_name": r.equipment_name,
|
||
"total_occupied_capacity": float(r.total_occupied),
|
||
} for r in rows]
|
||
finally:
|
||
db.close()
|
||
|
||
|
||
# ============================================================
|
||
# 工具 3: get_product_load — 查询品号在各设备的标准产能
|
||
# ============================================================
|
||
|
||
@bexell_server.tool()
|
||
async def get_product_load(product_code: str, equipment_type: str) -> List[Dict]:
|
||
"""
|
||
获取指定品号在目标设备类型各设备上的标准产能。
|
||
|
||
Args:
|
||
product_code: 品号，如"A0003"
|
||
equipment_type: 设备类型，如"冲压"、"表面"、"检测"
|
||
|
||
Returns:
|
||
品号设备负荷列表，每条记录包含：
|
||
- product_code: 品号
|
||
- product_name: 品名
|
||
- equipment_code: 设备编号
|
||
- equipment_name: 设备名称
|
||
- standard_capacity: 该品号在该设备上的标准产能
|
||
"""
|
||
db = next(get_db())
|
||
try:
|
||
rows = db.query(ProductEquipmentLoad).join(
|
||
EquipmentCapacity,
|
||
ProductEquipmentLoad.equipment_code == EquipmentCapacity.equipment_code
|
||
).filter(
|
||
ProductEquipmentLoad.product_code == product_code,
|
||
EquipmentCapacity.equipment_type == equipment_type
|
||
).order_by(
|
||
ProductEquipmentLoad.equipment_code
|
||
).all()
|
||
|
||
return [{
|
||
"product_code": r.product_code,
|
||
"product_name": r.product_name,
|
||
"equipment_code": r.equipment_code,
|
||
"equipment_name": r.equipment_name,
|
||
"standard_capacity": float(r.standard_capacity),
|
||
} for r in rows]
|
||
finally:
|
||
db.close()
|
||
|
||
|
||
# ============================================================
|
||
# 工具 4: evaluate_trial_capacity — 一体化试产产能评估
|
||
# ============================================================
|
||
|
||
@bexell_server.tool()
|
||
async def evaluate_trial_capacity(
|
||
product_code: str,
|
||
trial_quantity: int
|
||
) -> Dict:
|
||
"""
|
||
一体化试产产能评估：计算指定品号在指定设备类型下所有设备的可用产能，
|
||
并推荐产能最大值设备。
|
||
|
||
计算公式：
|
||
可用产能 = 极限产能 − 占用产能合计 − 换线成本 − 最小开机×0.1 − 试产数量×品号标准产能
|
||
|
||
Args:
|
||
product_code: 品号，如"A0003"
|
||
trial_quantity: 试产数量，如 10
|
||
|
||
Returns:
|
||
评估结果字典，包含：
|
||
- product_code: 品号
|
||
- product_name: 品名
|
||
- trial_quantity: 试产数量
|
||
- evaluated_at: 评估时间
|
||
- results: 设备评估明细列表（按可用产能降序）
|
||
- recommended: 推荐设备（可用产能最大且>=0）
|
||
"""
|
||
db = next(get_db())
|
||
try:
|
||
# 查询品名
|
||
product = db.execute(text(
|
||
"SELECT product_name FROM product_info WHERE product_code = :code"
|
||
), {"code": product_code}).first()
|
||
|
||
if not product:
|
||
return {"error": f"品号 {product_code} 不存在"}
|
||
|
||
# 核心聚合 SQL
|
||
sql = text("""
|
||
SELECT
|
||
ec.equipment_code,
|
||
ec.equipment_name,
|
||
ec.equipment_type,
|
||
ec.max_capacity,
|
||
ec.line_change_cost,
|
||
ec.min_startup * 0.1 AS min_startup_buffer,
|
||
COALESCE(occ.total_occupied, 0) AS occupied_capacity,
|
||
pel.standard_capacity AS product_standard_capacity,
|
||
:trial_qty * pel.standard_capacity AS trial_demand,
|
||
(ec.max_capacity
|
||
- COALESCE(occ.total_occupied, 0)
|
||
- ec.line_change_cost
|
||
- ec.min_startup * 0.1
|
||
- :trial_qty * pel.standard_capacity
|
||
) AS available_capacity
|
||
FROM equipment_capacity ec
|
||
LEFT JOIN (
|
||
SELECT equipment_code, SUM(occupied_capacity) AS total_occupied
|
||
FROM product_equipment_schedule
|
||
GROUP BY equipment_code
|
||
) occ ON ec.equipment_code = occ.equipment_code
|
||
JOIN product_equipment_load pel
|
||
ON ec.equipment_code = pel.equipment_code
|
||
AND pel.product_code = :product_code
|
||
WHERE ec.equipment_type = (
|
||
SELECT equipment_type FROM equipment_capacity ec2
|
||
WHERE ec2.equipment_code = pel.equipment_code
|
||
LIMIT 1
|
||
)
|
||
ORDER BY available_capacity DESC
|
||
""")
|
||
|
||
rows = db.execute(sql, {
|
||
"product_code": product_code,
|
||
"trial_qty": trial_quantity
|
||
}).mappings().all()
|
||
|
||
# 组装结果
|
||
results = []
|
||
for idx, r in enumerate(rows, 1):
|
||
available = float(r.available_capacity)
|
||
results.append({
|
||
"rank": idx,
|
||
"equipment_code": r.equipment_code,
|
||
"equipment_name": r.equipment_name,
|
||
"equipment_type": r.equipment_type,
|
||
"max_capacity": float(r.max_capacity),
|
||
"occupied_capacity": float(r.occupied_capacity),
|
||
"line_change_cost": float(r.line_change_cost),
|
||
"min_startup_buffer": float(r.min_startup_buffer),
|
||
"product_standard_capacity": float(r.product_standard_capacity),
|
||
"trial_demand": float(r.trial_demand),
|
||
"available_capacity": available,
|
||
"status": "可用" if available >= 0 else "已满负荷",
|
||
})
|
||
|
||
# 推荐设备（第一个可用且 >= 0 的）
|
||
recommended = None
|
||
for r in results:
|
||
if r["available_capacity"] >= 0:
|
||
recommended = {
|
||
"equipment_code": r["equipment_code"],
|
||
"equipment_name": r["equipment_name"],
|
||
"available_capacity": r["available_capacity"],
|
||
}
|
||
break
|
||
|
||
return {
|
||
"product_code": product_code,
|
||
"product_name": product.product_name,
|
||
"trial_quantity": trial_quantity,
|
||
"evaluated_at": datetime.now().isoformat(),
|
||
"results": results,
|
||
"recommended": recommended,
|
||
}
|
||
finally:
|
||
db.close()
|