Space between resource tasks in an interval

[dreinon]

It would be interesting to have a constraint that lets you control the space between a resource tasks in a certain time interval or intervals.

We could apply the min/exact/max options too.

Then we could say for instance that for Worker A, tasks that are scheduled between 0 and 20 can have a max distance of 0, which given a tasks in a period would force other tasks either to be consecutive to the first or not to be in that period.

[tpaviot]

What is the exact definition of the space you mean? Given two tasks t_i and t_j I guess this is

space = t_j.start-t_i.end if t_j.start>=t_j.end OR t_i.start-t_j.end if t_i.start>=t_j.end

But in case of three or more tasks ?

[dreinon]

Since we are talking about task of the same worker, then they can't overlap. Thus, in order of time, you go through the tasks two by two and compute the difference between the end of the first and the start of the second.

Imagine a worker has the following tasks of duration 1: [[0,1],[3,4],[4,5], [6,7], [10, 11]]

Then, to compute the distance from the tasks in the interval [0,8] (for instance), we would:

• Get the tasks that are inside that interval: [[0,1],[3,4],[4,5], [6,7]]
• Then we go through them two by two:

  tasks = [[0,1],[3,4],[4,5], [6,7]]
  MAX_DISTANCE = 2
  for i, _ in enumerate(tasks[:-1]):
  current_task_end = tasks[i][1]
  next_task_start = tasks[i+1][0]
  distance = next_task_start - current_task_end

  Which means that in order of three tasks to be consecutive, in the first place the second has to be consecutive with the first.

[tpaviot]

The problem is to get the tasks "in order of time", which is not known until the solver has completed the schedule computation. Or, in other words, you have to find a formulation that is independent of tasks order. Or, another solution, use a set of z3 constraints that sort tasks according to their start time. I don't know how to do this properly, there are only a few examples available (https://gist.github.com/philzook58/4aa17037bcb9b1ee57f4afdf87f7f3fd for instance, or https://github.com/Z3Prover/z3/commit/785fab74f473f1ce7a58ac6420824ef275e65a4c#diff-477eb2485214c3a39e36664624ea1eea021a202e461482af1678fa1e9c4c44b3)

[dreinon]

What about using workload constraint and implies instead? Imagine that max distance between tasks is 2 between the time interval [0,7]. Then for all tasks of the resource we want, we say that if the task is scheduled and is inside the period And(task.start >= 0, task.end <= 7), then workload of [0, task.start-2] is 0 and the same for [task.end+2, 7].

The implementation would probably need refinement but just wanted to share the idea.

[dreinon]

I have implemented a first draft of this constraint:

from z3 import If, And, ArithRef
import processscheduler as ps

class Worker(ps.Worker):
    def __init__(self,
                 name: str,
                 productivity: Optional[int] = 1,
                 cost: Optional[int] = None) -> None:
        super().__init__(name, productivity, cost)

        self.workload = Sum([end-start for start, end in self.get_busy_intervals()])

    def get_tasks(self) -> list[ps.FixedDurationTask]:
        return list(self.busy_intervals.keys())

def z3_max(a, b):
    return If(a > b, a, b)

def z3_min(a, b):
    return If(a < b, a, b)

def create_resource_tasks_distance_constraint(resource: Worker, distance: int, horizon: int, mode: str):
    if mode not in {'min', 'max', 'exact'}:
        raise Exception('Mode should be min, max or exact')

    constraints = []
    for task in resource.get_tasks():
        before_point = z3_max(task.start - distance, 0)
        after_point = z3_min(task.start + distance, horizon)
        workload = {}
        if mode == 'min':
            workload.update({
                (before_point, task.start): 0,
                (task.end, after_point): 0
            })
        elif mode == 'max':
            before_point = z3_max(before_point - task.duration, 0)
            after_point = z3_min(after_point + task.duration, horizon)
            workload.update({
                (0, before_point): 0,
                (after_point, horizon): 0
            })
        elif mode == 'exact':
            before_task_end = before_point
            before_task_start = z3_max(before_task_end - task.duration, 0)

            after_task_start = after_point
            after_task_end = z3_min(after_task_start + task.duration, horizon)

            workload.update({
                (0, before_task_start): 0,
                (before_task_end, task.start): 0,
                (task.end, after_task_start): 0,
                (after_task_end, horizon): 0
            })

        workload_ctr = ps.WorkLoad(resource, workload, kind='exact')
        constraints.append(ps.implies(task.scheduled, [workload_ctr]))

    return ps.and_(constraints)

I'm aware of the following issues:

1. The workload dict not only deals with int but also with ArithRef, and I don't know if Workload constraint was made thinking of being compatible with this.
2. There can be a case when the workload dict has (0,0) or (horizon, horizon) periods, I don't know if this would work with the current implementation of workload constraint.

Please, feel free to analize the code and modify it, change it, or use it to implement an actual constraint for the current project. Thanks!

[tpaviot]

Thanks for this update. This should be turned into a new constraint, I'll have a look at the code.

[dreinon]

Awesome :) Let me know with any updates!

[dreinon]

By the way, I forgot that having a workload of 0 is being unavailable, so maybe an implementation with ResourceUnavailable would be cleaner? What do you think?

from z3 import If, And, ArithRef
import processscheduler as ps

class Worker(ps.Worker):
    def __init__(self,
                 name: str,
                 productivity: Optional[int] = 1,
                 cost: Optional[int] = None) -> None:
        super().__init__(name, productivity, cost)

        self.workload = Sum([end-start for start, end in self.get_busy_intervals()])

    def get_tasks(self) -> list[ps.FixedDurationTask]:
        return list(self.busy_intervals.keys())

def z3_max(a, b):
    return If(a > b, a, b)

def z3_min(a, b):
    return If(a < b, a, b)

def create_resource_tasks_distance_constraint(resource: Worker, distance: int, horizon: int, mode: str):
    if mode not in {'min', 'max', 'exact'}:
        raise Exception('Mode should be min, max or exact')

    constraints = []
    for task in resource.get_tasks():
        before_point = z3_max(task.start - distance, 0)
        after_point = z3_min(task.start + distance, horizon)
        unavailability = []
        if mode == 'min':
            unavailability.extend([(before_point, task.start), (task.end, after_point)])
        elif mode == 'max':
            before_point = z3_max(before_point - task.duration, 0)
            after_point = z3_min(after_point + task.duration, horizon)
            unavailability.extend([(0, before_point), (after_point, horizon)])
        elif mode == 'exact':
            before_task_end = before_point
            before_task_start = z3_max(before_task_end - task.duration, 0)

            after_task_start = after_point
            after_task_end = z3_min(after_task_start + task.duration, horizon)

            unavailability.extend(
                [(0, before_task_start),
                 (before_task_end, task.start),
                 (task.end, after_task_start),
                 (after_task_end, horizon)])

        unavailability_ctr = ps.ResourceUnavailable(resource, unavailability)
        constraints.append(ps.implies(task.scheduled, [unavailability_ctr]))

    return ps.and_(constraints)

[dreinon]

I'm also thinking that with the current implementation, when task.duration is used, it's suposing that other resource tasks would have the same duration that the actual task in the for loop, which is incorrect. For "setting" the correct available space, all tasks would need to be the same duration, so I propose either adding a task_duration parameter or creating a subclass of Resource which initializer has this task_duration parameter and which all its tasks must have the same duration. Then, this constraint would only be available for this kind of resource and would get the tasks duration from the resource instead of being a separate parameter.

[tpaviot]

Can you please extend the provided code with a short example so that I can check the result? I'm a bit confused with using a ResourceUnavailable constraint, it's a kind of tweak that does not really conform to this constraint semantics. Just to know, does it work as expected?

[tpaviot]

I have another implementation in my, let me send you sthg in a few minutes

[dreinon]

Ok, thanks! About the ResourceUnavailable constraint, I changed it because I think it's the same to say that the workload of a resource should be 0 between period [2,6], and to say that the resource is unavailable between period [2,6]. Don't you think?

[tpaviot]

Please test, there seems to be an issue with the task duration (space is 6 whereas it should be 8).

import processscheduler as ps
from z3 import *
# same problem, but we force two tasks to be scheduled at start and end
# of the planning

pb = ps.SchedulingProblem("DistanceBetweenTasks")
N = 4  # number of tasks
tasks = [ps.FixedDurationTask("Task_%i" %i, duration=2) for i in range(N)]
worker_1 = ps.Worker("Worker1")

for t in tasks:
    t.add_required_resource(worker_1)

def sort_list_of_z3_var(lst):
    N = len(lst)
    a = [Int(f"x_{i}") for i in range(N)] #build initial array in z3 variables
    new_constraints = []
    # add the related constraints
    for i in range(N):
        new_constraints.append(Or([a[i] == lst[j] for j in range(N)]))
    new_constraints.append(And([a[i] < a[i+1] for i in range(N-1)]))
    return a, new_constraints

def create_resource_tasks_distance_constraint(resource, distance, mode):
    if mode not in {'min', 'max', 'exact'}:
        raise Exception('Mode should be min, max or exact')

    starts = []
    ends = []
    for start_var, end_var in resource.busy_intervals.values():
        starts.append(start_var)
        ends.append(end_var)
    # sort both lists
    sorted_starts, c1 = sort_list_of_z3_var(starts)
    sorted_ends, c2 = sort_list_of_z3_var(starts)
    # from now, starts and ends are sorted in asc order
    # the space between two consecutive tasks is the sorted_start[i+1]-sorted_end[i]
    # we just have to constraint this variable
    c3 = []
    for i in range(1, len(sorted_starts)):
        if mode == "min":
            new_cstr = sorted_starts[i] - sorted_ends[i-1] >= distance
        elif mode == "max":
            new_cstr = sorted_starts[i] - sorted_ends[i-1] <= distance
        elif mode == "exact":
            new_cstr = sorted_starts[i] - sorted_ends[i-1] == distance
        c3.append(new_cstr)
    return c1 + c2 + c3

c = create_resource_tasks_distance_constraint(worker_1, distance=8, mode="exact")

# plot solution
solver = ps.SchedulingSolver(pb)
solver.add_constraint(c)

solution = solver.solve()
print(solution)

solution.render_gantt_matplotlib()

[tpaviot]

The previous implementation is based on sorting start and end times for tasks of a resource

[tpaviot]

I guess this won't work with optional tasks because optional task start and ent times are always -1

[dreinon]

Thanks! I'll look at your implementation tomorrow :)

[tpaviot]

I fixed optional tasks issue, will create a PR with the relevant code

[dreinon]

I noticed that mode parameter is named kind in other constraints.

[tpaviot]

Yes, mode, kind I don't really know which one is better. I chose kind, but it's definitely not a good name

[dreinon]

Task distance constraint working awesome :+1:. Closing this issue, thanks @tpaviot.

[tpaviot]

Good to know!