467 lines
20 KiB
Python
467 lines
20 KiB
Python
# ============================================================
|
||
# 分析视图（moduleId: domain-analytics, 可重生 ✅）
|
||
# EX-07 方案对比表（KPI 逐项 diff）/ EX-08 KPI 仪表盘
|
||
# ============================================================
|
||
from __future__ import annotations
|
||
|
||
from typing import Any
|
||
|
||
from server.aps_domain.views import world_summary
|
||
from server.timeutil import parse_dt
|
||
|
||
World = dict[str, Any]
|
||
|
||
|
||
def build_utilization_report(world: World, track: str = "flex", days: int = 7) -> dict:
|
||
"""EX-06：资源利用率增强报告（柔性设备 + 班组）。
|
||
|
||
设备口径与「设备负荷」视口一致：只统计落在班次工作窗内的时间，
|
||
统计窗口跟着排产结果走；否则计划排在 10 月时整屏都会是 0%/休。
|
||
"""
|
||
from server.aps_domain.views import _latest_flex_version, _shift_windows_for
|
||
from server.state.seed import ensure_flex_seed
|
||
from server.timeutil import add_minutes, fmt_date, parse_dt, today0
|
||
|
||
tr = (track or "flex").lower()
|
||
if tr != "flex":
|
||
# 固定产线走「设备负荷」视口的产线口径，这里只做指路
|
||
from server.aps_domain.views import load_view
|
||
lv = load_view(world, days=days)
|
||
s = world_summary(world)
|
||
return {
|
||
"track": "fixed",
|
||
"summary": {
|
||
"avgUtilization": s.get("avgUtilization"),
|
||
"versionNo": s.get("versionNo"),
|
||
"overloaded": sum(1 for r in lv.get("rows", [])
|
||
for c in r.get("cells", []) if (c.get("ratio") or 0) > 1),
|
||
},
|
||
"equipment": [], "teams": [], "days": lv.get("days", []),
|
||
"hint": "固定产线请看「设备负荷」视图；这里只统计柔性试排的设备与班组。",
|
||
"load": lv,
|
||
}
|
||
|
||
ensure_flex_seed(world)
|
||
latest = _latest_flex_version(world)
|
||
vid = latest["id"] if latest else None
|
||
wos = [w for w in world.get("flexWorkOrders") or []
|
||
if (not vid or w.get("versionId") == vid) and not w.get("frozen")]
|
||
|
||
# 统计窗口跟着排产结果走：从最早开工那天开始，没有排产才退回今天
|
||
starts = []
|
||
for w in wos:
|
||
try:
|
||
starts.append(parse_dt(w["plannedStartTime"]))
|
||
except Exception:
|
||
continue
|
||
base = min(starts).replace(hour=0, minute=0, second=0, microsecond=0) if starts else today0()
|
||
day_list = [fmt_date(add_minutes(base, i * 24 * 60)) for i in range(max(1, days))]
|
||
|
||
day_windows = {d: _shift_windows_for(world, d) for d in day_list} # 当日有效工作窗
|
||
shift_total = sum((b - a).total_seconds() / 60
|
||
for ws in day_windows.values() for a, b in ws) # 窗口内总可用分钟
|
||
|
||
# ---- 设备行 ----
|
||
eq_rows = []
|
||
for eq in world.get("flexEquipment", []) or []:
|
||
eid = eq["id"]
|
||
rate = float(eq.get("availabilityRate") if eq.get("availabilityRate") is not None else 1.0)
|
||
used_total = 0.0
|
||
cells = []
|
||
for d in day_list:
|
||
windows = day_windows[d]
|
||
used = 0.0
|
||
for w in wos:
|
||
if w.get("equipmentId") != eid:
|
||
continue
|
||
try:
|
||
ws, we = parse_dt(w["plannedStartTime"]), parse_dt(w["plannedEndTime"])
|
||
except Exception:
|
||
continue
|
||
for a, b in windows: # 只累计与工作窗的重叠
|
||
lo, hi = max(ws, a), min(we, b)
|
||
if hi > lo:
|
||
used += (hi - lo).total_seconds() / 60
|
||
avail = round(sum((b - a).total_seconds() / 60 for a, b in windows) * rate) if windows else 0
|
||
ratio = (used / avail) if avail else 0.0
|
||
cells.append({
|
||
"date": d, "used": round(used, 1), "avail": round(avail, 1),
|
||
"ratio": round(ratio, 3), "offShift": not windows,
|
||
})
|
||
used_total += used
|
||
avail_total = sum(c["avail"] for c in cells)
|
||
util = used_total / avail_total if avail_total else 0.0
|
||
eq_rows.append({
|
||
"code": eq.get("code"), "name": eq.get("name"), "zone": eq.get("zone"),
|
||
"status": eq.get("status"), "movable": bool(eq.get("movable")),
|
||
"usedMinutes": round(used_total, 1),
|
||
"availMinutes": round(avail_total, 1),
|
||
"utilization": round(util, 3),
|
||
"peakRatio": max((c["ratio"] for c in cells), default=0),
|
||
"cells": cells,
|
||
"alert": "crit" if util >= 0.95 or any(c["ratio"] > 1 for c in cells)
|
||
else "warn" if util >= 0.85 else "ok",
|
||
})
|
||
eq_rows.sort(key=lambda r: -r["utilization"])
|
||
|
||
# ---- 班组行（SC-11）：按工序并发峰值 / 人数 ----
|
||
team_rows = []
|
||
for team in world.get("flexTeams", []) or []:
|
||
ops = list(team.get("supportOps") or [])
|
||
members = int(team.get("memberCount") or 0)
|
||
# 扫描工单：同班组覆盖工序的重叠最大并发
|
||
intervals = []
|
||
for w in wos:
|
||
if w.get("operationCode") not in ops:
|
||
continue
|
||
if team.get("code") and w.get("teamCode") and w.get("teamCode") != team.get("code"):
|
||
# 若工单已钉具体班组，只计本班组
|
||
continue
|
||
try:
|
||
intervals.append((parse_dt(w["plannedStartTime"]), parse_dt(w["plannedEndTime"])))
|
||
except Exception:
|
||
pass
|
||
peak = 0
|
||
for i, (s, e) in enumerate(intervals):
|
||
concurrent = sum(1 for s2, e2 in intervals if s < e2 and e > s2)
|
||
peak = max(peak, concurrent)
|
||
# 粗利用率：本班组工单总分钟 / (人数 × 班次 × 工作日)
|
||
used = 0.0
|
||
for s, e in intervals:
|
||
used += (e - s).total_seconds() / 60
|
||
avail = members * shift_total # 班组可用人时 = 人数 × 窗口内班次分钟
|
||
util = used / avail if avail else 0.0
|
||
team_rows.append({
|
||
"code": team.get("code"), "name": team.get("name"),
|
||
"memberCount": members, "supportOps": ops,
|
||
"skillLevel": team.get("skillLevel"),
|
||
"peakConcurrent": peak,
|
||
"usedMinutes": round(used, 1),
|
||
"availMinutes": round(avail, 1),
|
||
"utilization": round(util, 3),
|
||
"saturated": members > 0 and peak >= members,
|
||
"alert": "crit" if (members > 0 and peak > members) or util >= 0.95
|
||
else "warn" if peak >= members or util >= 0.85 else "ok",
|
||
})
|
||
team_rows.sort(key=lambda r: -r["utilization"])
|
||
|
||
utils = [r["utilization"] for r in eq_rows if r["status"] == "RUNNING"]
|
||
avg = round(sum(utils) / len(utils), 3) if utils else 0.0
|
||
report = {
|
||
"track": "flex",
|
||
"versionNo": latest.get("versionNo") if latest else None,
|
||
"versionStatus": latest.get("status") if latest else None,
|
||
"trialOnly": bool(latest.get("trialOnly")) if latest else None,
|
||
"shiftMinutes": round(shift_total),
|
||
"windowStart": day_list[0] if day_list else None,
|
||
"windowDays": len(day_list),
|
||
"days": day_list,
|
||
"summary": {
|
||
"avgUtilization": avg,
|
||
"equipmentCount": len(eq_rows),
|
||
"runningCount": sum(1 for r in eq_rows if r["status"] == "RUNNING"),
|
||
"overloaded": sum(1 for r in eq_rows if r["alert"] == "crit"),
|
||
"warnCount": sum(1 for r in eq_rows if r["alert"] == "warn"),
|
||
"idleCount": sum(1 for r in eq_rows if r["utilization"] < 0.15 and r["status"] == "RUNNING"),
|
||
"teamSaturated": sum(1 for r in team_rows if r.get("saturated")),
|
||
},
|
||
"equipment": eq_rows,
|
||
"teams": team_rows,
|
||
"hint": "占用时间只统计班次内的时间：绿色=空闲，黄色=偏忙，橙色=接近排满，"
|
||
"红色=排满还超了；班组「人手到顶」表示同一时段需要的操作人数已达到班组人数。",
|
||
}
|
||
if shift_total <= 0:
|
||
report["emptyReason"] = "这个工厂还没有登记班次，算不出可用工时：请先在制造主数据里补班次。"
|
||
elif not wos:
|
||
report["emptyReason"] = "这一版柔性试排没有排任何工序，设备占用为空。"
|
||
return report
|
||
|
||
|
||
def build_kpi_dashboard(world: World) -> dict:
|
||
"""EX-08：固定轨 + 柔性轨 + 执行进度聚合仪表盘（只读）。"""
|
||
from server.state.seed import ensure_flex_seed
|
||
ensure_flex_seed(world)
|
||
|
||
fixed = world_summary(world)
|
||
# 准时率（固定）：非超期 PO / 全部 PO
|
||
on_time_rate = None
|
||
wip = 0
|
||
overdue_po = 0
|
||
if fixed.get("hasVersion"):
|
||
v = world["scheduleVersions"][-1]
|
||
pos = [p for p in world["productionOrders"] if p.get("schedulingVersionId") == v["id"]]
|
||
so_by = {s["id"]: s for s in world["salesOrders"]}
|
||
for po in pos:
|
||
so = so_by.get(po.get("salesOrderId"))
|
||
if so and po.get("plannedEndDate"):
|
||
late = parse_dt(po["plannedEndDate"]) > parse_dt(so["deliveryDate"] + " 18:00")
|
||
if late:
|
||
overdue_po += 1
|
||
if po.get("status") in ("CONFIRMED", "IN_PROGRESS", "RELEASED"):
|
||
wip += 1
|
||
if pos:
|
||
on_time_rate = round((len(pos) - overdue_po) / len(pos), 3)
|
||
|
||
flex_ver = (world.get("flexScheduleVersions") or [None])[-1]
|
||
flex = None
|
||
if flex_ver:
|
||
oc = flex_ver.get("orderCount") or flex_ver.get("vlCount") or 0
|
||
ot = flex_ver.get("onTimeCount") or 0
|
||
flex = {
|
||
"versionNo": flex_ver.get("versionNo"),
|
||
"status": flex_ver.get("status"),
|
||
"sortMode": flex_ver.get("sortMode"),
|
||
"woCount": flex_ver.get("woCount", 0),
|
||
"vlCount": flex_ver.get("vlCount", 0),
|
||
"conflictCount": flex_ver.get("conflictCount", 0),
|
||
"totalTardiness": round(flex_ver.get("totalTardiness") or 0, 1),
|
||
"avgUtilization": round(flex_ver.get("avgUtilization") or 0, 3),
|
||
"onTimeCount": ot,
|
||
"orderCount": oc,
|
||
"onTimeRate": round(ot / oc, 3) if oc else None,
|
||
"makespan": flex_ver.get("makespan"),
|
||
}
|
||
|
||
# 冲突未解决
|
||
flex_cf = sum(1 for c in world.get("flexConflicts", []) if not c.get("isResolved"))
|
||
fixed_cf = sum(1 for c in world.get("conflicts", [])
|
||
if not c.get("isResolved") and (
|
||
not fixed.get("hasVersion")
|
||
or c.get("versionId") == world["scheduleVersions"][-1]["id"]))
|
||
|
||
# MES 执行
|
||
mes_rows = [w for w in world.get("flexWorkOrders", []) if w.get("mesExternalId")]
|
||
if flex_ver:
|
||
mes_rows = [w for w in mes_rows if w.get("versionId") == flex_ver["id"]]
|
||
mes_done = sum(1 for w in mes_rows
|
||
if w.get("status") == "COMPLETED" or (w.get("progressPct") or 0) >= 100)
|
||
mes_pct = round(mes_done / len(mes_rows), 3) if mes_rows else None
|
||
|
||
# 柔性在制
|
||
flex_wip = sum(1 for w in world.get("flexWorkOrders", [])
|
||
if flex_ver and w.get("versionId") == flex_ver["id"]
|
||
and w.get("status") in ("RUNNING", "RELEASED", "PENDING"))
|
||
|
||
cards = [
|
||
{"id": "util", "label": "平均利用率",
|
||
"value": (f"{round((fixed.get('avgUtilization') or 0) * 100)}%"
|
||
if fixed.get("hasVersion") else
|
||
(f"{round((flex or {}).get('avgUtilization', 0) * 100)}%" if flex else "—")),
|
||
"hint": "设备实际占用时间 ÷ 可用时间"},
|
||
{"id": "ontime", "label": "准时率",
|
||
"value": (f"{round(on_time_rate * 100)}%" if on_time_rate is not None
|
||
else (f"{round(flex['onTimeRate'] * 100)}%" if flex and flex.get("onTimeRate") is not None
|
||
else "—")),
|
||
"hint": "能在客户交期前完工的订单占比"},
|
||
{"id": "delay", "label": "总延误",
|
||
"value": (f"{fixed.get('totalTardiness')}h" if fixed.get("hasVersion")
|
||
else (f"{flex['totalTardiness']}h" if flex else "—")),
|
||
"hint": "所有订单超出交期的小时数合计"},
|
||
{"id": "wip", "label": "在制工序",
|
||
"value": str(wip + flex_wip),
|
||
"hint": f"固定产线 {wip} 条 + 柔性排产 {flex_wip} 条"},
|
||
{"id": "conflict", "label": "待处理冲突",
|
||
"value": str(fixed_cf + flex_cf),
|
||
"hint": f"固定产线 {fixed_cf} 条 / 柔性排产 {flex_cf} 条"},
|
||
{"id": "mes", "label": "车间完工率",
|
||
"value": (f"{round(mes_pct * 100)}%" if mes_pct is not None else "—"),
|
||
"hint": (f"已下发车间 {len(mes_rows)} 条 · 已完工 {mes_done} 条"
|
||
if mes_rows else "还没有工序下发到车间，暂时算不出完工率")},
|
||
]
|
||
|
||
# 数据来源说明：这张表算的是哪一版、有没有下发车间，必须写在页面上。
|
||
if fixed.get("hasVersion"):
|
||
source = {"track": "FIXED", "versionNo": fixed.get("versionNo"),
|
||
"status": fixed.get("status"), "trialOnly": False}
|
||
elif flex:
|
||
source = {"track": "FLEX", "versionNo": flex.get("versionNo"),
|
||
"status": flex.get("status"), "trialOnly": True}
|
||
else:
|
||
source = None
|
||
|
||
return {
|
||
"cards": cards,
|
||
"fixed": fixed if fixed.get("hasVersion") else None,
|
||
"flex": flex,
|
||
"source": source,
|
||
"execution": {
|
||
"dispatched": len(mes_rows), "completed": mes_done,
|
||
"progressRate": mes_pct,
|
||
},
|
||
"conflicts": {"fixed": fixed_cf, "flex": flex_cf, "total": fixed_cf + flex_cf},
|
||
"generatedAt": None,
|
||
}
|
||
|
||
|
||
def _matrix_from_cards(
|
||
columns: list[dict],
|
||
metrics: list[tuple[str, str, bool]],
|
||
) -> dict:
|
||
"""metrics: (key, label, lower_is_better)。从 columns[].kpi 抽逐项表。"""
|
||
rows = []
|
||
for key, label, lower_better in metrics:
|
||
values = []
|
||
for col in columns:
|
||
kpi = col.get("kpi") or col
|
||
val = kpi.get(key)
|
||
if key == "avgUtilization" and val is not None and val <= 1:
|
||
val = round(val * 100, 1)
|
||
values.append(val)
|
||
numeric = [(i, v) for i, v in enumerate(values) if isinstance(v, (int, float))]
|
||
best_idx = None
|
||
if numeric:
|
||
best_idx = (min if lower_better else max)(numeric, key=lambda x: x[1])[0]
|
||
best_val = values[best_idx] if best_idx is not None else None
|
||
deltas = []
|
||
for v in values:
|
||
if best_val is None or not isinstance(v, (int, float)):
|
||
deltas.append(None)
|
||
else:
|
||
deltas.append(round(v - best_val, 2) if isinstance(best_val, float)
|
||
else v - best_val)
|
||
rows.append({
|
||
"metric": key, "label": label, "lowerIsBetter": lower_better,
|
||
"values": values, "bestIdx": best_idx, "deltas": deltas,
|
||
})
|
||
return {"columns": columns, "rows": rows}
|
||
|
||
|
||
def matrix_from_fixed_cards(cards: list[dict], *, baseline: str | None = None) -> dict:
|
||
cols = [{
|
||
"id": c["strategy"], "label": c["label"], "strategy": c["strategy"],
|
||
"kpi": c["kpi"], "diffVsBaseline": c.get("diffVsBaseline"),
|
||
"risks": c.get("risks") or [],
|
||
} for c in cards]
|
||
metrics = [
|
||
("totalTardiness", "总延迟 (h)", True),
|
||
("conflictCount", "冲突数", True),
|
||
("avgUtilization", "利用率 (%)", False),
|
||
("totalCost", "预估成本", True),
|
||
("woCount", "工单数", True),
|
||
("poCount", "生产订单", True),
|
||
]
|
||
section = _matrix_from_cards(cols, metrics)
|
||
best = min(cards, key=lambda c: (c["kpi"]["totalTardiness"], c["kpi"]["conflictCount"]))
|
||
section.update({
|
||
"id": "fixed", "title": "固定产线 · 排产策略对比",
|
||
"baseline": baseline,
|
||
"recommendation": best["label"],
|
||
"recommendationId": best["strategy"],
|
||
})
|
||
return section
|
||
|
||
|
||
def matrix_from_flex_rows(rows: list[dict], *, hint: str | None = None,
|
||
data_quality: dict | None = None,
|
||
due_basis: str | None = None) -> dict:
|
||
cols = [{
|
||
"id": r["sortMode"], "label": r["label"], "sortMode": r["sortMode"],
|
||
"strategyStatus": str(r.get("strategyStatus") or "READY").upper(),
|
||
"strategyEvidence": r.get("strategyEvidence") or [],
|
||
"resultStatus": r.get("resultStatus"),
|
||
"kpi": {
|
||
"totalTardiness": r["totalTardiness"],
|
||
"conflictCount": r["conflictCount"],
|
||
"avgUtilization": r["avgUtilization"],
|
||
"onTimeCount": r["onTimeCount"],
|
||
"orderCount": r["orderCount"],
|
||
"vlCount": r["vlCount"],
|
||
},
|
||
} for r in rows]
|
||
metrics = [
|
||
("totalTardiness", "总延误（小时）", True),
|
||
("onTimeCount", "能按时交付的订单", False),
|
||
("orderCount", "订单总数", True),
|
||
("conflictCount", "需要处理的冲突", True),
|
||
("vlCount", "排产产线数", True),
|
||
("avgUtilization", "设备利用率 (%)", False),
|
||
]
|
||
section = _matrix_from_cards(cols, metrics)
|
||
quality = data_quality or {}
|
||
if due_basis:
|
||
section["dueBasis"] = due_basis
|
||
production_ready = bool(quality.get("productionReady", True))
|
||
ready_rows = [
|
||
row for row in rows
|
||
if str(row.get("strategyStatus") or "READY").upper() == "READY"
|
||
and str(row.get("resultStatus") or "READY").upper() in {"READY", "WITH_CONFLICTS"}
|
||
]
|
||
best = min(
|
||
ready_rows,
|
||
key=lambda r: (
|
||
r["totalTardiness"], r["conflictCount"], -int(r.get("onTimeCount") or 0),
|
||
),
|
||
) if ready_rows else None
|
||
if best is None:
|
||
recommendation_status = "UNAVAILABLE"
|
||
recommendation_reason = "没有 strategyStatus=READY 的策略，请先补齐策略输入数据后重新试排。"
|
||
elif production_ready:
|
||
recommendation_status = "READY"
|
||
recommendation_reason = "策略输入与生产数据质量检查通过。"
|
||
else:
|
||
recommendation_status = "TRIAL_ONLY"
|
||
warning_messages = [
|
||
str(warning.get("message"))
|
||
for warning in (quality.get("warnings") or [])
|
||
if warning.get("message")
|
||
]
|
||
recommendation_reason = "数据待补：" + (
|
||
"；".join(warning_messages) if warning_messages else "生产数据质量尚未就绪"
|
||
)
|
||
section.update({
|
||
"id": "flex", "title": "柔性排产 · 五种排序方式对比",
|
||
"hint": hint,
|
||
"dataQuality": quality,
|
||
"productionReady": production_ready,
|
||
"recommendationStatus": recommendation_status,
|
||
"recommendationReason": recommendation_reason,
|
||
"recommendation": best["label"] if best else None,
|
||
"recommendationId": best["sortMode"] if best else None,
|
||
"eligibleRecommendationIds": [row["sortMode"] for row in ready_rows],
|
||
})
|
||
return section
|
||
|
||
|
||
def build_compare_table(world: World, track: str = "both") -> dict:
|
||
"""EX-07：方案对比逐项 diff 表。track ∈ fixed|flex|both。"""
|
||
from server.aps_domain.scenario import compare_scenarios
|
||
from server.aps_domain.flex import compare_sort_modes
|
||
from server.state.seed import ensure_flex_seed
|
||
|
||
tr = (track or "both").lower()
|
||
out: dict[str, Any] = {"track": tr, "sections": []}
|
||
|
||
if tr in ("fixed", "both"):
|
||
fixed_orders = len(world.get("productionOrders") or []) + len(world.get("workOrders") or [])
|
||
if fixed_orders == 0:
|
||
# 固定产线一张单都没排过，硬算只会得到一排 0，看起来像"策略都一样"。
|
||
out["sections"].append({
|
||
"id": "fixed", "title": "固定产线 · 排产策略对比",
|
||
"unavailable": True,
|
||
"unavailableReason": (
|
||
"这个项目还没有固定产线的生产订单和工单，无法做策略对比。"
|
||
"下方柔性排产对比使用的是已导入的柔性订单数据。"
|
||
),
|
||
"columns": [], "rows": [],
|
||
})
|
||
else:
|
||
_, block = compare_scenarios(world)
|
||
cards = block.props.get("cards") or []
|
||
out["sections"].append(matrix_from_fixed_cards(
|
||
cards, baseline=block.props.get("baseline")))
|
||
|
||
if tr in ("flex", "both"):
|
||
ensure_flex_seed(world)
|
||
|
||
class _Tmp:
|
||
def __init__(self, data):
|
||
self.data = data
|
||
|
||
# 按订单真实交期试排（压缩交期只用于显式压测，不能当默认值展示）
|
||
flex = compare_sort_modes(_Tmp(world), compress_due=False)
|
||
out["sections"].append(matrix_from_flex_rows(
|
||
flex["rows"], hint=flex.get("hint"), data_quality=flex.get("dataQuality"),
|
||
due_basis=flex.get("dueBasis")))
|
||
|
||
return out
|