周二上午,FMS控制室。
“老郑,2号加工中心又空转了,前面缓冲站堆了5个托盘,后面却等夹具,”老郑盯着调度屏,“缓冲区设了4位,混流一来,A类件占满,B类件没地方放,只能堵在线上等,可夹具库那边也排着队,两边都在等,产能就漏了。”
我接上导出的托盘到达日志、缓冲区占用序列、夹具占用、MC状态、零件等待时长、夹具等待时长。
“这里面有啥?”我问。
“托盘ID、到达时刻、零件类型A/B/C、缓冲区位占用、目标MC、所需夹具号、夹具释放时刻都有,”老郑说,“可系统只画个Gantt,不模拟‘缓冲区容量从4改到8改到12,夹具等待和零件等待怎么此消彼长’。想定容量,得真跑几天产线看堵不堵。”
“最亏的是拐点区,”老郑补一句,“容量4的时候零件等得久,加到8零件等待降了,但夹具被占着不释放,夹具等待反而涨;加到12,两边都没明显改善,还占场地。中间有个最优解,系统算不出来。”
“我就想干一件事,”老郑说,“给到达过程+服务时间+夹具约束,扫不同缓冲区容量,仿真出零件等待、夹具等待、利用率,画出来对比,标出最优容量,像个小缓冲站容量仿真器,不用真改产线布局。”
“FMS不是看设备转没转,”我接话,“是看‘托盘到了没位子、有位子没夹具、有夹具MC又忙’这条链。用 numpy+heapq 做事件驱动离散仿真,scipy 拟合到达分布,pandas 管工单,matplotlib 画容量对比曲线+等待分布,networkx 建‘到达-缓冲-夹具-MC’二部图,sklearn 做等待等级分类。”
“对,”老郑点头,“要能说清‘容量4:零件等12.4min,夹具等5.1min;容量8:零件4.8min,夹具6.3min;容量10:零件4.1min,夹具6.5min,拐点就在8~10之间,主因是B类密集到达+夹具数不足’。”
“OOP 封好,”我开工程,“工单加载器、事件调度器、缓冲区模型、夹具资源池、等待分析器、容量扫描器、可视化器,合成多容量对比,下载就能跑。”
敲了行原型:
# 目标: 扫缓冲区容量 → 事件驱动仿真 → 零件等待 vs 夹具等待 → 最优容量
# 方法: 离散事件(heapq)+二部资源图+RF分类+对比曲线
老郑凑近看:“那以后看报告:容量-等待对比曲线,零件/夹具等待分布直方图,资源竞争二部图,最优容量标绿,分类散点。新线设计前先扫容量,不用先焊货架。”
“对,”我接话,“缓冲站仿真不是‘画个队列’,是‘提前看见容量加到几刚好不堵’。数字孪生里挂这个缓冲看板,就是老郑的‘容量标尺’。”
一、实际应用场景(真实痛点)
场景设定:柔性制造系统(FMS)含 3 台加工中心(MC)、1 个中央缓冲站(托盘位)、共享夹具库(F1×2、F2×1)。混流加工 A/B/C 三类零件,B 类到货密集且需 F2。缓冲站容量固定为 4 位时,混流高峰频繁出现“托盘堵在入口 / 夹具排着队释放”并存现象,设备空转与在制品积压同时发生。
现场原话(叙事化):
“不是设备不够,”老郑说,“是缓冲位卡得死。B类一波来5个,位子被占满,A类进不来;可F2就1套,B类占着F2加工完还不释放,后面B类在缓冲里干等,夹具也在等空托盘位回流,两头耗。”
“最亏的是定容量,”老郑说,“以前按经验焊了4个位,后来加焊到12个,发现零件等待是降了,可夹具等待没降反升,场地还浪费。到底几个位最划算,没人算得清。”
核心矛盾:“固定缓冲容量+看Gantt” 与 “容量扫描→事件驱动仿真→零件等待/夹具等待解耦量化→最优容量+资源竞争图” 之间的断层。
二、痛点分析(映射到滨州职业学院《先进制造技术》课程模型)
《先进制造技术》课程模块 本篇痛点对应
柔性制造系统FMS与先进生产管理:缓冲站、托盘系统、资源调度、在制品控制、产能平衡 容量-等待权衡全链建模
先进制造技术基础:系统效率、瓶颈分析、排队现象 零件/夹具双等待量化
数控加工与CAD/CAM技术:MC加工节拍、装夹基准 夹具占用与装夹耦合
智能制造与数字孪生:FMS状态数字映射、缓冲看板 缓冲资源挂孪生
先进制造新模式:数据驱动产线配置优化 容量知识库沉淀
一句话总结:我们需要一个“缓冲区容量扫描→离散事件仿真→零件等待 vs 夹具等待→资源竞争二部图→最优容量判定+分类”程序,实现从“凭经验焊货架”到“仿真定容量”的闭环。
三、核心逻辑讲解(大白话)
3.1 问题本质:把缓冲站想成“车间门口的候车室”
把缓冲站想成一个候车室,托盘是乘客,夹具是登机口,MC是飞机:
* 容量4 = 候车室只有4把椅子,人多了站门口堵着 → 零件等待长
* 零件等待 = 托盘到了没位子,或有位子但MC忙
* 夹具等待 = 有托盘有位子,但对应夹具被别的件占着
* 夹具释放卡住 = 加工完的件占着位子不流转,夹具也跟着卡
* 容量加大 = 椅子多了,零件不站门口了,但登机口(夹具)还是那几个,反而被“占座式”占用,夹具等待涨
* 最优容量 = 椅子加到某个数,两边等待都收敛,再多加就是浪费
* 试产定容量 = 真焊货架试几天,贵且不可逆
3.2 业务逻辑 → 代码映射
输入工单流+资源配置
│
▼ WorkOrderLoader (pandas)
读取:
工单号, 到达时刻, 类型A/B/C, 目标MC, 所需夹具, 加工时长
│
▼ EventScheduler (numpy + heapq)
离散事件引擎:
事件: 到达 / 入缓冲 / 申请夹具 / 开工 / 完工 / 出缓冲
按仿真钟推进, 记录各阶段时间戳
│
▼ BufferStation (纯numpy结构)
缓冲站模型:
容量C, FIFO入站, 满则入口排队
统计入站等待(零件等待组成)
│
▼ FixturePool (numpy)
夹具资源池:
F1×2, F2×1, 按类型绑定
申请不到则进夹具等待队列
│
▼ WaitAnalyzer (numpy + scipy)
等待分析:
零件等待 = 到达→开工
夹具等待 = 可加工→拿到夹具
拟合等待分布(lognorm), 算P95
│
▼ CapacitySweeper
容量扫描:
C ∈ [4,6,8,10,12] 各跑N次复现
输出等待随容量曲线
│
▼ WaitClassifier (sklearn)
等待等级分类:
特征: 容量, 类型, 到达间隔, 夹具竞争度
标签: 优(≤3min)/良(3~8min)/劣(>8min)
RF分类 + 5折宏F1
│
▼ FMSViz (matplotlib + networkx)
可视化:
1. 容量-零件等待/夹具等待对比曲线
2. 不同容量等待分布直方图
3. 资源竞争二部图(托盘↔夹具↔MC)
4. 容量-利用率热力条
5. 等待等级预测vs实际散点
6. 单容量甘特示意
│
▼ SyntheticOrders (numpy)
合成数据:
60工单 A/B/C混流, B类密集到达, 需F2
3.3 为什么不能“看Gantt”
视角 问题
看MC甘特 空转时看不出是等件还是等夹具
看缓冲占用 只知满了,不知零件等还是夹具等
凭经验加位 加到12才发现夹具等待反升
容量扫描曲线 看见零件等待↓夹具等待↑的拐点
二部图 谁在跟谁抢资源一眼清
RF分类 新工单流直接判等待等级
3.4 分析前后对比
维度 传统方式 本程序
定容量 经验焊4位 扫容量出拐点
等待归因 混在一起看 零件等待/夹具等待解耦
最优解 拍脑袋 加权成本最小
资源竞争 靠调度屏猜 二部图量化
知识沉淀 产线改完才知 容量知识库可复用
四、OOP 代码实现
4.1 项目结构
fms_buffer_sim/
├── fms_buffer_sim/
│ ├── __init__.py
│ ├── work_order_loader.py # 工单加载
│ ├── event_scheduler.py # 离散事件引擎(heapq+numpy)
│ ├── buffer_station.py # 缓冲站模型
│ ├── fixture_pool.py # 夹具资源池
│ ├── wait_analyzer.py # 等待分析(scipy)
│ ├── capacity_sweeper.py # 容量扫描
│ ├── wait_classifier.py # 等待分类(sklearn)
│ ├── fms_viz.py # 可视化
│ └── synthetic_orders.py # 合成工单
├── tests/
│ ├── __init__.py
│ └── test_buffer.py
├── results/
│ ├── capacity_wait_curve.png
│ ├── wait_dist_hist.png
│ ├── resource_bipartite.png
│ ├── util_heatbar.png
│ ├── wait_pred_scatter.png
│ ├── gantt_sample.png
│ ├── buffer_detail.csv
│ └── buffer_report.txt
└── run_buffer.py
4.2 核心源码
<details>
<summary></summary>
"""FMS工单加载器。"""
import pandas as pd
from pathlib import Path
class WorkOrderLoader:
"""加载混流工单表。"""
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", "arr_t", "ptype", "mc", "fixture", "proc_t"]
miss = [c for c in req if c not in df.columns]
if miss:
raise ValueError(f"缺列: {miss}")
for c in ["arr_t", "proc_t", "mc"]:
df[c] = pd.to_numeric(df[c], errors="coerce")
return df.dropna(subset=req).reset_index(drop=True)
def summary(self, df: pd.DataFrame) -> str:
s = f"工单数: {len(df)}\n"
s += f"类型分布: {df['ptype'].value_counts().to_dict()}\n"
s += f"到达时段: {df.arr_t.min():.0f}~{df.arr_t.max():.0f}s\n"
s += f"夹具需求: {df['fixture'].value_counts().to_dict()}"
return s.rstrip()
</details>
<details>
<summary></summary>
"""离散事件调度引擎 (heapq + numpy)。"""
import heapq
import numpy as np
from dataclasses import dataclass, field
from typing import List, Dict
@dataclass
class OrderTrace:
oid: str
ptype: str
arr_t: float
in_buf_t: float
req_fix_t: float
start_t: float
end_t: float
part_wait: float
fix_wait: float
mc: int
fixture: str
class EventScheduler:
"""
事件类型:
ARR 到达, BUF_IN 入缓冲, REQ_FIX 申请夹具,
START 开工, END 完工, BUF_OUT 出缓冲
仿真钟用堆维护
"""
def __init__(self, buffer_cap: int, fixture_pool, buffer_station,
rng: np.random.RandomState = None):
self.cap = buffer_cap
self.fp = fixture_pool
self.bs = buffer_station
self.rng = rng or np.random.RandomState(42)
self.traces: List[OrderTrace] = []
def run(self, orders: List[dict]) -> List[OrderTrace]:
heap = []
for o in orders:
heapq.heappush(heap, (o["arr_t"], "ARR", o))
clock = 0.0
mc_free = {1: 0.0, 2: 0.0, 3: 0.0}
pending = [] # 已入缓冲待夹具
while heap:
clock, typ, payload = heapq.heappop(heap)
if typ == "ARR":
o = payload
in_buf_t = clock
self.bs.occupy()
heapq.heappush(pending, o)
o["_in_buf"] = in_buf_t
# 尝试分配
heapq.heappush(heap, (clock, "TRY_ALLOC", o))
elif typ == "TRY_ALLOC":
o = payload
mc = int(o["mc"])
fix = o["fixture"]
if self.bs.count < self.cap and clock >= mc_free[mc]:
got = self.fp.try_acquire(fix, clock)
if got is not None:
req_fix_t = clock
start = max(clock, mc_free[mc])
end = start + float(o["proc_t"])
mc_free[mc] = end
self.bs.release()
self.fp.release(fix, end)
trace = OrderTrace(
oid=o["oid"], ptype=o["ptype"],
arr_t=o["arr_t"], in_buf_t=o["_in_buf"],
req_fix_t=req_fix_t, start_t=start,
end_t=end,
part_wait=start - o["arr_t"],
fix_wait=start - req_fix_t,
mc=mc, fixture=fix)
self.traces.append(trace)
else:
# 等夹具, 推后重试
heapq.heappush(heap, (clock+0.5, "TRY_ALLOC", o))
else:
# 缓冲满或MC忙, 推后
heapq.heappush(heap, (clock+0.5, "TRY_ALLOC", o))
return self.traces
</details>
<details><summary></summary>
"""缓冲站模型。"""
class BufferStation:
def __init__(self, cap: int):
self.cap = cap
self.count = 0
def occupy(self):
if self.count < self.cap:
self.count += 1
return True
return False
def release(self):
if self.count > 0:
self.count -= 1
def reset(self):
self.count = 0
</details>
<details><summary></summary>
"""夹具资源池。"""
from dataclasses import dataclass
from typing import Dict, Optional
class FixturePool:
"""
F1: 2套, F2: 1套
"""
def __init__(self):
self.stock: Dict[str, int] = {"F1": 2, "F2": 1}
self.busy: Dict[str, int] = {"F1": 0, "F2": 0}
def try_acquire(self, fix: str, t: float) -> Optional[float]:
if self.busy[fix] < self.stock[fix]:
self.busy[fix] += 1
return t
return None
def release(self, fix: str, t: float):
if self.busy[fix] > 0:
self.busy[fix] -= 1
def reset(self):
self.busy = {"F1": 0, "F2": 0}
</details>
<details><summary></summary>
"""等待分析 (numpy + scipy)。"""
import numpy as np
from dataclasses import dataclass
from scipy.stats import lognorm
@dataclass
class WaitStat:
part_mean: float
part_p95: float
fix_mean: float
fix_p95: float
params: tuple
class WaitAnalyzer:
@staticmethod
def stat(part_wait, fix_wait):
pw = np.array(part_wait, dtype=float)
fw = np.array(fix_wait, dtype=float)
# lognorm拟合零件等待
pos = np.clip(pw[pw>0], 1e-3, None)
s, loc, scale = lognorm.fit(pos, floc=0)
p95 = lognorm.ppf(0.95, s, loc, scale)
return WaitStat(float(pw.mean()), float(p95),
float(fw.mean()), float(np.percentile(fw,95)),
(s,loc,scale))
@staticmethod
def cost(part_mean, fix_mean, cap,
cost_part=1.0, cost_fix=1.2, cost_slot=0.4):
# 综合成本: 等待损失+占位成本
return part_mean*cost_part + fix_mean*cost_fix + cap*cost_slot
</details>
<details><summary></summary>
"""容量扫描器。"""
import numpy as np
import pandas as pd
from typing import List, Dict
from .event_scheduler import EventScheduler
from .buffer_station import BufferStation
from .fixture_pool import FixturePool
from .wait_analyzer import WaitAnalyzer
class CapacitySweeper:
def __init__(self, orders: List[dict], caps: List[int] = None,
reps: int = 5, seed: int = 42):
self.orders = orders
self.caps = caps or [4,6,8,10,12]
self.reps = reps
self.seed = seed
def sweep(self) -> pd.DataFrame:
rows = []
for cap in self.caps:
part_all, fix_all = [], []
for r in range(self.reps):
rng = np.random.RandomState(self.seed + r)
bs = BufferStation(cap)
fp = FixturePool()
sch = EventScheduler(cap, fp, bs, rng=rng)
orders = [dict(o) for o in self.orders]
traces = sch.run(orders)
part_all += [t.part_wait for t in traces]
fix_all += [t.fix_wait for t in traces]
st = WaitAnalyzer.stat(part_all, fix_all)
c = WaitAnalyzer.cost(st.part_mean, st.fix_mean, cap)
rows.append({
"cap": cap,
"part_wait_mean": st.part_mean,
"part_wait_p95": st.part_p95,
"fix_wait_mean": st.fix_mean,
"fix_wait_p95": st.fix_p95,
"total_cost": c,
})
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 WaitClassifier:
"""优(≤3min)/良(3~8min)/劣(>8min) 三分类, 按零件等待。"""
def __init__(self, random_state: int = 42):
self.model_ = None
self.feat = ["cap", "arr_gap", "fix_contend", "ptype_enc", "mc"]
@staticmethod
def _label(w: float) -> str:
w = w/60.0 # 秒转分
if w <= 3: return "优"
if w <= 8: return "良"
return "劣"
def fit(self, df: pd.DataFrame, part_wait_sec: np.ndarray):
y = np.array([self._label(w) for w in part_wait_sec])
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, part_wait_sec: np.ndarray) -> Dict:
y = np.array([self._label(w) for w in part_wait_sec])
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
GRADE = {"优":"#27AE60","良":"#F39C12","劣":"#E74C3C"}
class FMSViz:
def __init__(self, results_dir: str = "results"):
self.results_dir = Path(results_dir)
self.results_dir.mkdir(exist_ok=True)
def capacity_curve(self, df: pd.DataFrame):
fig, ax1 = plt.subplots(figsize=(11,6))
ax1.plot(df.cap, df.part_wait_mean/60, "o-", color="#2980B9", lw=2,
label="零件等待(分)")
ax1.plot(df.cap, df.fix_wait_mean/60, "s--", color="#E67E22", lw=2,
label="夹具等待(分)")
ax1.set_xlabel("缓冲区容量 (位)", fontsize=12)
ax1.set_ylabel("平均等待 (min)", fontsize=12)
ax2 = ax1.twinx()
ax2.plot(df.cap, df.total_cost, "^-", color="#7F8C8D", lw=1.5,
label="综合成本")
ax1.set_title("容量-等待对比曲线(找拐点)",
fontsize=13, fontweight="bold")
# 标最优
best = df.loc[df.total_cost.idxmin()]
ax1.axvline(best.cap, color="#27AE60", ls=":", lw=2,
label=f"最优容量{best.cap}")
lines1,labels1 = ax1.get_legend_handles_labels()
lines2,labels2 = ax2.get_legend_handles_labels()
ax1.legend(lines1+lines2, labels1+labels2, loc="upper right")
ax1.grid(alpha=0.3)
plt.tight_layout()
plt.savefig(self.results_dir/"capacity_wait_curve.png",
dpi=150, bbox_inches="tight")
plt.close()
def wait_hist(self, traces_by_cap: dict):
fig, axes = plt.subplots(1,2, figsize=(14,5))
for cap, arr in traces_by_cap.items():
axes[0].hist(np.array(arr["part"])/60, bins=20, alpha=0.5,
label=f"cap={cap}")
axes[1].hist(np.array(arr["fix"])/60, bins=20, alpha=0.5,
label=f"cap={cap}")
axes[0].set_title("零件等待分布", fontsize=12, fontweight="bold")
axes[1].set_title("夹具等待分布", fontsize=12, fontweight="bold")
axes[0].set_xlabel("min"); axes[1].set_xlabel("min")
axes[0].legend(); axes[1].legend(); axes[0].grid(alpha=0.3)
plt.tight_layout()
plt.savefig(self.results_dir/"wait_dist_hist.png",
dpi=150, bbox_inches="tight")
plt.close()
def bipartite(self, traces):
fig, ax = plt.subplots(figsize=(11,7))
G = nx.Graph()
for i,t in enumerate(traces[:40]):
pn = f"P{i}"
G.add_node(pn, bipartite=0)
fn = t.fixture
G.add_node(fn, bipartite=1)
G.add_edge(pn, fn)
G.add_node(f"MC{t.mc}", bipartite=1)
G.add_edge(pn, f"MC{t.mc}")
pos = {}
pos.update({n:(0, i*0.3) for i,n in enumerate([n for n in G if G.nodes[n]['bipartite']==0])})
pos.update({n:(1, i*0.6) for i,n in enumerate([n for n in G if G.nodes[n]['bipartite']==1])})
nx.draw_networkx_nodes(G,pos,nodelist=[n for n in G if G.nodes[n]['bipartite']==0],
node_color="#3498DB",node_size=300,ax=ax)
nx.draw_networkx_nodes(G,pos,nodelist=[n for n in G if G.nodes[n]['bipartite']==1],
node_color="#E67E22",node_size=600,ax=ax)
nx.draw_networkx_edges(G,pos,edge_color="#888",ax=ax,alpha=0.6)
nx.draw_networkx_labels(G,pos,font_size=7,ax=ax)
ax.set_title("托盘-夹具-MC 资源竞争二部图",
fontsize=14, fontweight="bold")
ax.axis("off")
plt.tight_layout()
plt.savefig(self.results_dir/"resource_bipartite.png",
dpi=150, bbox_inches="tight")
plt.close()
def util_bar(self, df, mc_util):
fig, ax = plt.subplots(figsize=(10,5))
x = np.arange(len(df))
ax.bar(x-0.2, df.part_wait_mean/60, 0.4, label="零件等待", color="#2980B9")
ax.bar(x+0.2, df.fix_wait_mean/60, 0.4, label="夹具等待", color="#E67E22")
ax.set_xticks(x); ax.set_xticklabels(df.cap)
for i,u in enumerate(mc_util):
ax.text(i, 0.1, f"MC{u:.2f}", ha="center", fontsize=8)
ax.set_xlabel("容量", fontsize=12)
ax.set_ylabel("等待(min)", fontsize=12)
ax.set_title("容量-等待柱状+MC利用率标注",
fontsize=13, fontweight="bold")
ax.legend(); ax.grid(alpha=0.3)
plt.tight_layout()
plt.savefig(self.results_dir/"util_heatbar.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/"wait_pred_scatter.png",
dpi=150,bbox_inches="tight")
plt.close()
def gantt(self, traces, cap):
fig, ax = plt.subplots(figsize=(12,4))
for i,t in enumerate(traces[:25]):
ax.broken_barh([(t.start_t, t.end_t-t.start_t)], (t.mc*10, 8),
facecolors="#27AE60" if t.ptype=="A" else
("#2980B9" if t.ptype=="B" else "#8E44AD"))
ax.plot([t.arr_t,t.start_t],[t.mc*10+4,t.mc*10+4],
color="#E74C3C",lw=1.2,alpha=0.7)
ax.set_yticks([10,20,30])
ax.set_yticklabels(["MC1","MC2","MC3"])
ax.set_xlabel("仿真时间 (s)", fontsize=12)
ax.set_title(f"单容量甘特示意(cap={cap}) 红线=等待段",
fontsize=13, fontweight="bold")
ax.grid(alpha=0.3)
plt.tight_layout()
plt.savefig(self.results_dir/"gantt_sample.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, List, Dict
class SyntheticOrders:
"""
60工单 A/B/C混流
B类密集到达, 需F2
A类需F1, C类需F1
"""
def __init__(self, rng: Optional[np.random.RandomState] = None):
self.rng = rng or np.random.Rand
利用AI解决实际问题,如果你觉得这个工具好用,欢迎关注长安牧笛!