120 lines
4.7 KiB
Python
120 lines
4.7 KiB
Python
# ============================================================
|
||
# 外部算法排产桩（moduleId: integrations-algo-stub, 可重生 ✅）
|
||
# local://stub 或 HTTP /schedule：简单串行排到首台可用设备
|
||
# ============================================================
|
||
from __future__ import annotations
|
||
|
||
import hashlib
|
||
from typing import Any
|
||
|
||
from server.aps_domain.scheduling_dto import (
|
||
SchedulingProblem,
|
||
SchedulingSolution,
|
||
SolutionConflict,
|
||
SolutionOp,
|
||
)
|
||
from server.timeutil import add_minutes, fmt_dt, parse_dt
|
||
|
||
|
||
def _stub_input_fingerprint(p: SchedulingProblem) -> str:
|
||
"""输入指纹：problemId + 订单/工艺/资源稳定投影（可复算 runId 用）。"""
|
||
import json as _json
|
||
|
||
payload = {
|
||
# 注意：problemId 含时间戳（运行元数据），若纳入指纹会使同输入两次运行
|
||
# runId 不同、trace 链不可复算——故只投影确定性业务输入（矩阵 114 行可复算）。
|
||
"orders": sorted(
|
||
(o.orderId, o.productCode, o.quantity, str(o.dueDate or ""))
|
||
for o in p.orders
|
||
),
|
||
"routings": sorted(
|
||
(s.productCode, s.seq, s.operationCode, s.stdTimePerUnit)
|
||
for s in p.routings
|
||
),
|
||
"resources": sorted((r.code, tuple(sorted(r.capabilities or []))) for r in p.resources),
|
||
}
|
||
return _json.dumps(payload, ensure_ascii=True, sort_keys=True, separators=(",", ":"))
|
||
|
||
|
||
def solve_problem(problem: SchedulingProblem | dict[str, Any]) -> SchedulingSolution:
|
||
"""确定性桩算法：按订单×工序顺序，选能力匹配的第一台资源串行占槽。"""
|
||
p = problem if isinstance(problem, SchedulingProblem) else SchedulingProblem(**problem)
|
||
cursor = parse_dt(p.startTime)
|
||
ops_out: list[SolutionOp] = []
|
||
conflicts: list[SolutionConflict] = []
|
||
|
||
# 资源占用游标
|
||
res_free: dict[str, Any] = {r.code: cursor for r in p.resources}
|
||
caps: dict[str, list[str]] = {r.code: list(r.capabilities or []) for r in p.resources}
|
||
|
||
route_by_product: dict[str, list] = {}
|
||
for step in sorted(p.routings, key=lambda s: (s.productCode, s.seq)):
|
||
route_by_product.setdefault(step.productCode, []).append(step)
|
||
|
||
for order in sorted(p.orders, key=lambda o: (o.dueDate or "9999", o.priority, o.orderId)):
|
||
steps = route_by_product.get(order.productCode) or []
|
||
if not steps:
|
||
conflicts.append(SolutionConflict(
|
||
type="NO_ROUTING", orderId=order.orderId,
|
||
description=f"订单 {order.orderId} 无工艺路线", severity="high"))
|
||
continue
|
||
for step in steps:
|
||
candidates = [
|
||
code for code, cap in caps.items()
|
||
if not cap or step.operationCode in cap
|
||
]
|
||
if not candidates:
|
||
candidates = list(res_free.keys())
|
||
if not candidates:
|
||
conflicts.append(SolutionConflict(
|
||
type="NO_CAPABILITY", orderId=order.orderId,
|
||
description=f"无设备可做 {step.operationCode}", severity="high"))
|
||
continue
|
||
# 选最早空闲
|
||
chosen = min(candidates, key=lambda c: res_free[c])
|
||
start_t = res_free[chosen]
|
||
run_min = max(1.0, float(step.stdTimePerUnit) * float(order.quantity))
|
||
end_t = add_minutes(start_t, int(run_min))
|
||
ops_out.append(SolutionOp(
|
||
orderId=order.orderId, seq=step.seq,
|
||
operationCode=step.operationCode,
|
||
operationName=step.operationName or step.operationCode,
|
||
resourceCode=chosen,
|
||
start=fmt_dt(start_t), end=fmt_dt(end_t),
|
||
runMin=run_min,
|
||
))
|
||
res_free[chosen] = end_t
|
||
|
||
return SchedulingSolution(
|
||
problemId=p.problemId,
|
||
runId=hashlib.sha256(_stub_input_fingerprint(p).encode()).hexdigest()[:12],
|
||
skillId="algo.stub",
|
||
status="FEASIBLE" if ops_out else "INFEASIBLE",
|
||
operations=ops_out,
|
||
conflicts=conflicts,
|
||
kpi={"avgUtilization": 0.55, "opCount": len(ops_out)},
|
||
solverMeta={"engine": "stub", "note": "local deterministic stub"},
|
||
)
|
||
|
||
|
||
def create_stub_app():
|
||
"""可选：独立 FastAPI 服务（python -m server.integrations.algo_skill_stub）。"""
|
||
from fastapi import FastAPI
|
||
app = FastAPI(title="APS Algo Skill Stub")
|
||
|
||
@app.get("/health")
|
||
def health():
|
||
return {"ok": True, "skill": "algo.stub"}
|
||
|
||
@app.post("/schedule")
|
||
def schedule(body: dict):
|
||
sol = solve_problem(body)
|
||
return sol.model_dump()
|
||
|
||
return app
|
||
|
||
|
||
if __name__ == "__main__":
|
||
import uvicorn
|
||
uvicorn.run(create_stub_app(), host="127.0.0.1", port=8101)
|