周一早会，FMS控制室。
“老郑，这批A类切完换B类，又停了快四十分钟，”计划员指着排产屏，“系统里只记‘换产完成’，不拆‘拆夹具多久、装新夹具多久、对刀多久、首件校验多久’。老师傅凭手感排，一换就乱。”
我接上导出的工单序列、夹具清单、机床绑定关系、对刀记录、首件检测、历史换产日志。
“这里面有啥？”我问。
“工单号、零件族、用哪台MC、配哪套夹具、刀库刀号、对刀点、首件偏差都有，”老郑说，“可系统只给一个换产总秒数，不模拟‘卸旧夹具→清定位面→装新夹具→找正→装刀→对刀→首件试切校验’这一串动作谁等谁。想优化换产，得真换几次掐表。”
“最亏的是混流，”老郑补一句，“A用液压压板，B用专用随行夹具，C用真空吸盘。A→B要拆液压站管路，B→C要重做定位基准。单看换产总时长都差不多，可拆装等待链里卡在‘对刀等夹具找正’，没人看见。”
“我就想干一件事，”老郑说，“给多品种工单序列，仿真每轮换产的拆夹-装夹-找正-对刀-首件校验全流程，算各段耗时、等待耦合、总换产时间，对比几种换产策略，标出瓶颈动作，像个小换产仿真器，不用真停线试。”
“FMS换产不是看总秒数，”我接话，“是看‘工单序列→夹具切换图→拆装动作状态机→对刀耦合→首件校验→换产甘特→策略对比’。用 numpy 做动作时序递推，pandas 管工单/夹具表，scipy 做动作时长分布拟合，networkx 建‘工单-夹具-机床-动作’二部图，sklearn 做换产等级分类。”
“对，”老郑点头，“要能说清‘A→B换产38min，其中对刀等找正占11min；按夹具预装策略后压到22min；主因是找正与对刀串行+液压管路拆装’。”
“OOP 封好，”我开工程，“工单加载器、夹具库模型、换产动作状态机、对刀耦合器、换产扫描器、换产分类器、可视化器，合成多品种序列，下载就能跑。”
敲了行原型：
# 目标: 工单序列 → 夹具切换 → 拆/装/找正/对刀/首件 → 换产甘特 → 策略对比
# 方法: 动作状态机+耦合等待+二部图+RF换产分级
老郑凑近看：“那以后看报告：换产甘特图，动作耗时分解堆叠，工单-夹具关联图，策略对比曲线，换产等级散点。新排产单先跑，红条就是卡住的动作。”
“对，”我接话，“换产仿真不是‘记总秒数’，是‘提前看见对刀在等找正’。数字孪生里挂这个换产看板，就是老郑的‘换产尺’。”
一、实际应用场景（真实痛点）
场景设定：FMS含3台加工中心（MC1/MC2/MC3），混流加工A（液压压板）/B（专用随行夹具）/C（真空吸盘）三类零件族。每轮工单切换需完成“卸旧夹具→清定位面→装新夹具→找正→装刀→对刀→首件试切校验”。当前系统仅记录换产总时长，A→B平均38min，其中对刀等待找正占11min，排产靠经验，换产越换越乱。
现场原话（叙事化）：
“不是机床慢，”老郑说，“是换产动作串着走，还互相等。夹具找正没完，对刀不敢动；液压管路拆完才能吊新夹具。系统就给个‘换产38分’，谁也看不出卡哪。”
“最亏的是排产，”老郑说，“以前按零件族批量堆着做，现在要小批量混流，一天换三四次，每次都掐表，停线时间比切削时间还显眼。”
核心矛盾：“看换产总秒数+人工排产” 与 “工单序列→夹具切换图→拆装动作状态机→对刀耦合等待→首件校验→换产甘特→策略对比+关联图” 之间的断层。
二、痛点分析（映射到滨州职业学院《先进制造技术》课程模型）
《先进制造技术》课程模块 本篇痛点对应
柔性制造系统FMS与先进生产管理：工装切换、换产排程、资源复用 多品种换产全流程仿真
数控加工与CAD/CAM技术：对刀、装夹找正、刀长补偿 对刀耦合+找正动作建模
先进制造技术基础：辅助时间、节拍、生产效能 换产辅助时间分解
智能制造与数字孪生：FMS换产过程数字映射 换产甘特挂孪生看板
先进制造新模式：数据驱动快速换型（类SMED思想，仅描述不引术语营销） 换产策略知识库
工业机器人技术基础（类比）：末端夹具快换思路互通 夹具快换类比
一句话总结：我们需要一个“多品种工单序列→夹具切换→拆夹/装夹/找正/对刀/首件校验动作链→等待耦合→换产总时长→策略对比+关联图”程序，实现从“记换产总秒数”到“仿真定位瓶颈动作”的闭环。
三、核心逻辑讲解（大白话）
3.1 问题本质：把换产想成“换桌布+摆餐具+对刻度”
把一次换产想成饭馆换桌布上菜：
- 先撤旧桌布（拆旧夹具，含管路/螺栓）
- 擦桌子（清定位面）
- 铺新桌布（装新夹具，吊装入位）
- 对齐边线（夹具找正）
- 摆刀叉（装刀到主轴）
- 对刻度（对刀，设工件坐标系）
- 先试一道菜（首件试切+检测）
关键坑：这些动作不是全串行，但对刀必须等找正完，装刀可以跟装夹具并行一部分，首件校验必须最后。谁等谁，就是“耦合等待”。
3.2 业务逻辑 → 代码映射
输入工单序列+夹具库
│
▼ OrderLoader (pandas)
读取:
工单号, 零件族, 目标MC, 用哪套夹具, 刀组, 对刀点
│
▼ FixtureBank (numpy)
夹具库模型:
每套夹具类型(液压/随行/真空), 拆装基准时长, 是否可预装
│
▼ ChangeoverFSM (状态机)
换产动作机:
状态: UNLOAD -> CLEAN -> MOUNT -> ALIGN -> LOAD_TOOL
-> TOOL_SET -> FIRST_PIECE -> DONE
各段耗时, 标记可并行段
│
▼ ToolAlignCoupler (numpy)
对刀耦合:
对刀开始时间 = max(找正完成, 装刀完成)
等待时长 = 对刀开始 - 理论最早可开始
│
▼ ChangeoverSweeper
策略扫描:
base : 全串行
preload : 下套夹具预装到缓存位
parallel : 装刀与装夹具并行
quickclamp: 快换接口(仅建模时间缩减)
│
▼ ChangeoverClassifier (sklearn)
换产等级:
特征: 夹具类型跳变, 是否跨液压, 等待占比, MC负载
标签: 优(≤18min)/良(18~28)/超差(>28)
RF分类 + 5折宏F1
│
▼ FMSViz (matplotlib + networkx)
可视化:
1. 换产甘特图(各动作分色, 标等待)
2. 动作耗时分解堆叠柱(按策略)
3. 工单-夹具-机床二部关联图
4. 策略对比曲线(换产时长vs策略)
5. 换产等级预测vs实际散点
6. 等待耦合桑基式流向(用networkx)
│
▼ SyntheticOrders (numpy)
合成数据:
A/B/C混流, 多轮切换, 3台MC
3.3 为什么不能“看换产总秒数”
视角 问题
看总秒数 38min，不知卡哪
看经验排产 混流一多就乱
真停线掐表 代价高，不可复现
换产甘特 红条=等待，蓝条=动作
动作分解堆叠 找正/对刀各占多少
策略对比 预装能压到多少
RF分类 新序列直接判换产等级
3.4 分析前后对比
维度 传统方式 本程序
换产监控 总时长 动作级分解+等待
优化依据 老师傅手感 仿真定位瓶颈动作
策略验证 真换几次 多策略前置扫描
归因 归“换产慢” 拆成找正/对刀耦合
知识沉淀 排产经验 换产策略知识库
四、OOP 代码实现
4.1 项目结构
fms_changeover_sim/
├── fms_changeover_sim/
│ ├── __init__.py
│ ├── order_loader.py # 工单加载
│ ├── fixture_bank.py # 夹具库模型
│ ├── changeover_fsm.py # 换产动作状态机
│ ├── tool_align_coupler.py # 对刀耦合
│ ├── changeover_sweeper.py # 策略扫描
│ ├── changeover_classifier.py # 换产分级(sklearn)
│ ├── fms_viz.py # 可视化
│ └── synthetic_orders.py # 合成工单
├── tests/
│ ├── __init__.py
│ └── test_fms.py
├── results/
│ ├── changeover_gantt.png
│ ├── action_breakdown.png
│ ├── order_fixture_bipartite.png
│ ├── strategy_curve.png
│ ├── grade_pred_scatter.png
│ ├── wait_flow.png
│ ├── changeover_detail.csv
│ └── changeover_report.txt
└── run_fms.py
4.2 核心源码
<details>
<summary></summary>
"""FMS工单加载器。"""
import pandas as pd
from pathlib import Path
class OrderLoader:
"""加载多品种工单序列。"""
def __init__(self, filepath: str = "orders.csv",
encoding: str = "utf-8"):
self.filepath = Path(filepath)
self.encoding = encoding
def load(self) -> pd.DataFrame:
if not self.filepath.exists():
raise FileNotFoundError(self.filepath)
df = pd.read_csv(self.filepath, encoding=self.encoding)
req = ["oid", "family", "mc", "fixture", "tools", "align_pt"]
miss = [c for c in req if c not in df.columns]
if miss:
raise ValueError(f"缺列: {miss}")
return df.dropna(subset=req).sort_values("oid").reset_index(drop=True)
def to_seq_by_mc(self, df: pd.DataFrame) -> dict:
return {mc: list(g["family"]) for mc, g in df.groupby("mc")}
</details>
<details><summary></summary>
"""夹具库模型 (numpy)。"""
import numpy as np
from dataclasses import dataclass, field
@dataclass
class FixtureSpec:
ftype: str
unload_t: float # 拆卸基准分钟
mount_t: float # 安装基准分钟
align_t: float # 找正分钟
has_hydraulic: bool
preloadable: bool
class FixtureBank:
"""三类夹具基准参数(教学级)。"""
def __init__(self):
self.specs = {
"HYD": FixtureSpec("液压压板", 6.5, 5.0, 3.0, True, False),
"PAL": FixtureSpec("随行夹具", 4.0, 4.5, 4.5, False, True),
"VAC": FixtureSpec("真空吸盘", 2.0, 2.5, 2.0, False, True),
}
self.map_family = {"A": "HYD", "B": "PAL", "C": "VAC"}
def spec_of(self, family: str) -> FixtureSpec:
return self.specs[self.map_family[family]]
def cross_penalty(self, f1: str, f2: str) -> float:
"""跨液压管路额外拆装惩罚(分钟)"""
s1 = self.specs[self.map_family[f1]]
s2 = self.specs[self.map_family[f2]]
if s1.has_hydraulic or s2.has_hydraulic:
if f1 != f2:
return 3.5
return 0.0
</details>
<details><summary></summary>
"""换产动作状态机。"""
import numpy as np
from dataclasses import dataclass, field
from .fixture_bank import FixtureBank
@dataclass
class ActionSpan:
name: str
start: float
dur: float
is_wait: bool = False
@dataclass
class ChangeoverResult:
spans: list
total: float
wait_total: float
align_wait: float
class ChangeoverFSM:
"""
状态: UNLOAD -> CLEAN -> MOUNT -> ALIGN
(并行: LOAD_TOOL 可与 MOUNT部分重叠)
-> TOOL_SET -> FIRST_PIECE
对刀(TOOL_SET)开始 = max(ALIGN完成, LOAD_TOOL完成)
"""
def __init__(self, bank: FixtureBank,
strategy: str = "base",
rng: np.random.RandomState = None):
self.bank = bank
self.strategy = strategy
self.rng = rng or np.random.RandomState(0)
def run(self, from_fam: str, to_fam: str,
tool_change_t: float = 2.5) -> ChangeoverResult:
spec_to = self.bank.spec_of(to_fam)
spec_from = self.bank.spec_of(from_fam)
jitter = lambda x: max(0.1, x + self.rng.uniform(-0.2,0.2))
spans = []
t = 0.0
# 1 拆旧
unload = spec_from.unload_t + self.bank.cross_penalty(from_fam, to_fam)*0.6
spans.append(ActionSpan("UNLOAD", t, jitter(unload))); t += spans[-1].dur
# 2 清面
clean = 1.2
spans.append(ActionSpan("CLEAN", t, jitter(clean))); t += spans[-1].dur
# 3 装新夹具
mount_start = t
mount = spec_to.mount_t
if self.strategy == "preload" and spec_to.preloadable:
mount *= 0.45 # 预装到缓存位, 现场只推入
spans.append(ActionSpan("MOUNT", mount_start, jitter(mount)));
mount_end = mount_start + spans[-1].dur
# 4 找正(装完才找)
align_start = mount_end
align = spec_to.align_t
spans.append(ActionSpan("ALIGN", align_start, jitter(align)))
align_end = align_start + spans[-1].dur
# 5 装刀(可并行)
load_tool_start = mount_start if self.strategy=="parallel" else align_end
load_tool = tool_change_t
spans.append(ActionSpan("LOAD_TOOL", load_tool_start, jitter(load_tool)))
load_tool_end = load_tool_start + spans[-1].dur
# 6 对刀(耦合等待)
tool_set_start = max(align_end, load_tool_end)
wait = tool_set_start - align_end
if wait > 0.01:
spans.append(ActionSpan("WAIT_ALIGN", align_end, wait, is_wait=True))
tool_set = 2.0
spans.append(ActionSpan("TOOL_SET", tool_set_start, jitter(tool_set)))
tool_set_end = tool_set_start + spans[-1].dur
# 7 首件校验
first = 3.5
spans.append(ActionSpan("FIRST_PIECE", tool_set_end, jitter(first)))
end = tool_set_end + spans[-1].dur
if self.strategy == "quickclamp":
end = end * 0.82 # 快换接口整体缩减
wait_total = sum(s.dur for s in spans if s.is_wait)
return ChangeoverResult(spans, end, wait_total,
align_wait=max(0.0, wait))
</details>
<details><summary></summary>
"""对刀耦合分析器(汇总用)。"""
import numpy as np
import pandas as pd
from .changeover_fsm import ChangeoverResult
class ToolAlignCoupler:
@staticmethod
def extract(df_detail: pd.DataFrame) -> pd.DataFrame:
"""从明细里抽对刀等待占比"""
wait_rows = df_detail[df_detail.action=="WAIT_ALIGN"]
return wait_rows
</details>
<details><summary></summary>
"""换产策略扫描器。"""
import numpy as np
import pandas as pd
from dataclasses import dataclass
from .order_loader import OrderLoader
from .fixture_bank import FixtureBank
from .changeover_fsm import ChangeoverFSM
@dataclass
class COStat:
strategy: str
total_co_time: float
n_change: int
avg_wait_ratio: float
grade: str
class ChangeoverSweeper:
def __init__(self, df: pd.DataFrame,
strategies: list = None,
seed: int = 42):
self.df = df.reset_index(drop=True)
self.strategies = strategies or ["base","preload","parallel","quickclamp"]
self.seed = seed
def sweep(self) -> pd.DataFrame:
rows = []
detail_frames = []
for st in self.strategies:
bank = FixtureBank()
fsm = ChangeoverFSM(bank, strategy=st,
rng=np.random.RandomState(self.seed))
total = 0.0
waits = []
cnt = 0
det = []
for i in range(1, len(self.df)):
prev = self.df.iloc[i-1]
cur = self.df.iloc[i]
if prev["mc"] != cur["mc"]:
continue
if prev["family"] == cur["family"]:
continue
res = fsm.run(prev["family"], cur["family"])
total += res.total
waits.append(res.align_wait)
cnt += 1
for s in res.spans:
det.append({
"strategy":st,"oid":cur["oid"],"mc":cur["mc"],
"from":prev["family"],"to":cur["family"],
"action":s.name,"start":s.start,"dur":s.dur,
"is_wait":s.is_wait})
wr = (sum(waits)/total) if total>0 else 0.0
grade = "优" if total/cnt<=18 else ("良" if total/cnt<=28 else "超差")
rows.append({
"strategy":st,"total_co_time":total,
"n_change":cnt,"avg_co":total/max(cnt,1),
"avg_wait_ratio":wr,"grade":grade})
detail_frames.append(pd.DataFrame(det))
self.detail_ = pd.concat(detail_frames, ignore_index=True) if detail_frames else pd.DataFrame()
return pd.DataFrame(rows)
</details>
<details><summary></summary>
"""换产等级分类 (sklearn)。"""
import numpy as np
import pandas as pd
from typing import Dict
from sklearn.ensemble import RandomForestClassifier
from sklearn.model_selection import cross_val_score, KFold
class ChangeoverClassifier:
"""优(≤18min)/良(18~28)/超差(>28)。"""
def __init__(self, random_state: int = 42):
self.model_ = None
self.feat = ["cross_hyd", "wait_ratio", "align_t", "mc_load", "tool_change"]
@staticmethod
def _label(avg: float) -> str:
if avg <= 18: return "优"
if avg <= 28: return "良"
return "超差"
def build_feature(self, df: pd.DataFrame) -> pd.DataFrame:
return df[self.feat]
def fit(self, df: pd.DataFrame, avg_arr: np.ndarray):
y = np.array([self._label(a) for a in avg_arr])
self.model_ = RandomForestClassifier(
n_estimators=300, max_depth=6, min_samples_leaf=2,
random_state=42, n_jobs=-1)
self.model_.fit(df[self.feat].values, y)
return self
def cv(self, df: pd.DataFrame, avg_arr: np.ndarray) -> Dict:
y = np.array([self._label(a) for a in avg_arr])
kf = KFold(5, shuffle=True, random_state=42)
sc = cross_val_score(self.model_, df[self.feat].values, y,
cv=kf, scoring="f1_macro")
imp = dict(zip(self.feat, self.model_.feature_importances_))
return {"f1_macro": float(sc.mean()),
"importance": dict(sorted(imp.items(),
key=lambda x:x[1], reverse=True))}
def predict(self, df: pd.DataFrame) -> np.ndarray:
return self.model_.predict(df[self.feat].values)
</details>
<details><summary></summary>
"""FMS换产可视化 (matplotlib + networkx)。"""
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
from pathlib import Path
import networkx as nx
plt.rcParams["font.sans-serif"] = ["SimHei", "WenQuanYi Micro Hei", "DejaVu Sans"]
plt.rcParams["axes.unicode_minus"] = False
ACT_COLOR = {
"UNLOAD":"#8E44AD","CLEAN":"#3498DB","MOUNT":"#2980B9",
"ALIGN":"#16A085","LOAD_TOOL":"#2C3E50","TOOL_SET":"#27AE60",
"FIRST_PIECE":"#F39C12","WAIT_ALIGN":"#E74C3C"
}
class FMSViz:
def __init__(self, results_dir: str = "results"):
self.results_dir = Path(results_dir)
self.results_dir.mkdir(exist_ok=True)
def gantt(self, df_detail: pd.DataFrame, title="换产甘特"):
fig, ax = plt.subplots(figsize=(13,5))
subs = df_detail[df_detail.strategy=="base"].head(40)
ylabels=[]
for i,(_,r) in enumerate(subs.iterrows()):
c = ACT_COLOR.get(r.action,"#999")
ax.broken_barh([(r.start, r.dur)], (i*9, 7),
facecolors=c, edgecolor="k", linewidth=0.3)
ylabels.append(f"{r.oid}:{r.from}→{r.to}/{r.action}")
ax.set_yticks([])
ax.set_xlabel("时间 (min)", fontsize=12)
ax.set_title(f"{title} (红=等待, 绿=对刀)",
fontsize=13, fontweight="bold")
ax.grid(alpha=0.3)
# 图例
from matplotlib.patches import Patch
handles=[Patch(color=c,label=k) for k,c in ACT_COLOR.items()]
ax.legend(handles=handles, loc="upper right", fontsize=7, ncol=2)
plt.tight_layout()
plt.savefig(self.results_dir/"changeover_gantt.png",
dpi=150, bbox_inches="tight")
plt.close()
def action_breakdown(self, df_detail: pd.DataFrame):
fig, ax = plt.subplots(figsize=(11,6))
strat = df_detail.strategy.unique()
actions = ["UNLOAD","CLEAN","MOUNT","ALIGN","LOAD_TOOL","TOOL_SET","FIRST_PIECE","WAIT_ALIGN"]
bottom = np.zeros(len(strat))
for a in actions:
vals = []
for s in strat:
v = df_detail[(df_detail.strategy==s)&(df_detail.action==a)].dur.sum()
vals.append(v)
vals = np.array(vals)
ax.bar(strat, vals, bottom=bottom, label=a,
color=ACT_COLOR.get(a,"#999"))
bottom += vals
ax.set_ylabel("累计耗时 (min)", fontsize=12)
ax.set_title("换产动作耗时分解堆叠(按策略)",
fontsize=13, fontweight="bold")
ax.legend(fontsize=7, ncol=2); ax.grid(alpha=0.3)
plt.tight_layout()
plt.savefig(self.results_dir/"action_breakdown.png",
dpi=150, bbox_inches="tight")
plt.close()
def bipartite(self, df: pd.DataFrame):
fig, ax = plt.subplots(figsize=(11,7))
G = nx.Graph()
for _,r in df.iterrows():
G.add_node(("O",r.oid), bipartite=0)
G.add_node(("F",r.family), bipartite=1)
G.add_node(("M",r.mc), bipartite=1)
G.add_edge(("O",r.oid),("F",r.family))
G.add_edge(("F",r.family),("M",r.mc))
pos = nx.spring_layout(G, seed=42, k=0.8)
nO=[n for n in G if n[0]=="O"]
nF=[n for n in G if n[0]=="F"]
nM=[n for n in G if n[0]=="M"]
nx.draw_networkx_nodes(G,pos,nodelist=nO,node_color="#2980B9",
node_size=300,ax=ax,label="工单")
nx.draw_networkx_nodes(G,pos,nodelist=nF,node_color="#27AE60",
node_size=700,ax=ax,label="零件族")
nx.draw_networkx_nodes(G,pos,nodelist=nM,node_color="#E67E22",
node_size=900,ax=ax,label="机床")
nx.draw_networkx_edges(G,pos,edge_color="#aaa",alpha=0.6,ax=ax)
labels={n:n[1] for n in G}
nx.draw_networkx_labels(G,pos,labels,font_size=8,ax=ax)
ax.set_title("工单-零件族-机床 二部关联图",
fontsize=14,fontweight="bold")
ax.axis("off")
plt.tight_layout()
plt.savefig(self.results_dir/"order_fixture_bipartite.png",
dpi=150,bbox_inches="tight")
plt.close()
def strategy_curve(self, sweep_df: pd.DataFrame):
fig, ax = plt.subplots(figsize=(10,5))
ax.plot(sweep_df.strategy, sweep_df.avg_co, "o-", color="#2980B9",
lw=2, label="平均换产min")
ax.plot(sweep_df.strategy, sweep_df.avg_wait_ratio*100, "s--",
color="#E74C3C", label="等待占比%")
ax.axhline(18, color="#27AE60", ls=":", lw=2, label="优级线18min")
ax.set_xlabel("策略", fontsize=12)
ax.set_ylabel("平均换产(min) / 等待占比(%)", fontsize=12)
ax.set_title("换产策略对比曲线", fontsize=13, fontweight="bold")
ax.legend(); ax.grid(alpha=0.3)
plt.tight_layout()
plt.savefig(self.results_dir/"strategy_curve.png",
dpi=150,bbox_inches="tight")
plt.close()
def pred_scatter(self, y_true, y_pred):
fig, ax = plt.subplots(figsize=(8,8))
labels=["优","良","超差"]
ct=np.array([labels.index(y) for y in y_true])
cp=np.array([labels.index(y) for y in y_pred])
ax.scatter(ct,cp,c="#2980B9",s=50,edgecolors="k",alpha=0.8)
ax.plot([-0.5,2.5],[-0.5,2.5],"r--",lw=2,label="理想")
ax.set_xticks([0,1,2]);ax.set_xticklabels(labels)
ax.set_yticks([0,1,2]);ax.set_yticklabels(labels)
ax.set_xlabel("实际等级",fontsize=12)
ax.set_ylabel("预测等级",fontsize=12)
ax.set_title("换产等级 预测vs实际",fontsize=13,fontweight="bold")
ax.legend();ax.grid(alpha=0.3)
plt.tight_layout()
plt.savefig(self.results_dir/"grade_pred_scatter.png",
dpi=150,bbox_inches="tight")
plt.close()
def wait_flow(self, df_detail: pd.DataFrame):
fig, ax = plt.subplots(figsize=(11,5))
sub = df_detail[df_detail.strategy=="base"]
acts = ["ALIGN","WAIT_ALIGN","TOOL_SET"]
cnt = {a: len(sub[sub.action==a]) for a in acts}
x = np.arange(len(acts))
ax.bar(x, [cnt[a] for a in acts],
color=[ACT_COLOR[a] for a in acts])
for i,a in enumerate(acts):
ax.text(i, cnt[a]+0.1, f"{cnt[a]}次", ha="center", fontsize=10)
ax.set_xticks(x); ax.set_xticklabels(acts)
ax.set_ylabel("出现次数", fontsize=12)
ax.set_title("对刀耦合等待流向统计(找正→等待→对刀)",
fontsize=13, fontweight="bold")
ax.grid(alpha=0.3)
plt.tight_layout()
plt.savefig(self.results_dir/"wait_flow.png",
dpi=150,bbox_inches="tight")
plt.close()
</details>
<details><summary></summary>
"""合成FMS混流工单。"""
import numpy as np
import pandas as pd
from pathlib import Path
from typing import Optional
class SyntheticOrders:
"""
利用AI解决实际问题，如果你觉得这个工具好用，欢迎关注长安牧笛！
