Add setup time to tasks in job shop scheduling

[uprox]

Hi all.
I wrote (copy, cut, paste, merge from many examples) an algorithm for flexible job shop scheduling, where I could have more than 1 alternative for each task. It works as expected, but now I'm trying to add a constraint to get a fixed setup time in advance an all tasks. I'm not figuring out how to achieve it.
I need just to have a setup time of 4 hours before each task. It not depends on any other variable, condition, constraint... just fixed 4 hours before each task start.

I already checked examples from this repository (job shop, job shop with distance, ...) but any of them helped me.

Can you suggest me what kind of constraint should I add to the model in order to get it, please?

Thank you!

import collections
from ortools.sat.python import cp_model
import csv
import numpy
import pandas as pd
import random
import plotly.express as px
from datetime import datetime, timedelta

demand = []
molds = []
jobs_data = []

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."""
        print('Solution %i, time = %f s, objective = %i' %
              (self.__solution_count, self.WallTime(), self.ObjectiveValue()))
        self.__solution_count += 1
        
def distance_between_jobs(x: int, y: int) -> int:
    """Returns the distance between tasks of job x and tasks of job y."""
    return abs(x - y)

def flexible_jobshop(jobs, num_machines, num_molds):
    num_jobs = len(jobs)
    all_jobs = range(num_jobs)

    all_machines = range(num_machines)
    all_molds = range(num_molds)

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

    horizon = 0
    for job in jobs:
        for task in job:
            max_task_duration = 0
            for alternative in task:
                max_task_duration = max(max_task_duration, alternative[0])
            horizon += max_task_duration

    print('Horizon = %i' % horizon)

    task_type = collections.namedtuple("task_type", "start end interval")

    # Global storage of variables.
    intervals_per_resources = collections.defaultdict(list)
    intervals_per_molds = collections.defaultdict(list)
    starts = {}  # indexed by (job_id, task_id).
    presences = {}  # indexed by (job_id, task_id, alt_id).
    job_ends = []  # indexed by job_id

    # Scan the jobs and create the relevant variables and intervals.
    for job_id in all_jobs:
        job = jobs[job_id]
        num_tasks = len(job)
        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)
            start = model.NewIntVar(0, horizon, 'start' + suffix_name)
            duration = model.NewIntVar(min_duration, max_duration,
                                       'duration' + suffix_name)
            end = model.NewIntVar(0, horizon, 'end' + suffix_name)
            interval = model.NewIntervalVar(start, duration, end,
                                            'interval' + suffix_name)

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

            # Add precedence with previous task in the same job.
            if previous_end is not None:
                model.Add(start >= previous_end)
            previous_end = end

            # Create alternative intervals.
            if num_alternatives > 1:
                l_presences = []
                for alt_id in all_alternatives:
                    alt_suffix = '_j%i_t%i_a%i' % (job_id, task_id, alt_id)
                    l_presence = model.NewBoolVar('presence' + alt_suffix)
                    l_start = model.NewIntVar(0, horizon, 'start' + alt_suffix)
                    l_duration = task[alt_id][0]
                    l_end = model.NewIntVar(0, horizon, 'end' + alt_suffix)
                    l_interval = model.NewOptionalIntervalVar(
                        l_start, l_duration, l_end, l_presence,
                        'interval' + alt_suffix)
                    l_presences.append(l_presence)

                    # Link the primary/global variables with the local ones.
                    model.Add(start == l_start).OnlyEnforceIf(l_presence)
                    model.Add(duration == l_duration).OnlyEnforceIf(l_presence)
                    model.Add(end == l_end).OnlyEnforceIf(l_presence)

                    # Add the local interval to the right machine.
                    intervals_per_resources[task[alt_id][1]].append(l_interval)
                    intervals_per_molds[task[alt_id][2]].append(l_interval)

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

                # Select exactly one presence variable.
                model.AddExactlyOne(l_presences)
            else:
                intervals_per_resources[task[0][1]].append(interval)
                intervals_per_molds[task[0][2]].append(interval)
                presences[(job_id, task_id, 0)] = model.NewConstant(1)

        job_ends.append(previous_end)

    # Create machines constraints.
    for machine_id in all_machines:
        intervals = intervals_per_resources[machine_id]
        if len(intervals) > 1:
            model.AddNoOverlap(intervals)
    # Create molds constraints.
    for mold_id in all_molds:
        intervals_molds = intervals_per_molds[mold_id]
        if len(intervals_molds) > 1:
            model.AddNoOverlap(intervals_molds)

    # Makespan objective
    makespan = model.NewIntVar(0, horizon, 'makespan')
    model.AddMaxEquality(makespan, job_ends)
    model.Minimize(makespan)

    # Solve model.
    solver = cp_model.CpSolver()
    solution_printer = SolutionPrinter()
    status = solver.Solve(model, solution_printer)

    print(solver.StatusName(status))

    outputlist = []
    # Print final solution.
    for job_id in all_jobs:
        print('Job %i:' % job_id)
        for task_id in range(len(jobs[job_id])):
            start_value = solver.Value(starts[(job_id, task_id)])
            machine = -1
            mold = -1
            item = ''
            duration = -1
            selected = -1
            for alt_id in range(len(jobs[job_id][task_id])):
                if solver.Value(presences[(job_id, task_id, alt_id)]):
                    duration = jobs[job_id][task_id][alt_id][0]
                    machine = jobs[job_id][task_id][alt_id][1]
                    mold = jobs[job_id][task_id][alt_id][2]
                    item = jobs[job_id][task_id][alt_id][3]
                    selected = alt_id
            print(
                '  task_%i_%i starts at %i (alt %i, machine %i, duration %i)' %
                (job_id, task_id, start_value, selected, machine, duration))
            outputlist.append([job_id, task_id, alt_id, machine, start_value, start_value + duration, mold, item])

    print(outputlist)
    print('Solve status: %s' % solver.StatusName(status))
    print('Optimal objective value: %i' % solver.ObjectiveValue())
    print('Statistics')
    print('  - conflicts : %i' % solver.NumConflicts())
    print('  - branches  : %i' % solver.NumBranches())
    print('  - wall time : %f s' % solver.WallTime())
    return(outputlist)

def importDemand():
    demand = []
    with open('demand.txt', newline = '') as demand_file:
        demand_reader = csv.reader(demand_file, delimiter='\t')
        next(demand_reader, None)
        for row in demand_reader:
            demand.append(row)
    return demand

def getMolds(demand, MoldColumn):
    return [row[MoldColumn] for row in demand]

def getMachines(demand, MachineColumn):
    return [row[MachineColumn] for row in demand]

def getJobID(demand, JobID):
    return [row[JobID] for row in demand]

def indexDemandWorkcenters(demand, workcenters, workcenterColumn):
    for jobs in demand:
        jobs[workcenterColumn] = numpy.where(workcenters == jobs[workcenterColumn])[0][0]
    return demand

def getJobs(jobid, demand, primaryWCIdx, auxWCIdx, timeColumn, itemIdx):
    jobs = []
    alts = []
    filtered_list = list(
        filter(
            lambda obj: obj[6] == jobid,
            demand
        )
    )
    for job in filtered_list:
        alts.append((int(job[timeColumn]), job[primaryWCIdx], int(job[auxWCIdx]), job[itemIdx]))
    jobs.append(alts)
    return jobs

def decodePlan(plan, demand, primaryWCs, planPrimaryWCIdx, auxWCs, demandAuxWCIdx, demandItemIdx, planAuxWCIdx, planItemIdx):
    for job in plan:
        # colonne 1 e 2 e 3 dell'oggetto job rappresentano job_id e task_id e alternativa_id
        # la somma delle 2 restituisce l'indice per accedere alla lista "demand" e decodificare gli id
        #index = int(job[1])+int(job[2])
        job[planPrimaryWCIdx] = primaryWCs[job[planPrimaryWCIdx]]
        job[planAuxWCIdx] = auxWCs[job[planAuxWCIdx]]
        #job[planItemIdx] = demand[index][demandItemIdx]
    return plan

def createColors(molds):
    r = lambda: random.randint(0,255)
    colors = ['#%02X%02X%02X' % (r(),r(),r())]
    for mold in molds:
        colors.append('#%02X%02X%02X' % (r(),r(),r()))
    return colors

if __name__ == '__main__':
    # import domanda
    demand = importDemand()
    # definizione liste macchine/stampi
    machines = getMachines(demand, 3)
    molds = getMolds(demand, 4)
    jobids = getJobID(demand, 6)
    machines = numpy.unique(machines)
    molds = numpy.unique(molds)
    jobids = numpy.unique(jobids)
    # encoding domanda con identificativi numerici per macchine/stampi
    demand = indexDemandWorkcenters(demand, machines, 3)
    demand = indexDemandWorkcenters(demand, molds, 4)
    
    # costruzione lista jobs da far elaborare al solver
    # 3 campi per ogni alternativa
    #   1. duration
    #   2. macchina
    #   3. stampo
    for jobid in jobids:
        jobs_data.append(getJobs(jobid, demand, 3, 4, 9, 0))

    # elaborazione scheduling tramite solver
    plan = flexible_jobshop(jobs_data, len(machines), len(molds))
    # decoding domanda per risalire da id numerici macchine/stampi a valori parlanti
    # def decodePlan(plan, demand, primaryWCs, planPrimaryWCIdx, auxWCs, demandAuxWCIdx, demandItemIdx, planAuxWCIdx, planItemIdx)
    plan = decodePlan(plan, demand, machines, 3, molds, 4, 0, 6, 2)
    # conversione piano in dataframe
    plan = pd.DataFrame(plan, columns=['job_id','task_id','alt_id','machine','start','finish', 'mold', 'item'])
    # conversione start/end task da ore a data/ora
    d = datetime.now()
    date_start = pd.to_timedelta(plan['start'].astype(int), unit='H')
    plan['start_date'] = d + date_start
    date_end = pd.to_timedelta(plan['finish'].astype(int), unit='H')
    plan['end_date'] = d + date_end
    plan['caption'] = plan['item'] + ' ' + plan['machine'] + ' ' + plan['mold']
    #print(plan)
    #creazione gantt
    colors = createColors(molds)
    fig = px.timeline(plan, x_start="start_date", x_end="end_date", y="machine", color="mold", color_discrete_sequence=colors, hover_data=["item", "machine", "mold", "start_date", "end_date"])
    fig.update_yaxes(title="Machines", autorange="reversed", type="category", categoryorder="category ascending")
    fig.update_traces(marker=dict(line=dict(color='rgba(58, 71, 80, 1.0)', width=1)))
    fig.show()

[gregy4]

I recommend to prefer a light version.
You can prolong duration of very task by 4 hours. In case you are not happy with occupied resources of a task over these 4 hours, you can add 4 hours helper task for every task. Helper task should finish at start of a task, so only one constraint should be needed for this.

[uprox]

Thank you for your suggestion.
So, if I understood well, I should add a task with duration = 4 for each job (to date I have just 1 task - with n alternatives - for each job) and after I have to add a constraint to bind the finish of this helper task to the start of the main one. Is it correct?
To bind the helper task to the main task, shoud I use something like model.Add(something) ?
Thank you

Sorry for my basic questions... I'm new to or tools.

[gregy4]

Yes. Every task has start, end and duration variable. In your case model.Add(helperTask.end = task.start) should do what you want.
Maybe some relations should be moved to helper task. For example when task1 should be before task2, task1 should be also before helper task of task2.