ИЛИ Tools cp sat гибкая мастерская с гибкой продолжительностью оптимального решения не найдено ⇐ Python

[Anonymous]

Я пытаюсь объединить гибкую оптимизацию цеха с гибкой продолжительностью на основе машинных календарей, как здесь: https://groups.google.com/g/or-tools-di ... 2vO5K1BgAJ
Мой код работает большую часть времени, но в определенных обстоятельствах я не понимаю, почему не найдено оптимальное решение.
В приведенном ниже примере я у меня есть две машины с двумя плановыми перерывами, которые должны продлить продолжительность работы. Есть 4 задания с одной задачей и двумя альтернативами, каждая из которых имеет одинаковую продолжительность. Каждая задача длиннее, чем период до первого перерыва, поэтому я ожидаю получить два задания, начиная с 0 на каждом компьютере и продлеваясь на продолжительность перерывов (2).
Тем не менее, алгоритм возвращает только одно задание, начинающееся с 0. На другой машине первое задание запускается после запланированного перерыва. Это дает неоптимальное решение. Интересно, что если я увеличу продолжительность первой альтернативы первого задания с 8 до 9, алгоритм вернет два задания, начиная с 0, на обеих машинах и, следовательно, найдет оптимальное решение.
Вот мой код очень похоже на доступные в Интернете примеры. Что я пропустил/перепутал? Заранее большое спасибо за вашу поддержку!
#!/usr/bin/env python3
# Copyright 2010-2024 Google LLC
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

"""Solves a flexible jobshop problems with the CP-SAT solver.

A jobshop is a standard scheduling problem when you must sequence a
series of task_types on a set of machines. Each job contains one task_type per
machine. The order of execution and the length of each job on each
machine is task_type dependent.

The objective is to minimize the maximum completion time of all
jobs. This is called the makespan.
"""

# overloaded sum() clashes with pytype.

import collections

from ortools.sat.python import cp_model

class SolutionPrinter(cp_model.CpSolverSolutionCallback):
"""Print intermediate solutions."""

def __init__(self):
cp_model.CpSolverSolutionCallback.__init__(self)
self.__solution_count = 0

def on_solution_callback(self):
"""Called at each new solution."""
strv = "Solution %i, time = %f s, objective = %i" % (self.__solution_count, self.wall_time, self.objective_value)
print(
strv
)

def flexible_jobshop():
"""solve a small flexible jobshop problem."""
# Data part.

# get jobs (duration, machine)
jobs = [[[(8, 0), #change this duration to 9 for the optimal solution
(7, 1)]],
[[(7, 0), (9, 1)]],
[[(7, 0), (8, 1)]],
[[(7, 0), (8, 1)]]]

num_machines = 2
all_machines = range(num_machines)

print("Jobs: ")
print(jobs)

downtimes_per_machine = {
'0': [[6, 7]],
'1': [[6, 7]]
}
print("Downtimes per Machine: ")
print(downtimes_per_machine)

num_jobs = len(jobs)
all_jobs = range(num_jobs)

# get max downtime duration
max_downtime_duration = 0
for machine in downtimes_per_machine.keys():
for downtime in downtimes_per_machine[machine]:
max_downtime_duration = max(max_downtime_duration, downtime[1]-downtime[0]+1)

# Model the flexible jobshop problem.
model = cp_model.CpModel()

horizon = 82
print("Horizon = %i" % horizon)

# Global storage of variables.
intervals_per_resources = collections.defaultdict(list)
presences_per_machines = collections.defaultdict(list)
starts_per_machines = collections.defaultdict(list)
ends_per_machines = collections.defaultdict(list)

starts = {} # indexed by (job_id, task_id).
durations = {} # indexed by (job_id, task_id).
presences = {} # indexed by (job_id, task_id, alt_id).
job_ends = []

# Scan the jobs and create the relevant variables and intervals.
for job_id in all_jobs:
job = jobs[job_id]
num_tasks = len(job)
task_previous_end = None
for task_id in range(num_tasks):
task = job[task_id]

min_duration = task[0][0]
max_duration = task[0][0]

num_alternatives = len(task)
all_alternatives = range(num_alternatives)

for alt_id in range(1, num_alternatives):
alt_duration = task[alt_id][0]
min_duration = min(min_duration, alt_duration)
max_duration = max(max_duration, alt_duration)

# Create main interval for the task.
suffix_name = "_j%i_t%i" % (job_id, task_id)
task_start = model.new_int_var(0, horizon, "start" + suffix_name)
task_duration = model.new_int_var(
min_duration, max_duration, "duration" + suffix_name
)
task_end = model.new_int_var(0, horizon, "end" + suffix_name)
interval = model.new_interval_var(
task_start, task_duration, task_end, "interval" + suffix_name
)

# Store the start for the solution.
starts[(job_id, task_id)] = task_start

# Add precedence with previous task in the same job.
if task_previous_end is not None:
model.add(task_start >= task_previous_end)
task_previous_end = task_end

# Create alternative intervals.
l_presences = []
for alt_id in all_alternatives:
l_machine = task[alt_id][1]
alt_suffix = "_j%i_t%i_a%i" % (job_id, task_id, alt_id)
l_presence = model.new_bool_var("presence" + alt_suffix)
duration = task[alt_id][0]
l_start = model.new_int_var_from_domain(
cp_model.Domain.from_intervals([(0,5), (8,74)]), "start" + alt_suffix) # no start allowed during planned downtime

l_end = model.new_int_var(0, horizon, "end" + alt_suffix)
l_duration = model.new_int_var(duration, duration+2, "duration" + alt_suffix)
l_interval = model.new_optional_interval_var(l_start, l_duration, l_end, l_presence, "interval" + alt_suffix)

# handle planned downtimes extending a task's duration
l_across = model.NewBoolVar("across " + alt_suffix)
l_not_across = model.NewBoolVar("not across " + alt_suffix)
model.add_linear_constraint(l_start, 0, 5).OnlyEnforceIf(l_across)
model.add_linear_constraint(l_start, 8, 74).OnlyEnforceIf(l_not_across)
model.Add(l_across + l_not_across == 1)
model.Add(l_duration == duration + 2).OnlyEnforceIf(l_across)
model.Add(l_duration == duration).OnlyEnforceIf(l_not_across)

l_presences.append(l_presence)

# Link the primary/global variables with the local ones.
model.add(task_start == l_start).only_enforce_if(l_presence)
model.add(task_duration == l_duration).only_enforce_if(l_presence)
model.add(task_end == l_end).only_enforce_if(l_presence)

# Add the local interval to the right machine.
l_machine = task[alt_id][1]
intervals_per_resources[l_machine].append(l_interval)
starts_per_machines[l_machine].append(l_start)
ends_per_machines[l_machine].append(l_end)
presences_per_machines[l_machine].append(l_presence)

# Store the presences for the solution.
presences[(job_id, task_id, alt_id)] = l_presence

# Select exactly one presence variable.
model.add_exactly_one(l_presences)
durations[(job_id, task_id)] = task_duration

job_ends.append(task_end)

# Create machines constraints.
for machine_id in all_machines:
intervals = intervals_per_resources[machine_id]
if len(intervals) > 1:
model.add_no_overlap(intervals)

# Makespan objective
makespan = model.new_int_var(0, horizon, "makespan")
model.add_max_equality(makespan, job_ends)

model.minimize(makespan)

# Solve model.
solver = cp_model.CpSolver()
solver.parameters.max_time_in_seconds = 12
solution_printer = SolutionPrinter()
status = solver.solve(model, solution_printer)

return_ob = []

# Print final solution.
print("solve status: %s" % solver.status_name(status))
for job_id in all_jobs:
print("Job %i:" % job_id)
return_ob.append({"jobId" : job_id, "tasks" : []})
for task_id in range(len(jobs[job_id])):

start_value = solver.value(starts[(job_id, task_id)])
machine = -1
task_duration = solver.value(durations[(job_id, task_id)])
selected = -1
act_task_id = -1
act_order_id = -1
for alt_id in range(len(jobs[job_id][task_id])):
if solver.value(presences[(job_id, task_id, alt_id)]):
#task_duration = jobs[job_id][task_id][alt_id][0]
machine = jobs[job_id][task_id][alt_id][1]
selected = alt_id
print(
" task_%i_%i starts at %i (alt %i, machine %i, duration %i)"
% (job_id, task_id, start_value, selected, machine, task_duration)
)
return_ob[job_id]["tasks"].append({"orderId" : act_order_id, "taskId" : act_task_id, "start": start_value, "selected" : selected, "machineId" : machine, "duration" : task_duration})

print("solve status: %s" % solver.status_name(status))
print("Optimal objective value: %i" % solver.objective_value)
print("Statistics")
print(" - conflicts : %i" % solver.num_conflicts)
print(" - branches : %i" % solver.num_branches)
print(" - wall time : %f s" % solver.wall_time)

return return_ob

flexible_jobshop()


Подробнее здесь: https://stackoverflow.com/questions/790 ... ion-not-fo