用Python实战JSSP车间调度：从编码到甘特图的全流程实现

车间作业调度问题（JSSP）是制造业和计算机科学领域的一个经典难题。传统教学中往往从复杂的数学公式入手，让许多学习者望而生畏。本文将采用完全不同的路径——通过Python代码实现整个调度流程，让抽象问题变得可视化、可操作。我们将从零开始构建一个完整的JSSP解决方案，包括工序编码、前插式解码，最终用matplotlib绘制专业甘特图。

1. 理解JSSP问题本质

JSSP的核心是解决多个工件在多台机器上的加工顺序问题。每个工件包含若干道工序，每道工序需要在特定机器上花费固定时间完成。关键在于安排工序顺序，使得所有工件都能在最短时间内完成加工。

举个例子，假设我们有3个工件（Job1、Job2、Job3）和3台机器（Machine1、Machine2、Machine3）。每个工件的加工路线和时间可能如下：

jobs_data = {
    'Job1': [('Machine1', 15), ('Machine2', 25), ('Machine3', 18)],
    'Job2': [('Machine2', 10), ('Machine3', 20), ('Machine1', 7)],
    'Job3': [('Machine3', 8), ('Machine1', 12), ('Machine2', 10)]
}

这种表示方法清晰地展示了每个工件的加工路径：

• Job1：先在Machine1加工15分钟，然后在Machine2加工25分钟，最后在Machine3加工18分钟
• Job2和Job3也有各自的加工顺序

JSSP的三大核心约束：

1. 工序顺序约束：每个工件的工序必须按指定顺序执行
2. 机器独占约束：一台机器同一时间只能加工一个工件
3. 不可中断约束：工序一旦开始就不能被中断

2. 基于工序的编码方案

编码是将调度问题转化为计算机可处理形式的关键步骤。我们采用基于工序的编码（Operation-based Encoding），这是一种直观且广泛使用的表示方法。

2.1 编码原理

染色体由所有工序的排列组成，每个基因代表一个工序。同一工件的不同工序用相同编号表示，通过出现顺序区分。例如，对于3工件3机器的问题，一个可能的编码是：

chromosome = [1, 2, 3, 1, 2, 3, 2, 1, 3]

这个编码表示：

• 前三个数字1,2,3分别代表三个工件的第一个工序
• 接下来的1,2,3是三个工件的第二个工序
• 最后三个数字2,1,3是三个工件的第三个工序（顺序被打乱）

2.2 Python实现编码生成

我们可以用以下函数随机生成合法的染色体：

import random

def generate_chromosome(jobs_data):
    num_jobs = len(jobs_data)
    num_operations = len(next(iter(jobs_data.values())))
    chromosome = []
    
    # 每个工件的工序编号列表
    job_ops = {job: list(range(1, num_operations+1)) for job in jobs_data}
    
    # 随机生成染色体
    while any(job_ops.values()):
        available_jobs = [job for job in job_ops if job_ops[job]]
        selected_job = random.choice(available_jobs)
        chromosome.append(selected_job)
        job_ops[selected_job].pop(0)
    
    return chromosome

这个函数确保生成的染色体满足：

• 每个工件恰好出现num_operations次
• 同一工件的工序按顺序出现（先工序1，再工序2，依此类推）

3. 前插式解码算法实现

解码是将染色体转化为实际调度方案的过程。我们采用前插式解码（Insertion Decoding），这是一种高效且能产生较优解的方法。

3.1 解码算法步骤

1. 初始化每台机器的可用时间表（初始为空）
2. 按染色体顺序处理每个工序
3. 对于当前工序：
   • 确定它需要在哪台机器上加工
   • 在该机器的已安排工序中寻找最早可插入的时间段
   • 考虑工件前序工序的完成时间约束
4. 将工序插入找到的时间段
5. 更新机器的可用时间表

3.2 Python实现解码

def decode_chromosome(chromosome, jobs_data):
    # 初始化数据结构
    schedule = {machine: [] for machine in set(op[0] for job in jobs_data.values() for op in job)}
    job_progress = {job: {'current_op': 0, 'completion_time': 0} for job in jobs_data}
    
    for job in chromosome:
        # 获取当前工序信息
        op_index = job_progress[job]['current_op']
        machine, duration = jobs_data[job][op_index]
        prev_completion = job_progress[job]['completion_time']
        
        # 在当前机器上寻找合适的插入位置
        machine_schedule = schedule[machine]
        inserted = False
        
        # 尝试在机器时间表的空隙中插入
        for i in range(len(machine_schedule) + 1):
            # 计算可能的开始和结束时间
            if i == 0:
                start_time = max(prev_completion, 0)
            else:
                start_time = max(prev_completion, machine_schedule[i-1]['end'])
                
            if i < len(machine_schedule):
                end_time = start_time + duration
                if end_time > machine_schedule[i]['start']:
                    continue  # 与已有工序冲突，尝试下一个位置
            else:
                end_time = start_time + duration
                
            # 找到合适位置，插入工序
            machine_schedule.insert(i, {
                'job': job,
                'op': op_index,
                'start': start_time,
                'end': end_time,
                'duration': duration
            })
            inserted = True
            break
        
        if not inserted:
            # 如果前面没有合适位置，则添加到末尾
            start_time = max(prev_completion, machine_schedule[-1]['end'] if machine_schedule else 0)
            machine_schedule.append({
                'job': job,
                'op': op_index,
                'start': start_time,
                'end': start_time + duration,
                'duration': duration
            })
        
        # 更新工件进度
        job_progress[job]['current_op'] += 1
        job_progress[job]['completion_time'] = start_time + duration
    
    return schedule

4. 甘特图可视化

可视化是理解调度结果的关键。我们将使用matplotlib绘制专业甘特图，直观展示调度方案。

4.1 甘特图绘制函数

import matplotlib.pyplot as plt
import matplotlib.patches as patches

def plot_gantt(schedule):
    fig, ax = plt.subplots(figsize=(12, 6))
    
    # 为每台机器创建y轴位置
    machines = sorted(schedule.keys())
    y_ticks = []
    y_labels = []
    
    for i, machine in enumerate(machines):
        y_pos = i + 1
        y_ticks.append(y_pos)
        y_labels.append(machine)
        
        # 绘制每个工序的矩形
        for op in schedule[machine]:
            job = op['job']
            start = op['start']
            duration = op['duration']
            
            # 创建矩形
            rect = patches.Rectangle(
                (start, y_pos - 0.4), duration, 0.8,
                linewidth=1, edgecolor='black',
                facecolor=f'C{int(job[-1])-1}', alpha=0.7
            )
            ax.add_patch(rect)
            
            # 添加文本标签
            plt.text(
                start + duration/2, y_pos,
                f'{job}-{op["op"]+1}',
                ha='center', va='center', color='black'
            )
    
    # 设置图表属性
    ax.set_yticks(y_ticks)
    ax.set_yticklabels(y_labels)
    ax.set_xlabel('Time')
    ax.set_title('Job Shop Schedule Gantt Chart')
    ax.grid(True, which='both', axis='x', linestyle='--', alpha=0.7)
    
    # 自动调整x轴范围
    max_time = max(op['end'] for machine in schedule.values() for op in machine)
    ax.set_xlim(0, max_time * 1.05)
    
    plt.tight_layout()
    plt.show()

4.2 完整流程示例

让我们将上述组件组合起来，完成一个完整的JSSP解决方案：

# 定义问题实例
jobs_data = {
    'Job1': [('Machine1', 15), ('Machine2', 25), ('Machine3', 18)],
    'Job2': [('Machine2', 10), ('Machine3', 20), ('Machine1', 7)],
    'Job3': [('Machine3', 8), ('Machine1', 12), ('Machine2', 10)]
}

# 生成染色体
chromosome = generate_chromosome(jobs_data)
print("Generated Chromosome:", chromosome)

# 解码染色体
schedule = decode_chromosome(chromosome, jobs_data)

# 计算最大完成时间
makespan = max(op['end'] for machine in schedule.values() for op in machine)
print("Schedule Makespan:", makespan)

# 绘制甘特图
plot_gantt(schedule)

运行这段代码将生成一个随机的调度方案，并显示对应的甘特图。甘特图中：

• 每行代表一台机器
• 每个彩色矩形代表一个工序
• 矩形长度表示工序持续时间
• 矩形上的标签显示工件编号和工序编号

5. 优化与扩展

虽然我们已经实现了基本的JSSP解决方案，但仍有改进空间。以下是几个可能的优化方向：

5.1 遗传算法优化

我们可以使用遗传算法来寻找更优的染色体：

def genetic_algorithm(jobs_data, pop_size=50, generations=100):
    population = [generate_chromosome(jobs_data) for _ in range(pop_size)]
    
    for _ in range(generations):
        # 评估适应度（makespan越小越好）
        fitness = []
        for chrom in population:
            schedule = decode_chromosome(chrom, jobs_data)
            makespan = max(op['end'] for machine in schedule.values() for op in machine)
            fitness.append(1/makespan)  # 倒数作为适应度
        
        # 选择（轮盘赌选择）
        selected = random.choices(
            population, weights=fitness, k=pop_size
        )
        
        # 交叉（顺序交叉）
        new_population = []
        for i in range(0, pop_size, 2):
            parent1, parent2 = selected[i], selected[i+1]
            
            # 选择交叉点
            size = len(parent1)
            cx1 = random.randint(0, size-1)
            cx2 = random.randint(cx1, size-1)
            
            # 创建子代
            child1 = parent1[cx1:cx2] + [g for g in parent2 if g not in parent1[cx1:cx2]]
            child2 = parent2[cx1:cx2] + [g for g in parent1 if g not in parent2[cx1:cx2]]
            
            new_population.extend([child1, child2])
        
        # 变异（交换两个随机基因）
        for i in range(pop_size):
            if random.random() < 0.1:  # 10%变异概率
                idx1, idx2 = random.sample(range(len(new_population[i])), 2)
                new_population[i][idx1], new_population[i][idx2] = new_population[i][idx2], new_population[i][idx1]
        
        population = new_population
    
    # 返回最佳染色体
    best_chrom = min(population, key=lambda x: max(op['end'] for machine in decode_chromosome(x, jobs_data).values() for op in machine))
    return best_chrom

5.2 多目标优化

除了最小化makespan，我们还可以考虑其他目标，如：

• 机器负载均衡
• 工件等待时间
• 交货期满足率

这需要修改适应度函数，考虑多个目标的加权组合。

5.3 实际应用扩展

在实际应用中，我们可能需要考虑更多因素：

• 机器故障和维修
• 工人技能限制
• 物料供应约束
• 紧急插单处理

这些都可以通过扩展我们的基本模型来实现。