最小化所有任务的总拖期(Total Tardiness),其数学表达式为:
其中:
- ( C_j ):任务 ( J_j ) 的完成时间。
- ( d_j ):任务 ( J_j ) 的交付期限(Due Date)。
其中:
- ( Y_{ij} ):二进制变量,表示任务 ( J_j ) 是否分配给机器 ( M_i ) (1为分配,0为不分配)。
其中:
- ( X_{ijg} ):二进制变量,表示任务 ( J_j ) 是否排在任务 ( J_g ) 之前。
确保任务只能在其释放时间后开始加工,且考虑到家族设置时间的影响:
其中:
- ( r_j ):任务的释放时间。
- ( S_{jg} ):家族设置时间(如果任务 ( J_j ) 和 ( J_g ) 来自不同家族,则有设置时间)。
- ( X_{ijg} ):二进制变量,表示任务 ( J_j ) 是否排在任务 ( J_g ) 之前。
- ( Y_{ij} ):二进制变量,表示任务 ( J_j ) 是否分配给机器 ( M_i )。
- ( C_j ):任务 ( J_j ) 的完成时间。
通过上述目标函数和约束条件,你可以使用 Gurobi 来实现并行机器调度问题的求解。
# PMSP调度问题
"""
目标函数:
- 最小化总延迟
"""
#
"""
约束条件:
- 每个作业只能分配给一个机器
- 每个机器上有且仅有一个前驱作业
- 每个机器上有且仅有一个后继作业
- 每台机器上最多只能有一个作业作为第一个作业
- 作业只能在到达时间之后开始处理
- 作业之间的重叠和设置时间的约束,我这里默认任何一个作业到另一个作业的转换设置时间都为1
- 每台机器上的作业的完成时间初始值都是0
"""
#
"""
决策变量:
- 作业Jj完成的时间
- 如果作业Jj是作业Jg在机器Mi上的前驱,则为1,否则为0
- 如果作业Jj被分配给机器Mi,则为1,否则为0
"""
from dataclasses import dataclass
from gurobipy import GRB
import gurobipy
import numpy
V = 0xffffff
@dataclass
class Job:
jobId : int
processTime : int
arrivalTime : int
endTime : int
def __post_init__(self):
self.name = "Job%d" % self.jobId
@dataclass
class Machine:
machId : int
speed : int
def __post_init__(self):
self.name = "Machine%d" % self.machId
# 机器
machines = [
Machine(0, 1),
Machine(1, 2),
Machine(2, 1),
Machine(3, 3),
Machine(4, 2)
]
# 工作
jobs = [
Job(0, 3, 0, 7),
Job(1, 2, 1, 4),
Job(2, 4, 2, 20),
Job(3, 1, 3, 15),
Job(4, 3, 5, 10),
Job(5, 2, 5, 15),
Job(6, 4, 7, 18),
Job(7, 3, 10, 25),
Job(8, 2, 11, 25),
Job(9, 1, 13, 26),
Job(10, 5, 14, 30),
Job(11, 3, 15, 35),
Job(12, 2, 16, 40)
]
num_jobs = len(jobs)
num_machines = len(machines)
# 创建模型
model = gurobipy.Model("Model")
# 决策变量
# 如果作业Jj是作业Jg在机器Mi上的前驱,则为1,否则为0
# X = model.addVars(num_machines, num_jobs, num_jobs, vtype=GRB.BINARY, name='X')
X = {}
for i in range(num_machines):
for j in range(num_jobs):
for g in range(num_jobs):
if j != g:
X[i, j, g] = model.addVar(vtype=GRB.BINARY)
# 如果作业Jj被分配给机器Mi,则为1,否则为0
Y = model.addVars(num_machines, num_jobs, vtype=GRB.BINARY, name='Y')
# 作业Jj完成的时间
C = model.addVars(num_jobs, vtype=GRB.CONTINUOUS, name='C')
# S是Jg紧接着转向Jj作业之后设置的时间,设置了1
S = numpy.full([num_jobs, num_jobs], 1)
# 传递变量
# for i in range(num_jobs):
# model.addConstr(C[i] == 0)
# 1.每个作业只能分配给一个机器
for j in range(num_jobs):
model.addConstr(
gurobipy.quicksum(Y[i, j] for i in range(num_machines)) == 1
)
# 2.每个机器上有且仅有一个前驱作业
for i in range(num_machines):
for g in range(num_jobs):
model.addConstr(
gurobipy.quicksum(
X[i, j, g] for j in range(num_jobs) if j != g
) == Y[i, g]
)
# 3.每个机器上有且仅有一个后继作业
for i in range(num_machines):
for j in range(num_jobs):
model.addConstr(
gurobipy.quicksum(
X[i, j, g] for g in range(num_jobs) if g != j
) == Y[i, j]
)
# 4.每台机器上最多只能有一个作业作为第一个作业
for i in range(num_machines):
model.addConstr(
gurobipy.quicksum(
X[i, 0, j] for j in range(num_jobs) if j != 0
) <= 1
)
# 5.作业只能在到达时间之后开始处理
for j in range(num_jobs):
for i in range(num_machines):
model.addConstr(
C[j] + V * (1 - Y[i, j]) >= jobs[j].arrivalTime + jobs[j].processTime / machines[i].speed
)
# 6.作业之间的重叠和设置时间的约束,我这里默认任何一个作业到另一个作业的转换设置时间都为1
for i in range(num_machines):
for g in range(num_jobs):
for j in range(num_jobs):
if j != g:
model.addConstr(
C[g] - C[j] + V * (1 - X[i, j, g]) >= S[g][j] + jobs[g].processTime / machines[i].speed
)
# 7.作业的完成时间也需要更新,这应该是最后一个约束公式想表达的意思
for i in range(num_machines):
for j in range(num_jobs):
for g in range(num_jobs):
if j != g:
model.addConstr(
C[g] >= C[j] + jobs[j].processTime / machines[i].speed - V * (1 - X[i, j, g])
)
# 目标函数:最小化总延迟
Total_times = []
# 遍历每个作业,计算完成时间与截止时间之间的延迟
for j in range(num_jobs):
diff_var = model.addVar(vtype=GRB.CONTINUOUS) # 常量不能直接和变量参与计算,通过变换到变量约束
model.addConstr(diff_var == C[j] - jobs[j].endTime)
delay_var = model.addVar(vtype=GRB.CONTINUOUS)
model.addGenConstrMax(delay_var, [0, diff_var])
Total_times.append(delay_var)
model.setObjective(gurobipy.quicksum(Total_times), gurobipy.GRB.MINIMIZE)
model.update()
model.optimize()
if model.status == GRB.OPTIMAL:
print("Get Optimal Solution:")
for j in range(num_jobs):
for i in range(num_machines):
if Y[i, j].x > 0: # 判断作业是否分配给机器
print(f"Job {j} assigned to Machine {i}, Completion Time: {C[j].x}")
else:
print("No optimal solution found.")
# Get Optimal Solution:
# Job 0 assigned to Machine 4, Completion Time: 7.0
# Job 1 assigned to Machine 0, Completion Time: 4.0
# Job 2 assigned to Machine 3, Completion Time: 20.0
# Job 3 assigned to Machine 1, Completion Time: 15.0
# Job 4 assigned to Machine 0, Completion Time: 10.0
# Job 5 assigned to Machine 0, Completion Time: 15.0
# Job 5 assigned to Machine 1, Completion Time: 15.0
# Job 5 assigned to Machine 2, Completion Time: 15.0
# Job 5 assigned to Machine 3, Completion Time: 15.0
# Job 5 assigned to Machine 4, Completion Time: 15.0
# Job 6 assigned to Machine 3, Completion Time: 18.0
# Job 7 assigned to Machine 2, Completion Time: 25.0
# Job 8 assigned to Machine 1, Completion Time: 25.0
# Job 9 assigned to Machine 4, Completion Time: 26.0
# Job 10 assigned to Machine 3, Completion Time: 30.0
# Job 11 assigned to Machine 2, Completion Time: 35.0
# Job 12 assigned to Machine 4, Completion Time: 40.0