数据¶
数据文件的格式如下：
- 第一行：手术室数量、护士数量、手术数量
- 第二行：每台手术的最早开始时间（单位：小时）
- 第三行：每台手术的最晚结束时间（单位：小时）
- 第四行：每台手术的持续时间（单位：分钟）
- 第五行：每台手术所需的护士数量
- 第六行：每位护士班次的最早开始时间（单位：小时）
- 第七行：每位护士班次的最晚结束时间（单位：小时）
- 第八行：班次的最大持续时长（单位：小时）
- 接下来的每一行给出一台手术与各手术室的不兼容标记，1 表示不兼容，0 表示兼容
建模思路¶
每台手术使用一个具有开始、结束和固定持续时间的 interval；每个手术室和每位护士分别使用一个 list 表示其手术顺序。手术室 lists 通过 partition 约束使每台手术恰好分配到一个兼容手术室，相邻 interval 的顺序约束保证手术室内不重叠。
护士 lists 可以共同包含同一手术。模型通过 contains 汇总参与每台手术的护士数量，并对每位护士的首台和末台手术约束班次起止时间及最大持续时长；护士序列中的相邻手术同样不得重叠。目标是最小化所有手术结束时间的最大值。
Python 实现¶
from pathlib import Path
from optagent import OptModel, solve
def read_instance(filename):
lines = Path(filename).read_text(encoding="utf-8").splitlines()
num_rooms, num_nurses, num_surgeries = map(int, lines[0].split())
min_start = [int(value) * 60 for value in lines[1].split()]
max_end = [int(value) * 60 for value in lines[2].split()]
duration = [int(value) for value in lines[3].split()]
needed_nurses = [int(value) for value in lines[4].split()]
shift_earliest_start = [int(value) * 60 for value in lines[5].split()]
shift_latest_end = [int(value) * 60 for value in lines[6].split()]
max_shift_duration = int(lines[7].split()[0]) * 60
incompatible_rooms = [
[int(value) for value in lines[8 + surgery].split()]
for surgery in range(num_surgeries)
]
return (
num_rooms,
num_nurses,
num_surgeries,
min_start,
max_end,
needed_nurses,
shift_earliest_start,
shift_latest_end,
max_shift_duration,
incompatible_rooms,
duration,
)
def main(input_file, output_file=None, time_limit=20):
(
num_rooms,
num_nurses,
num_surgeries,
min_start,
max_end,
needed_nurses,
shift_earliest_start,
shift_latest_end,
max_shift_duration,
incompatible_rooms,
duration,
) = read_instance(input_file)
model = OptModel()
# Each surgery is assigned to exactly one compatible room.
surgery_order = [
model.list(num_surgeries)
for room in range(num_rooms)
]
rooms = model.array(surgery_order)
model.constraint(model.partition(rooms))
for surgery in range(num_surgeries):
for room in range(num_rooms):
if incompatible_rooms[surgery][room]:
model.constraint(
model.contains(surgery_order[room], surgery) == 0,
)
selected_room = [model.find(rooms, surgery) for surgery in range(num_surgeries)]
surgeries = [
model.interval(
min_start[surgery],
max_end[surgery],
)
for surgery in range(num_surgeries)
]
for surgery in range(num_surgeries):
model.constraint(
surgeries[surgery].length() == duration[surgery],
)
surgery_array = model.array(surgeries)
def make_no_overlap_lambda(sequence):
return model.lambda_function(
lambda position: surgery_array[sequence[position]]
< surgery_array[sequence[position + 1]]
)
# A room can process only one surgery at a time.
for room, sequence in enumerate(surgery_order):
adjacent_positions = model.range(0, sequence.count() - 1)
model.constraint(
model.and_(
adjacent_positions, make_no_overlap_lambda(sequence)
),
)
nurse_order = [
model.list(num_surgeries)
for nurse in range(num_nurses)
]
for nurse, sequence in enumerate(nurse_order):
count = sequence.count()
first_surgery_start = model.iif(
count > 0,
surgery_array[sequence[0]].start,
shift_earliest_start[nurse],
)
last_surgery_end = model.iif(
count > 0,
surgery_array[sequence[count - 1]].end,
shift_earliest_start[nurse],
)
model.constraint(
first_surgery_start >= shift_earliest_start[nurse],
)
model.constraint(
last_surgery_end <= shift_latest_end[nurse],
)
model.constraint(
last_surgery_end - first_surgery_start <= max_shift_duration,
)
adjacent_positions = model.range(0, count - 1)
model.constraint(
model.and_(
adjacent_positions, make_no_overlap_lambda(sequence)
),
)
nurse_order_array = model.array(nurse_order)
def make_nurse_assignment_lambda(surgery):
return model.lambda_function(
lambda nurse: model.contains(nurse_order_array[nurse], surgery)
)
for surgery in range(num_surgeries):
assigned_nurses = model.sum(
model.range(0, num_nurses),
make_nurse_assignment_lambda(surgery),
)
model.constraint(
assigned_nurses >= needed_nurses[surgery],
)
makespan = model.max([surgery.end for surgery in surgeries])
model.minimize(makespan)
solution = solve(model, time_limit_s=float(time_limit))
if not solution.feasible:
print(f"No feasible surgery schedule found; Status = {solution.status}")
return solution
nurses_by_surgery = [[] for _ in range(num_surgeries)]
for nurse, sequence in enumerate(nurse_order):
for surgery in sequence.value:
nurses_by_surgery[int(surgery)].append(nurse)
output_lines = []
for surgery in range(num_surgeries):
interval = surgeries[surgery].value
output_lines.append(
f"{surgery}\t\t{int(selected_room[surgery].value)}"
f"\t{interval.start()}\t{interval.end()}"
f"\t{nurses_by_surgery[surgery]}"
)
print(
f"Makespan = {makespan.value}; Status = {solution.status}\n"
+ "\n".join(output_lines)
)
if output_file is not None:
Path(output_file).write_text(
"\n".join(output_lines) + "\n", encoding="utf-8"
)
return solution
INSTANCE_DIR = Path.cwd() / "instances"
print("Instances:", INSTANCE_DIR)
solution = main(
INSTANCE_DIR / "instancesurgery.txt",
time_limit=1,
)