276 lines
17 KiB
Python
276 lines
17 KiB
Python
# ============================================================
|
||
# 世界数据视图（moduleId: domain-views, 可重生 ✅）
|
||
# 把世界状态投影为前端三个视口所需的数据（P0 只读）：
|
||
# summary(KPI) / gantt(甘特) / load(负荷热力) / due(交期承诺看板)
|
||
# 过滤/高亮所需的标记（客户、超期、冲突）由后端一次性算好附在数据上
|
||
# ============================================================
|
||
from __future__ import annotations # 前向类型引用
|
||
|
||
from typing import Any # 类型标注
|
||
|
||
from server.engines.queries import get_available_minutes # 班次可用分钟（热力图）
|
||
from server.timeutil import add_minutes, fmt_date, parse_dt, today0 # 日期工具
|
||
|
||
# 世界状态类型别名
|
||
World = dict[str, Any]
|
||
|
||
|
||
def _latest_version(world: World) -> dict | None:
|
||
"""取最新排产版本（无版本返回 None）。"""
|
||
versions = world["scheduleVersions"] # 版本表
|
||
return versions[-1] if versions else None # 末尾即最新
|
||
|
||
|
||
def _latest_flex_version(world: World) -> dict | None:
|
||
"""取最新柔性排产版本（无版本返回 None）。"""
|
||
versions = world.get("flexScheduleVersions") or [] # 柔性版本表
|
||
return versions[-1] if versions else None # 末尾即最新
|
||
|
||
|
||
def _version_pos(world: World, version_id: int) -> list[dict]:
|
||
"""取某版本的全部生产订单。"""
|
||
return [p for p in world["productionOrders"] if p["schedulingVersionId"] == version_id] # 按版本过滤
|
||
|
||
|
||
def _flex_promised_by_order(world: World, version_id: int) -> dict[str, str]:
|
||
"""柔性版本下各订单的承诺完成＝该订单虚拟产线的最晚完成时间（键=订单号）。"""
|
||
promised: dict[str, str] = {} # 订单号 → 最晚完成时间
|
||
for vl in world.get("flexVirtualLines") or []: # 遍历虚拟产线
|
||
if vl.get("versionId") != version_id or not vl.get("plannedEnd"):
|
||
continue # 非本版本 / 无完成时间不参与
|
||
order_no = str(vl.get("orderNo") or "") # 订单号
|
||
if not order_no: # 无订单号跳过
|
||
continue
|
||
current = promised.get(order_no) # 已记录的完成时间
|
||
if current is None or vl["plannedEnd"] > current: # "YYYY-MM-DD HH:MM" 串序即时间序
|
||
promised[order_no] = vl["plannedEnd"]
|
||
return promised
|
||
|
||
|
||
def _is_overdue_po(world: World, po: dict) -> bool:
|
||
"""判断生产订单是否超期：计划完成晚于销售订单交期当日 18:00。"""
|
||
so = next((s for s in world["salesOrders"] if s["id"] == po["salesOrderId"]), None) # 关联订单
|
||
if not so or not po.get("plannedEndDate"): # 数据缺失视为不超期
|
||
return False
|
||
return parse_dt(po["plannedEndDate"]) > parse_dt(so["deliveryDate"] + " 18:00") # 比较交期
|
||
|
||
|
||
def world_summary(world: World) -> dict[str, Any]:
|
||
"""KPI 摘要：最新版本核心指标（视口工具栏 KPI 卡组数据源）。"""
|
||
v = _latest_version(world) # 最新版本
|
||
if not v: # 尚无版本
|
||
return {"hasVersion": False}
|
||
return { # 有版本 → 摘要字段
|
||
"hasVersion": True, "versionNo": v["versionNo"], "status": v["status"],
|
||
"woCount": v["woCount"], "poCount": v["poCount"], "conflictCount": v["conflictCount"],
|
||
"totalTardiness": round(v["totalTardiness"], 1),
|
||
"avgUtilization": round(v["avgUtilization"], 3),
|
||
"totalCost": round(v["totalCost"], 0),
|
||
}
|
||
|
||
|
||
def gantt_view(world: World) -> dict[str, Any]:
|
||
"""甘特视图数据：产线/工位骨架 + 最新版本工单（附过滤与高亮标记）。"""
|
||
v = _latest_version(world) # 最新版本
|
||
pos = _version_pos(world, v["id"]) if v else [] # 该版本 PO
|
||
po_by_id = {p["id"]: p for p in pos} # PO 索引
|
||
so_by_id = {s["id"]: s for s in world["salesOrders"]} # SO 索引
|
||
wos_out: list[dict] = [] # 输出工单集
|
||
for wo in world["workOrders"]: # 遍历全部工单
|
||
po = po_by_id.get(wo["productionOrderId"]) # 关联 PO（仅最新版本）
|
||
if not po: # 非本版本工单不输出
|
||
continue
|
||
so = so_by_id.get(po["salesOrderId"], {}) # 关联 SO（客户标记）
|
||
wos_out.append({ # 投影为前端 GanttWO 契约
|
||
"id": wo["id"], "orderNo": wo["orderNo"], "productId": wo["productId"],
|
||
"productName": wo["productName"], "quantity": wo["quantity"],
|
||
"operationName": wo["operationName"], "lineId": wo["lineId"],
|
||
"workstationId": wo["workstationId"], "workstationName": wo["workstationName"],
|
||
"start": wo["plannedStartTime"], "end": wo["plannedEndTime"],
|
||
"customerName": so.get("customerName", ""), "customerLevel": so.get("customerLevel", ""),
|
||
"overdue": _is_overdue_po(world, po), # 超期标记（高亮用）
|
||
# 冲突标记：齐套非通过 或 工单/PO 冲突计数 > 0
|
||
"conflict": wo["kitStatus"] != "PASSED" or wo["conflictCount"] > 0 or po["conflictCount"] > 0,
|
||
})
|
||
return { # 视图整体响应
|
||
"lines": [{"id": l["id"], "code": l["code"], "name": l["name"]} for l in world["lines"]], # 产线骨架
|
||
"workstations": [{"id": w["id"], "lineId": w["lineId"], "name": w["name"], "seq": w["sequenceNo"]}
|
||
for w in world["workstations"]], # 工位骨架
|
||
"workOrders": wos_out, # 工单条
|
||
}
|
||
|
||
|
||
def load_view(world: World, days: int = 14) -> dict[str, Any]:
|
||
"""负荷热力视图：行 × 未来 N 天的负荷率（占用/可用分钟）。
|
||
|
||
固定产线有版本时按产线展示；这个工厂只有柔性试排结果时，按设备展示，
|
||
否则计划员明明已经排完产，负荷页却整屏都是"休"。
|
||
"""
|
||
v = _latest_version(world) # 最新固定产线版本
|
||
if v:
|
||
pos = _version_pos(world, v["id"]) # 该版本 PO
|
||
po_ids = {p["id"] for p in pos} # PO ID 集
|
||
day_list = [fmt_date(add_minutes(today0(), i * 24 * 60)) for i in range(days)] # 日期列
|
||
rows: list[dict] = [] # 输出行集
|
||
for line in world["lines"]: # 逐产线
|
||
cells: list[dict] = [] # 该产线的逐日单元格
|
||
for d in day_list: # 逐日
|
||
used = sum( # 当日占用分钟：本版本、本线、当日开工的工单
|
||
(parse_dt(w["plannedEndTime"]) - parse_dt(w["plannedStartTime"])).total_seconds() / 60
|
||
for w in world["workOrders"]
|
||
if w["productionOrderId"] in po_ids and w["lineId"] == line["id"]
|
||
and w["plannedStartTime"][:10] == d)
|
||
avail = get_available_minutes(world, line["id"], d) # 当日可用分钟
|
||
ratio = (used / avail) if avail else 0.0 # 负荷率（无班次记 0，前端按 avail=0 显示"休"）
|
||
cells.append({"ratio": round(ratio, 3), "used": round(used), "avail": avail}) # 单元格
|
||
rows.append({"lineId": line["id"], "lineCode": line["code"], # 产线行（带编码供聚焦匹配）
|
||
"lineName": line["name"], "cells": cells})
|
||
return {"days": day_list, "rows": rows, "track": "FIXED", "rowLabel": "产线"} # 视图响应
|
||
|
||
flex = _latest_flex_version(world) # 固定通道无版本 → 看柔性试排
|
||
if not flex:
|
||
day_list = [fmt_date(add_minutes(today0(), i * 24 * 60)) for i in range(days)]
|
||
return {"days": day_list, "rows": [], "track": None, "rowLabel": "产线",
|
||
"emptyReason": "这个项目还没有排产结果：固定产线没有排产版本，柔性排产也还没有试排记录。"}
|
||
# 未来排产的窗口跟着计划走：计划 10 月开工时，从今天起算两周只会看到满屏空白。
|
||
version_wos = [w for w in world.get("flexWorkOrders") or []
|
||
if w.get("versionId") == flex["id"] and w.get("plannedStartTime")]
|
||
starts = [parse_dt(w["plannedStartTime"]) for w in version_wos]
|
||
base = max(today0(), 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(days)]
|
||
return _flex_load_view(world, flex, day_list) # 柔性设备负荷
|
||
|
||
|
||
def _shift_windows_for(world: World, day: str) -> list[tuple[Any, Any]]:
|
||
"""柔性班次当天的有效工作窗（已扣休息；未登记班次返回空列表）。"""
|
||
shifts = world.get("flexCalendar") or [] # 柔性班次表
|
||
if not shifts: # 没有班次数据不猜
|
||
return []
|
||
weekday = parse_dt(day).isoweekday() # 1=周一 … 7=周日
|
||
windows: list[tuple[Any, Any]] = [] # 当天全部工作窗
|
||
for shift in shifts: # 逐条班次
|
||
if shift.get("enabled") is False: # 备用/默认关闭的班次不占产能
|
||
continue
|
||
if "关闭" in str(shift.get("sourceStatus") or ""): # 资料里标注停用的班次
|
||
continue
|
||
workdays = shift.get("workdays") # 适用星期
|
||
if workdays and weekday not in workdays: # 当天不上班
|
||
continue
|
||
start, end = shift.get("startTime"), shift.get("endTime") # 班次起止
|
||
if not start or not end:
|
||
continue
|
||
start_min, end_min = _clock_minutes(start), _clock_minutes(end) # 起止分钟
|
||
breaks = sorted( # 休息按时间排序
|
||
(_clock_minutes(b.get("start")), _clock_minutes(b.get("end")))
|
||
for b in (shift.get("breaks") or [])
|
||
)
|
||
cursor = start_min # 从班次起点开始切
|
||
for b_start, b_end in breaks: # 用休息切出工作窗
|
||
if b_start > cursor:
|
||
windows.append((parse_dt(f"{day} 00:00") + _minutes(cursor),
|
||
parse_dt(f"{day} 00:00") + _minutes(min(b_start, end_min))))
|
||
cursor = max(cursor, b_end) # 跳到休息后
|
||
if end_min > cursor: # 收尾工作窗
|
||
windows.append((parse_dt(f"{day} 00:00") + _minutes(cursor),
|
||
parse_dt(f"{day} 00:00") + _minutes(end_min)))
|
||
return windows
|
||
|
||
|
||
def _minutes(value: int):
|
||
"""分钟数转 timedelta（避免上层重复 import）。"""
|
||
from datetime import timedelta
|
||
return timedelta(minutes=max(0, value))
|
||
|
||
|
||
def _clock_minutes(value: str | None) -> int:
|
||
"""把 "HH:MM" 转成当日分钟数（缺失返回 0）。"""
|
||
text = str(value or "") # 容错
|
||
if ":" not in text:
|
||
return 0
|
||
hh, _, mm = text.partition(":") # 拆时/分
|
||
try:
|
||
return int(hh) * 60 + int(mm)
|
||
except ValueError:
|
||
return 0
|
||
|
||
|
||
def _flex_load_view(world: World, flex: dict, day_list: list[str]) -> dict[str, Any]:
|
||
"""柔性设备负荷：行=设备，单元格=当日已排工序占用分钟 ÷ 当日可动分钟。
|
||
|
||
只统计落在班次工作窗内的时间，跨夜工单不会把夜里算成设备占用。
|
||
"""
|
||
version_id = flex["id"] # 目标版本
|
||
wos = [w for w in world.get("flexWorkOrders") or [] if w.get("versionId") == version_id]
|
||
if not wos: # 版本内没有工单 → 如实空态
|
||
return {"days": day_list, "rows": [], "track": "FLEX", "rowLabel": "设备",
|
||
"versionNo": flex.get("versionNo"),
|
||
"emptyReason": f"柔性试排版本 {flex.get('versionNo')} 里还没有排到设备上的工序。"}
|
||
used_ids = {w.get("equipmentId") for w in wos} # 本次排产用到的设备
|
||
equipment = [e for e in world.get("flexEquipment") or [] if e.get("id") in used_ids]
|
||
day_bounds = { # 预解析每张工单的起止
|
||
id(w): (parse_dt(w["plannedStartTime"]), parse_dt(w["plannedEndTime"]))
|
||
for w in wos if w.get("plannedStartTime") and w.get("plannedEndTime")
|
||
}
|
||
rows: list[dict] = [] # 输出行
|
||
for eq in sorted(equipment, key=lambda e: (str(e.get("zone") or ""), str(e.get("code") or ""))):
|
||
rate = float(eq.get("availabilityRate") if eq.get("availabilityRate") is not None else 1.0)
|
||
cells: list[dict] = [] # 该设备逐日单元格
|
||
for d in day_list: # 逐日
|
||
windows = _shift_windows_for(world, d) # 当日工作窗
|
||
avail = round(sum((b - a).total_seconds() / 60 for a, b in windows) * rate) if windows else 0
|
||
used = 0.0 # 当日占用
|
||
for w in wos: # 逐工单算与工作窗的重叠
|
||
if w.get("equipmentId") != eq.get("id"):
|
||
continue
|
||
bounds = day_bounds.get(id(w))
|
||
if not bounds:
|
||
continue
|
||
ws, we = bounds # 工单起止
|
||
for a, b in windows: # 逐工作窗求交集
|
||
lo, hi = max(ws, a), min(we, b)
|
||
if hi > lo:
|
||
used += (hi - lo).total_seconds() / 60
|
||
ratio = (used / avail) if avail else 0.0 # 负荷率
|
||
cells.append({"ratio": round(ratio, 3), "used": round(used), "avail": avail})
|
||
rows.append({
|
||
"lineId": eq.get("id"), "lineCode": eq.get("code"),
|
||
"lineName": eq.get("name") or eq.get("code"),
|
||
"zone": eq.get("zone"), "cells": cells,
|
||
})
|
||
return {"days": day_list, "rows": rows, "track": "FLEX", "rowLabel": "设备",
|
||
"versionNo": flex.get("versionNo"), "trialOnly": bool(flex.get("trialOnly"))} # 视图响应
|
||
|
||
|
||
def due_view(world: World) -> list[dict[str, Any]]:
|
||
"""交期承诺看板：逐销售订单的承诺完成、裕量与风险（PPT 交期承诺看板的 M1 形态）。
|
||
|
||
承诺完成优先取固定产线最新版本的工单；固定通道还没有版本时回退柔性通道最新版本的
|
||
虚拟产线。少了这层回退，已经用柔性排出来的计划在交期看板上会整列显示"未排产"。
|
||
"""
|
||
v = _latest_version(world) # 最新固定产线版本
|
||
flex = _latest_flex_version(world) if not v else None # 仅在没有固定版本时回退柔性
|
||
pos = _version_pos(world, v["id"]) if v else [] # 该版本 PO
|
||
flex_promised = _flex_promised_by_order(world, flex["id"]) if flex else {} # 柔性承诺完成
|
||
rows: list[dict] = [] # 输出行
|
||
for so in world["salesOrders"]: # 遍历订单
|
||
if so["status"] == "CANCELLED": # 已取消不展示
|
||
continue
|
||
if flex: # 柔性通道：该订单虚拟产线的最晚完成
|
||
promised = flex_promised.get(str(so["orderNo"]))
|
||
else: # 固定通道：该订单 PO 的最晚完成
|
||
my_pos = [p for p in pos if p["salesOrderId"] == so["id"]]
|
||
promised = max((p["plannedEndDate"] for p in my_pos), default=None)
|
||
margin: float | None = None # 裕量小时
|
||
risk = "none" # 默认：未排产
|
||
if promised: # 已排产 → 计算裕量与风险
|
||
due = parse_dt(so["deliveryDate"] + " 18:00") # 交期基准
|
||
margin = round((due - parse_dt(promised)).total_seconds() / 3600, 1) # 正=提前
|
||
risk = "high" if margin < 0 else ("mid" if margin < 24 else "low") # 风险分档
|
||
rows.append({ # 投影为 DueRow 契约
|
||
"orderNo": so["orderNo"], "customerName": so["customerName"],
|
||
"customerLevel": so["customerLevel"], "isRush": so["isRush"],
|
||
"deliveryDate": so["deliveryDate"], "promisedEnd": promised,
|
||
"marginHours": margin, "risk": risk,
|
||
})
|
||
return rows # 看板行集
|