《生产车间多智能体Agent实战笔记(7)：排程逻辑详解》

基于约束规划的生产排程系统：原理、实现与优化

概述

本系统是一个基于约束规划（Constraint Programming, CP）的生产排程解决方案，专门用于解决作业车间调度问题（Job Shop Scheduling Problem, JSSP）。系统采用 CP-SAT 求解器，在满足所有生产约束的前提下，自动寻找使总完工时间（makespan）最小的最优排程方案。相比传统启发式方法，本方案能提供数学意义上的最优解保证。

问题定义

输入数据

输出结果

• 每个工序的开始时间、结束时间、分配设备
• 总完工时间（makespan）
• 设备利用率、交期达成率、瓶颈设备

核心流程

1. 数据预处理 (_preprocess_data)

目的

将原始数据转换为求解器可用的结构，构建索引以提高查询效率。

构建的数据结构

# 产品到工艺路线的映射
product_to_processes: Dict[str, List[Process]]
# 例如: {"P01": [工序1, 工序2, 工序3]}

# 设备组到设备列表的映射
equipment_group_to_equipment: Dict[str, List[Equipment]]
# 例如: {"M01,M02": [设备M01, 设备M02]}

# 设备每日工作时长（分钟）
equipment_daily_capacity: Dict[str, float]
# 例如: {"M01": 480.0} # 8小时 = 480分钟

# 所有需要排程的(订单, 工序)对
job_operations: List[Tuple[Order, Process]]

时间范围估算

# 计算总工作量
total_work_minutes = Σ(订单数量 × 单件工时)

# 估算需要的天数（添加2倍缓冲）
estimated_days = (total_work_minutes / 平均每日产能) × 2
estimated_days = max(estimated_days, 30) # 至少30天

# 计算时间范围（horizon）
horizon = estimated_days × 24小时 × 60分钟

说明：horizon 是求解器搜索空间的上界，设置得太小可能无解，太大会影响求解效率。

2. 构建优化模型 (build_model)

决策变量

对每个工序 (订单i, 工序j) 创建以下变量：

示例：

工序: 车削, 可用设备: [M01, M02]

变量:
– presence_M01: 是否在M01上执行
– presence_M02: 是否在M02上执行
– interval_M01: 在M01上的时间区间（可选）
– interval_M02: 在M02上的时间区间（可选）

约束:
– presence_M01 + presence_M02 = 1

工序持续时间计算

duration = 单件标准工时 × 订单数量

示例：

• 单件工时：5分钟
• 订单数量：100件
• 持续时间：500分钟

3. 添加约束 (add_constraints)

约束1：工艺顺序约束

规则：工序必须按照工艺路线的顺序执行

# 对于同一订单的连续工序
工序i的结束时间 <= 工序i+1的开始时间

# 对于有显式前置工序的情况
前置工序的结束时间 <= 当前工序的开始时间

示例：

订单SO001的工艺路线：车削 → 铣削 → 检验

约束：
– 车削.end_time <= 铣削.start_time
– 铣削.end_time <= 检验.start_time

约束2：设备互斥约束

规则：同一台设备不能同时执行多个工序

model.AddNoOverlap(设备k的所有区间变量)

这个约束确保分配到同一设备的所有工序在时间上不重叠。

示例：

设备M01上的工序：
– 订单1的车削: [0, 500]
– 订单2的车削: [500, 1000] ✓ 不重叠
– 订单3的车削: [450, 950] ✗ 与订单2重叠，不允许

约束3：时间约束

规则：结束时间 = 开始时间 + 持续时间

end_time = start_time + (单件工时 × 订单数量)

约束4：时间窗口约束

规则：所有工序的开始时间必须非负

start_time >= 0

扩展：可以根据设备的 available_start 和 available_end 添加更复杂的时间窗口约束。

4. 设置优化目标 (set_objective)

目标函数

最小化总完工时间（makespan）

makespan = max(所有工序的结束时间)
minimize(makespan)

目标变量

# 创建makespan变量
makespan_var = model.NewIntVar(0, horizon, 'makespan')

# 添加约束：makespan >= 所有工序的结束时间
for each operation:
model.Add(makespan_var >= operation.end_time)

# 设置优化目标
model.Minimize(makespan_var)

其他可能的优化目标

虽然当前系统使用 makespan 作为目标，但也可以考虑：

• 最小化总延期时间
• 最小化设备切换次数
• 最大化设备利用率
• 多目标优化（加权组合）

5. 求解 (solve)

求解流程

1. 构建模型 (build_model)
2. 添加约束 (add_constraints)
3. 设置优化目标 (set_objective)
4. 创建求解器
5. 设置求解器参数
6. 执行求解
7. 提取结果

求解器参数

solver.parameters.max_time_in_seconds = 60.0 # 最大求解时间60秒

求解状态

6. 提取结果 (extract_solution)

步骤1：获取工序时间

从求解器获取每个工序的开始和结束时间（工作分钟）：

start_time = solver.Value(start_var) # 例如: 0
end_time = solver.Value(end_var) # 例如: 500
duration = end_time – start_time # 例如: 500分钟

步骤2：确定分配的设备

检查哪个 presence_var 的值为 1：

for equipment in available_equipment:
if solver.Value(presence_var[equipment]) == 1:
assigned_equipment = equipment
break

步骤3：转换为日历时间

核心函数：_work_minutes_to_datetime

转换逻辑：

# 输入：工作分钟数（连续时间）
# 输出：日历datetime（考虑每日工作时长）

# 1. 计算是第几个工作日
work_day = 工作分钟 // 每日工作分钟

# 2. 计算当天的工作分钟数
minutes_in_day = 工作分钟 % 每日工作分钟

# 3. 计算实际日期时间
result_date = 基准日期 + work_day天
result_datetime = result_date的8:00 + minutes_in_day

示例：

设备M01每日工作: 480分钟（8小时）
基准日期: 2024–03–06 08:00

工作分钟 = 1000
work_day = 1000 // 480 = 2（第2天）
minutes_in_day = 1000 % 480 = 40

result_datetime = 2024–03–08 08:40

为什么需要转换？

• 求解器使用连续时间（工作分钟）进行优化
• 用户需要看到实际的日历时间（考虑每天的工作时间）
• 转换确保结果符合实际生产场景

步骤4：计算 makespan

# 找到最早开始和最晚结束
min_start_time = min(op.start_time for op in operations)
max_end_time = max(op.end_time for op in operations)

# 计算日历时间跨度（小时）
makespan = (max_end_time – min_start_time).total_seconds() / 3600.0

算法特性

✅ 优势

⚠️ 局限性

性能优化与大规模问题处理

当订单数量达到数百、设备数量达到数十台时，CP-SAT 求解器的搜索空间会急剧膨胀，直接求解往往难以在可接受时间内收敛。本节介绍几种行之有效的性能优化策略，帮助系统在保持解质量的同时，显著提升求解效率。

1. 问题分解（Decomposition）

核心思想：将大规模问题拆分为若干规模较小的子问题，分别求解后再合并结果。常用的分解方式有两种：

按产品族分解

将工艺路线相似、共用设备组的产品归为同一产品族，每个产品族独立建模求解。这样可以大幅缩小每个子问题的变量规模。

def decompose_by_product_family(orders, processes, family_map):
"""
按产品族拆分订单，返回每个产品族的订单子集。
family_map: {产品编码: 产品族ID}
"""
families = {}
for order_id, product, quantity in orders:
fam = family_map[product]
families.setdefault(fam, []).append((order_id, product, quantity))
return families

# 使用示例
family_map = {"P01": "F1", "P02": "F1", "P03": "F2", "P04": "F2"}
families = decompose_by_product_family(orders, processes, family_map)

# 对每个产品族分别求解
for fam_id, fam_orders in families.items():
job_ops, horizon = preprocess_data(fam_orders, processes)
model, op_vars, all_intervals = build_model(job_ops, horizon)
add_constraints(model, job_ops, op_vars, all_intervals)
makespan_var = set_objective(model, job_ops, op_vars, horizon)
solver, status = solve(model, max_time_seconds=30.0)
# 合并各产品族的排程结果

按时间窗分解

将排程周期划分为多个时间窗口（如按周、按天），逐窗口滚动求解。每个窗口只考虑该时间段内的订单，求解完成后固定已排工序，再排下一个窗口。

def decompose_by_time_window(orders, processes, window_days=7):
"""
按时间窗拆分订单。这里简化为按交期所在周分组。
返回: {周序号: [订单列表]}
"""
from datetime import timedelta

windows = {}
base = datetime(2024, 3, 6, 8, 0)
for order_id, product, quantity, due_date in orders:
# 计算订单交期距离基准日期的周数
week_idx = (due_date – base).days // (window_days)
windows.setdefault(week_idx, []).append((order_id, product, quantity, due_date))
return windows

# 滚动求解：先排第1周，固定结果，再排第2周
for week_idx in sorted(windows.keys()):
week_orders = windows[week_idx]
job_ops, horizon = preprocess_data(week_orders, processes)
model, op_vars, all_intervals = build_model(job_ops, horizon)
add_constraints(model, job_ops, op_vars, all_intervals)
solver, status = solve(model, max_time_seconds=30.0)
# 固定本周已排工序，作为下一周的硬约束

注意：分解会损失一定的全局最优性，但通常能获得接近最优的可行解，且求解时间大幅缩短。

2. 启发式初始解（Warm Start）

CP-SAT 支持在求解前提供一个初始可行解，求解器会以此为起点继续搜索，从而显著加快收敛速度。启发式初始解可以通过贪心算法、最早交期优先（EDD）等简单规则快速生成。

def greedy_initial_solution(job_operations, daily_capacity):
"""
使用贪心算法生成初始解：按交期排序，依次为每个工序分配最早可用的设备。
返回: {工序key: (设备, 开始分钟, 结束分钟)}
"""
# 按交期排序（这里简化为按订单号排序）
sorted_ops = sorted(job_operations, key=lambda op: op["order"])

# 记录每台设备的最后可用时间
equipment_available_at = {equip: 0 for equip in daily_capacity}
initial_solution = {}

for op in sorted_ops:
key = (op["order"], op["op"])
duration = op["duration"]

# 选择最早可用的设备
best_equip = min(op["equipment"], key=lambda e: equipment_available_at[e])
start = equipment_available_at[best_equip]
end = start + duration

initial_solution[key] = (best_equip, start, end)
equipment_available_at[best_equip] = end

return initial_solution

# 将初始解注入模型
def apply_warm_start(model, op_vars, initial_solution):
"""将贪心初始解作为 hint 提供给求解器。"""
for key, (equip, start, end) in initial_solution.items():
if key in op_vars and equip in op_vars[key]:
model.AddHint(op_vars[key][equip]["start"], start)
model.AddHint(op_vars[key][equip]["end"], end)
model.AddHint(op_vars[key][equip]["presence"], 1)

说明：AddHint 只是引导搜索方向，不构成硬约束；即使初始解不可行，求解器也会自动忽略并继续搜索。

3. 求解器参数调优

CP-SAT 提供了丰富的参数来控制搜索行为，合理配置可以大幅提升求解效率。

并行搜索

现代 CPU 通常有多个核心，开启并行搜索可以让求解器同时探索多个搜索分支。

def solve_parallel(model, max_time_seconds=60.0, num_workers=8):
"""开启并行搜索的求解函数。"""
solver = cp_model.CpSolver()
solver.parameters.max_time_in_seconds = max_time_seconds
solver.parameters.num_workers = num_workers # 使用8个线程并行搜索
solver.parameters.log_search_progress = True # 打印搜索日志
status = solver.Solve(model)
return solver, status

搜索策略

CP-SAT 默认使用自动搜索策略，但对于特定问题，手动指定策略往往能获得更好的效果。

def solve_with_strategy(model, max_time_seconds=60.0):
"""配置搜索策略的求解函数。"""
solver = cp_model.CpSolver()
solver.parameters.max_time_in_seconds = max_time_seconds

# 搜索分支策略：优先选择影响最大的变量
solver.parameters.search_branching = cp_model.FIXED_SEARCH
# 或使用自动策略（默认）
# solver.parameters.search_branching = cp_model.AUTOMATIC_SEARCH

# 线性化级别：0=不线性化，1=部分，2=全部
solver.parameters.linearization_level = 2

# 记录冲突和分支，便于分析瓶颈
solver.parameters.log_search_progress = True

status = solver.Solve(model)
return solver, status

其他常用参数

def solve_tuned(model, max_time_seconds=60.0):
"""综合调优的求解函数。"""
solver = cp_model.CpSolver()
solver.parameters.max_time_in_seconds = max_time_seconds
solver.parameters.num_workers = 8

# 限制每次分支的候选变量数，减少内存占用
solver.parameters.max_num_conflicts = 100000

# 启用对称性破缺，减少等价解的搜索
solver.parameters.symmetry_level = 2

# 设置随机种子，保证结果可复现
solver.parameters.random_seed = 42

status = solver.Solve(model)
return solver, status

4. 综合优化示例

下面是一个综合运用上述策略的完整示例，演示如何在大规模场景下组织求解流程：

def solve_large_scale(orders, processes, daily_capacity, family_map,
max_time_seconds=60.0, num_workers=8):
"""
大规模排程综合求解入口。
策略：按产品族分解 + 贪心初始解 + 并行搜索。
"""
all_results = []

# 1. 按产品族分解
families = decompose_by_product_family(orders, processes, family_map)

for fam_id, fam_orders in families.items():
print(f"\\n===== 求解产品族 {fam_id} =====")

# 2. 数据预处理
job_ops, horizon = preprocess_data(fam_orders, processes)

# 3. 构建模型
model, op_vars, all_intervals = build_model(job_ops, horizon)

# 4. 添加约束
add_constraints(model, job_ops, op_vars, all_intervals)

# 5. 设置优化目标
makespan_var = set_objective(model, job_ops, op_vars, horizon)

# 6. 生成贪心初始解并注入
initial_solution = greedy_initial_solution(job_ops, daily_capacity)
apply_warm_start(model, op_vars, initial_solution)

# 7. 并行求解
solver, status = solve_parallel(model, max_time_seconds, num_workers)

# 8. 提取结果
if status in (cp_model.OPTIMAL, cp_model.FEASIBLE):
results = extract_solution(solver, job_ops, op_vars)
all_results.extend(results)
print(f"产品族 {fam_id} 求解完成，状态: {solver.StatusName(status)}")
else:
print(f"产品族 {fam_id} 求解失败，状态: {solver.StatusName(status)}")

return all_results

# 使用示例
if __name__ == "__main__":
# 假设有大量订单
orders = [
("SO001", "P01", 100, datetime(2024, 3, 10)),
("SO002", "P01", 200, datetime(2024, 3, 12)),
("SO003", "P02", 150, datetime(2024, 3, 15)),
# … 更多订单
]
family_map = {"P01": "F1", "P02": "F1", "P03": "F2"}

results = solve_large_scale(
orders, processes, daily_capacity, family_map,
max_time_seconds=60.0, num_workers=8
)
print(f"\\n共排程 {len(results)} 个工序")

5. 优化策略选择建议

实践建议：实际项目中，通常组合使用多种策略。例如「按产品族分解 + 贪心初始解 + 并行搜索」是兼顾求解速度与解质量的常用组合。建议先用小规模数据验证各策略效果，再逐步扩展到全量数据。

完整示例

输入数据

订单：

订单号: SO001
产品: P01
数量: 100件
交期: 2024-03-10

工艺路线（产品P01）：

1. 车削: 单件5分钟, 设备: M01,M02
2. 铣削: 单件3分钟, 设备: M03
3. 检验: 单件1分钟, 设备: M04

设备：

M01: 车床, 每日8小时
M02: 车床, 每日8小时
M03: 铣床, 每日8小时
M04: 检验台, 每日8小时

求解过程

1. 数据预处理

job_operations = [
(SO001, 车削),
(SO001, 铣削),
(SO001, 检验)
]

总工作量 = (5 + 3 + 1) × 100 = 900分钟

2. 构建模型

# 为车削工序创建变量
start_车削 = IntVar(0, horizon)
end_车削 = IntVar(0, horizon)
duration_车削 = 5 × 100 = 500分钟

presence_M01 = BoolVar()
presence_M02 = BoolVar()

约束: presence_M01 + presence_M02 = 1

3. 添加约束

# 工艺顺序
end_车削 <= start_铣削
end_铣削 <= start_检验

# 设备互斥
AddNoOverlap(M01的所有区间)
AddNoOverlap(M02的所有区间)
...

# 时间约束
end_车削 = start_车削 + 500
end_铣削 = start_铣削 + 300
end_检验 = start_检验 + 100

4. 求解

求解器找到最优解:
– 车削在M01上执行: 0–500分钟
– 铣削在M03上执行: 500–800分钟
– 检验在M04上执行: 800–900分钟

5. 转换为日历时间

基准日期: 2024–03–06 08:00
M01每日工作: 480分钟

车削:
start: 0分钟 → 2024–03–06 08:00
end: 500分钟 → 2024–03–07 08:20
(第1天: 0–480, 第2天: 480–500)

铣削:
start: 500分钟 → 2024–03–07 08:20
end: 800分钟 → 2024–03–08 09:40

检验:
start: 800分钟 → 2024–03–08 09:40
end: 900分钟 → 2024–03–08 11:20

输出结果

订单: SO001
总完工时间: 51.33小时（2天3小时20分）

工序详情:
┌────────┬──────┬─────────────────┬─────────────────┬──────────┐
│ 工序 │ 设备 │ 开始时间 │ 结束时间 │ 工时(h) │
├────────┼──────┼─────────────────┼─────────────────┼──────────┤
│ 车削 │ M01 │ 03-06 08:00 │ 03-07 08:20 │ 8.33 │
│ 铣削 │ M03 │ 03-07 08:20 │ 03-08 09:40 │ 5.00 │
│ 检验 │ M04 │ 03-08 09:40 │ 03-08 11:20 │ 1.67 │
└────────┴──────┴─────────────────┴─────────────────┴──────────┘

设备利用率:
– M01: 16.2%
– M02: 0%
– M03: 9.7%
– M04: 3.2%

交期达成: ✓ (完工时间 < 交期)

完整可运行代码

下面是一个完整的、可直接运行的 Python 示例，演示了如何调用上述核心函数来构建模型、添加约束、求解并解析结果。本示例使用 ortools 库的 CP-SAT 求解器。

from ortools.sat.python import cp_model
from datetime import datetime, timedelta

# ========== 示例数据 ==========
# 订单: (订单号, 产品, 数量)
orders = [
("SO001", "P01", 100),
]

# 工艺路线: (产品, 工序名, 单件工时(分钟), 可用设备列表)
processes = {
"P01": [
("车削", 5, ["M01", "M02"]),
("铣削", 3, ["M03"]),
("检验", 1, ["M04"]),
],
}

# 设备每日工作分钟数
daily_capacity = {"M01": 480, "M02": 480, "M03": 480, "M04": 480}

# 基准日期（用于将工作分钟转换为日历时间）
base_date = datetime(2024, 3, 6, 8, 0)

# ========== 1. 数据预处理 (_preprocess_data) ==========
def preprocess_data(orders, processes):
"""将原始数据转换为求解器可用的结构。"""
job_operations = [] # 所有需要排程的 (订单, 工序) 对
total_work_minutes = 0

for order_id, product, quantity in orders:
for op_name, unit_time, equipment_list in processes[product]:
duration = unit_time * quantity # 工序持续时间
total_work_minutes += duration
job_operations.append({
"order": order_id,
"product": product,
"op": op_name,
"duration": duration,
"equipment": equipment_list,
})

# 估算 horizon（时间范围上界）
avg_daily_capacity = sum(daily_capacity.values()) / len(daily_capacity)
estimated_days = max((total_work_minutes / avg_daily_capacity) * 2, 30)
horizon = int(estimated_days * 24 * 60)

return job_operations, horizon

# ========== 2. 构建优化模型 (build_model) ==========
def build_model(job_operations, horizon):
"""创建 CP-SAT 模型，为每个工序创建设备选择变量和时间变量。"""
model = cp_model.CpModel()

# 存储每个工序的变量
op_vars = {} # (order, op) -> {equipment: (start, end, presence, interval)}
all_intervals = {} # equipment -> [interval_var, …]

for idx, op in enumerate(job_operations):
key = (op["order"], op["op"])
op_vars[key] = {}
duration = op["duration"]

for equip in op["equipment"]:
# 为每台可用设备创建变量
start = model.NewIntVar(0, horizon, f"start_{key[0]}_{key[1]}_{equip}")
end = model.NewIntVar(0, horizon, f"end_{key[0]}_{key[1]}_{equip}")
presence = model.NewBoolVar(f"presence_{key[0]}_{key[1]}_{equip}")
interval = model.NewOptionalIntervalVar(
start, duration, end, presence, f"interval_{key[0]}_{key[1]}_{equip}"
)

op_vars[key][equip] = {
"start": start,
"end": end,
"presence": presence,
"interval": interval,
}

# 收集到设备区间列表（用于互斥约束）
all_intervals.setdefault(equip, []).append(interval)

# 约束：必须选择恰好一台设备
if len(op["equipment"]) > 1:
model.Add(
sum(op_vars[key][e]["presence"] for e in op["equipment"]) == 1
)

return model, op_vars, all_intervals

# ========== 3. 添加约束 (add_constraints) ==========
def add_constraints(model, job_operations, op_vars, all_intervals):
"""添加工艺顺序、设备互斥和时间约束。"""
# 约束1：工艺顺序约束（同一订单的连续工序）
for order_id, product, _ in orders:
ops = processes[product]
for i in range(len(ops) – 1):
cur_key = (order_id, ops[i][0])
next_key = (order_id, ops[i + 1][0])

# 当前工序的结束时间 <= 下一工序的开始时间
# 注意：需要处理多设备选择的情况，这里简化处理：
# 使用所有可能设备中的结束时间最大值作为工序结束时间
cur_end = max(op_vars[cur_key][e]["end"] for e in op_vars[cur_key])
next_start = min(op_vars[next_key][e]["start"] for e in op_vars[next_key])
model.Add(cur_end <= next_start)

# 约束2：设备互斥约束（同一设备不能同时执行多个工序）
for equip, intervals in all_intervals.items():
model.AddNoOverlap(intervals)

# 约束3：时间约束（结束时间 = 开始时间 + 持续时间）
# 已在创建 OptionalIntervalVar 时通过 duration 参数隐式满足

# ========== 4. 设置优化目标 (set_objective) ==========
def set_objective(model, job_operations, op_vars, horizon):
"""最小化总完工时间（makespan）。"""
makespan = model.NewIntVar(0, horizon, "makespan")

# makespan >= 所有工序的结束时间
for op in job_operations:
key = (op["order"], op["op"])
for equip in op_vars[key]:
model.Add(makespan >= op_vars[key][equip]["end"])

model.Minimize(makespan)
return makespan

# ========== 5. 求解 (solve) ==========
def solve(model, max_time_seconds=60.0):
"""创建求解器并求解。"""
solver = cp_model.CpSolver()
solver.parameters.max_time_in_seconds = max_time_seconds
status = solver.Solve(model)
return solver, status

# ========== 6. 提取结果 (extract_solution) ==========
def extract_solution(solver, job_operations, op_vars):
"""从求解结果中提取每个工序的时间、设备，并转换为日历时间。"""
results = []

for op in job_operations:
key = (op["order"], op["op"])

# 步骤1&2：获取时间并确定分配的设备
assigned_equip = None
start_min = None
end_min = None

for equip, vars in op_vars[key].items():
if solver.Value(vars["presence"]) == 1:
assigned_equip = equip
start_min = solver.Value(vars["start"])
end_min = solver.Value(vars["end"])
break

# 步骤3：转换为日历时间
start_dt = work_minutes_to_datetime(start_min, daily_capacity[assigned_equip])
end_dt = work_minutes_to_datetime(end_min, daily_capacity[assigned_equip])

results.append({
"order": op["order"],
"op": op["op"],
"equipment": assigned_equip,
"start_min": start_min,
"end_min": end_min,
"start_dt": start_dt,
"end_dt": end_dt,
"duration_h": (end_min – start_min) / 60.0,
})

return results

def work_minutes_to_datetime(work_minutes, daily_work_minutes):
"""将工作分钟数转换为日历时间（考虑每日工作时长）。"""
work_day = work_minutes // daily_work_minutes
minutes_in_day = work_minutes % daily_work_minutes
return base_date + timedelta(days=work_day, minutes=minutes_in_day)

# ========== 7. 格式化输出甘特图 ==========
def print_gantt_chart(results):
"""以文本甘特图形式展示各工序在设备上的时间安排。"""
if not results:
print("无排程结果可展示")
return

# 计算整体时间范围
all_start = min(r["start_min"] for r in results)
all_end = max(r["end_min"] for r in results)
total_span = max(all_end – all_start, 1)

# 甘特图宽度（字符数）
chart_width = 60

print("\\n" + "=" * 80)
print("排程甘特图（时间轴单位：工作分钟）")
print("=" * 80)

# 按设备分组
equipment_ops = {}
for r in results:
equipment_ops.setdefault(r["equipment"], []).append(r)

# 时间轴刻度
print(f"{'设备':<6}", end="")
for i in range(0, chart_width + 1, 10):
pos = int(i / chart_width * total_span)
print(f"{pos:>10}", end="")
print()

# 为每个设备绘制甘特条
for equip in sorted(equipment_ops.keys()):
ops = sorted(equipment_ops[equip], key=lambda x: x["start_min"])
print(f"{equip:<6}", end="")

# 创建空白时间轴
line = [" "] * chart_width

# 填充工序块
for op in ops:
start_pos = int((op["start_min"] – all_start) / total_span * chart_width)
end_pos = int((op["end_min"] – all_start) / total_span * chart_width)
end_pos = max(end_pos, start_pos + 1) # 至少占1个字符

# 用工序名首字填充
label = op["op"][0]
for i in range(start_pos, min(end_pos, chart_width)):
line[i] = label

print("".join(line))

# 图例
print("-" * 80)
legend = {r["op"][0]: r["op"] for r in results}
print("图例: " + ", ".join(f"{k}={v}" for k, v in legend.items()))
print()

# ========== 8. 计算并输出设备利用率 ==========
def print_equipment_utilization(results, daily_capacity, horizon_minutes=None):
"""计算并打印每台设备的利用率。"""
if not results:
print("无排程结果可计算利用率")
return

# 计算每台设备的总占用时间
equip_busy_time = {}
for r in results:
equip = r["equipment"]
busy = r["end_min"] – r["start_min"]
equip_busy_time[equip] = equip_busy_time.get(equip, 0) + busy

# 计算总时间跨度（从最早开始到最晚结束）
all_start = min(r["start_min"] for r in results)
all_end = max(r["end_min"] for r in results)
total_span = all_end – all_start

print("\\n" + "=" * 80)
print("设备利用率报告")
print("=" * 80)
print(f"{'设备':<8}{'占用时间(分)':<16}{'总跨度(分)':<16}{'利用率':<10}")
print("-" * 50)

for equip in sorted(daily_capacity.keys()):
busy = equip_busy_time.get(equip, 0)
utilization = (busy / total_span * 100) if total_span > 0 else 0
print(f"{equip:<8}{busy:<16}{total_span:<16}{utilization:<10.1f}%")

# 找出瓶颈设备
if equip_busy_time:
bottleneck = max(equip_busy_time.items(), key=lambda x: x[1])
print("-" * 50)
print(f"瓶颈设备: {bottleneck[0]}（占用 {bottleneck[1]} 分钟）")
print()

# ========== 主流程 ==========
if __name__ == "__main__":
# 1. 数据预处理
job_operations, horizon = preprocess_data(orders, processes)
print(f"总工作量: {sum(op['duration'] for op in job_operations)} 分钟")
print(f"Horizon: {horizon} 分钟 ({horizon / 1440:.1f} 天)")

# 2. 构建模型
model, op_vars, all_intervals = build_model(job_operations, horizon)

# 3. 添加约束
add_constraints(model, job_operations, op_vars, all_intervals)

# 4. 设置优化目标
makespan_var = set_objective(model, job_operations, op_vars, horizon)

# 5. 求解
solver, status = solve(model, max_time_seconds=60.0)

# 6. 提取并打印结果
if status in (cp_model.OPTIMAL, cp_model.FEASIBLE):
print(f"\\n求解状态: {solver.StatusName(status)}")
print(f"Makespan: {solver.Value(makespan_var)} 分钟")

results = extract_solution(solver, job_operations, op_vars)

print("\\n工序详情:")
print(f"{'工序':<6}{'设备':<6}{'开始时间':<20}{'结束时间':<20}{'工时(h)':<8}")
print("-" * 60)
for r in results:
print(
f"{r['op']:<6}{r['equipment']:<6}"
f"{r['start_dt'].strftime('%m-%d %H:%M'):<20}"
f"{r['end_dt'].strftime('%m-%d %H:%M'):<20}"
f"{r['duration_h']:<8.2f}"
)

# 7. 输出甘特图
print_gantt_chart(results)

# 8. 输出设备利用率
print_equipment_utilization(results, daily_capacity)

else:
print(f"求解失败，状态: {solver.StatusName(status)}")

运行说明：

扩展方向

1. 多目标优化

# 加权目标函数
objective = w1 × makespan + w2 × 延期惩罚 + w3 × 设备切换成本

2. 动态排程

• 支持插单、急单
• 实时调整排程
• 考虑在制品状态

3. 高级约束

• 设备维护时间窗口
• 人员技能匹配
• 物料可用性
• 能源消耗限制

4. 启发式算法

对于超大规模问题，可以结合：

• 遗传算法
• 模拟退火
• 禁忌搜索

5. 机器学习

• 学习历史数据预测工时
• 智能调整优先级
• 预测瓶颈设备

总结

本排程系统通过约束规划技术，将复杂的生产排程问题转化为数学优化问题，由求解器自动找到最优解。系统的核心优势在于：

这使得系统能够高效地处理实际生产中的复杂排程问题