基于Python的作业车间调度变体开发：机器压缩及周调度需求

问题分析 #

这本质上是带时间窗的多资源调度问题，核心是将固定时段的任务（支持周内重复规则）无重叠地分配到最少机器上，最大化任务容纳量。和经典作业车间调度不同，这里任务没有工序依赖，只有固定时间约束，还要处理周内差异化的重复逻辑。

核心思路 #

• 任务建模：为每个任务定义固定时段（每天的起始/结束分钟）和周运行规则（如周一、周四运行）。
• 时间冲突检测：判断两个任务在同一机器上是否存在时间重叠（考虑周运行规则的交集）。
• 贪心调度算法：按任务时长从长到短排序，依次将任务分配到第一个能容纳它的机器，若没有则新增机器。这种方法计算效率高，适合50个任务的规模，且能得到接近最优的机器数量。
• 甘特图可视化：用matplotlib生成压缩后的甘特图，直观展示分配结果。

代码实现 #

1. 任务类与冲突检测函数 #

from dataclasses import dataclass
from typing import List, Set
import matplotlib.pyplot as plt
import matplotlib.dates as mdates
from datetime import datetime, timedelta

# 定义任务数据类
@dataclass
class Task:
    task_id: str
    start_min: int  # 每日开始分钟（0-1439）
    end_min: int    # 每日结束分钟（0-1439）
    run_days: Set[str]  # 运行的星期几，如{"Monday", "Thursday"}

# 判断两个任务是否存在时间冲突（考虑周运行规则）
def has_conflict(task1: Task, task2: Task) -> bool:
    # 先检查是否有共同的运行日期
    common_days = task1.run_days & task2.run_days
    if not common_days:
        return False
    # 再检查时段是否重叠：task1的时段和task2的时段有交集
    return not (task1.end_min <= task2.start_min or task2.end_min <= task1.start_min)

2. 调度核心逻辑 #

def schedule_tasks(tasks: List[Task]) -> List[List[Task]]:
    # 贪心策略：按任务时长从长到短排序，优先安排长任务（减少碎片化）
    sorted_tasks = sorted(tasks, key=lambda t: (t.end_min - t.start_min), reverse=True)
    
    machines = []  # 每个元素是一台机器上的任务列表
    
    for task in sorted_tasks:
        assigned = False
        # 遍历现有机器，尝试分配
        for machine in machines:
            # 检查该机器上所有任务是否都和当前任务无冲突
            if all(not has_conflict(task, existing_task) for existing_task in machine):
                machine.append(task)
                assigned = True
                break
        # 没有可用机器，新增一台
        if not assigned:
            machines.append([task])
    return machines

3. 周度任务扩展与甘特图生成 #

def generate_gantt_chart(machines: List[List[Task]], week_start_date: str = "2024-01-01"):
    # 定义星期映射，用于将字符串转换为日期偏移
    day_map = {
        "Monday": 0,
        "Tuesday": 1,
        "Wednesday": 2,
        "Thursday": 3,
        "Friday": 4,
        "Saturday": 5,
        "Sunday": 6
    }
    start_date = datetime.strptime(week_start_date, "%Y-%m-%d")
    
    fig, ax = plt.subplots(figsize=(12, len(machines)*0.8))
    
    for machine_idx, tasks in enumerate(machines):
        for task in tasks:
            for day in task.run_days:
                # 计算任务的实际开始和结束datetime
                day_offset = day_map[day]
                task_start = start_date + timedelta(days=day_offset, minutes=task.start_min)
                task_end = start_date + timedelta(days=day_offset, minutes=task.end_min)
                
                # 绘制甘特图条形
                ax.barh(machine_idx + 1, task_end - task_start, left=task_start, height=0.6, label=f"Task {task.task_id}")
    
    # 设置坐标轴格式
    ax.set_yticks(range(1, len(machines)+1))
    ax.set_yticklabels([f"Machine {i+1}" for i in range(len(machines))])
    ax.xaxis.set_major_formatter(mdates.DateFormatter("%a %H:%M"))
    plt.xticks(rotation=45)
    plt.xlabel("Time")
    plt.ylabel("Machines")
    plt.title("Compressed Job Shop Schedule (Weekly)")
    # 避免重复标签
    handles, labels = plt.gca().get_legend_handles_labels()
    by_label = dict(zip(labels, handles))
    plt.legend(by_label.values(), by_label.keys(), bbox_to_anchor=(1.05, 1), loc='upper left')
    plt.tight_layout()
    plt.show()

4. 测试示例 #

if __name__ == "__main__":
    # 构造测试任务（模拟50个任务的场景，这里简化为10个示例）
    tasks = [
        Task("T1", 300, 480, {"Monday", "Thursday"}),   # 5:00-8:00
        Task("T2", 400, 520, {"Monday", "Wednesday"}),  # 6:40-8:40
        Task("T3", 600, 720, {"Tuesday", "Friday"}),    # 10:00-12:00
        Task("T4", 300, 480, {"Tuesday", "Friday"}),    # 5:00-8:00
        Task("T5", 700, 840, {"Monday", "Thursday"}),   # 11:40-14:00
        Task("T6", 400, 520, {"Thursday", "Friday"}),   # 6:40-8:40
        Task("T7", 900, 1020, {"Wednesday", "Saturday"}),# 15:00-17:00
        Task("T8", 300, 480, {"Wednesday", "Saturday"}),# 5:00-8:00
        Task("T9", 1000, 1140, {"Monday", "Tuesday"}),  # 16:40-19:00
        Task("T10", 1200, 1340, {"Thursday", "Sunday"}) # 20:00-22:20
        # 可继续添加至50个任务
    ]
    
    # 执行调度
    machines = schedule_tasks(tasks)
    print(f"压缩后机器数量: {len(machines)}")
    
    # 打印每台机器的任务
    for idx, machine in enumerate(machines):
        print(f"\nMachine {idx+1} 任务列表:")
        for task in machine:
            print(f"- {task.task_id}: {task.start_min//60}:{task.start_min%60:02d} - {task.end_min//60}:{task.end_min%60:02d}, 运行日: {', '.join(task.run_days)}")
    
    # 生成甘特图
    generate_gantt_chart(machines)

关键说明 #

• 贪心算法的合理性：对于固定时段的任务调度，按时长从长到短排序的贪心策略能有效减少机器数量，因为长任务更难找到空闲时段，优先安排可避免后期因碎片化无法容纳。
• 周度规则处理：通过任务的run_days集合，仅在共同运行日期上检测时间冲突，确保跨天任务的正确调度。
• 扩展性：若需要更优的机器数量，可将贪心算法替换为整数规划（如用pulp库），但计算复杂度会显著提升，50个任务的场景下贪心已足够高效。

内容的提问来源于stack exchange，提问作者Viz