pyschedule - resource scheduling in python
Available items
The developer of this repository has not created any items for sale yet. Need a bug fixed? Help with integration? A different license? Create a request here:
pyschedule is python package to compute resource-constrained task schedules. Some features are:
Previous use-cases include:
A simple pyschedule scenario where three houshold tasks need to get assigned to two persons, Alice and Bob:
from pyschedule import Scenario, solvers, plotters, altthe planning horizon has 10 periods
S = Scenario('household',horizon=10)
two resources: Alice and Bob
Alice, Bob = S.Resource('Alice'), S.Resource('Bob')
three tasks: cook, wash, and clean
cook = S.Task('cook',length=1,delay_cost=1) wash = S.Task('wash',length=2,delay_cost=1) clean = S.Task('clean',length=3,delay_cost=2)
every task can be done either by Alice or Bob
cook += Alice | Bob wash += Alice | Bob clean += Alice | Bob
compute and print a schedule
solvers.mip.solve(S,msg=1) print(S.solution())
INFO: execution time for solving mip (sec) = 0.025493860244750977 INFO: objective = 1.0 [(clean, Alice, 0, 3), (cook, Bob, 0, 1), (wash, Bob, 1, 3)]
We can also plot the schedule as a GANTT-chart and write it to a file:
plotters.matplotlib.plot(S,img_filename='pics/household.png')
There are example notebooks here and simpler examples in the examples folder. Install it with pip:
pip install pyschedule
Note that pyschedule aims to be a general solver for small to medium-sized scheduling problems. A typical scenario that pyschedule consists of 10 resources and 100 tasks with a planning horizon of 100 periods. If your requirements are much larger than this, then an out-of-the box solution is hard to obtain. There are some ways to speed-up pyschedule (e.g. see task groups and solver parameters). It is also possible to build heuristics on top of pyschedule to solve large-scaled scheduling problems.
When creating a scenario using pyschedule, the following imports are recommended:
from pyschedule import Scenario, solvers, plotters, alt
This allows the creation of a scenario:
S = Scenario('hello_world',horizon=10)
This scenario is named
hello_worldand has a time horizon of 10 periods. The granularity of the periods depends on your problem, e.g. a period could be an hour, a week, or a day. However, having far more than 100 periods makes the computation of a schedule quite hard. Some tricks to reduce the number of periods are:
We need at least one resource in a scenario:
R = S.Resource('R')
It is convenient to have identical resource and variable names, like
R. During each period, some task can be schedule this period. A resource can be anything from a person to an object like a machine in a factory. It is only required that a resource can be used by at most one task in every period.
Next we add a task to the scenario:
T = S.Task('T',length=1,delay_cost=1)
This task has length 1, that is, it requires only 1 period to finish. Since 1 is the default length of a task, we would not have to set this explicitely. Moreover, we set the delay cost to 1, that is, delaying this job for one period increases the cost of a schedule by 1, which motivates to finish this task as early as possible.
We define that task
Trequires resource
Ras follows:
T += R
Then we compute and print a schedule as follows:
solvers.mip.solve(S,msg=0) print(S.solution())
INFO: execution time for solving mip (sec) = 0.014132976531982422 INFO: objective = 0.0 [(T, R, 0, 1)]
The output first shows the time required to solve the problem. Also the objective is plotted. Since the cost of this schedule is only the delay cost of task
T, which is schedule in period 0, the total cost is 0 as well. The standard way to present a schedule is a list of task-resource pairs with time requirements. E.g. the output above says that task
Tshould be scheduled on resource
Rfrom period 0 to 1.
It is not necessary to define cost in a scenario. In this case, a solver will simply try to find a feasible schedule. Not defining any cost will sometimes even speed up the computation. However, in most scenarios, setting at least some delay cost makes sense.
We set the delay cost of a task to 1 as follows:
T = S.Task('T',delay_cost=1)
This means that if this task is scheduled in period 0, then there will no delay cost, if it is schedule in period 1, there will be total cost 1 and so on. Hence, it makes sense to schedule this task as early as possible. Note that delay cost can also be negative, in which case a task will be pushed to the end of a schedule. Also note that a task with a higher delay cost is more likely to be scheduled earlier if there are no other constraints that are preventing this. The default delay cost is
None.
Schedule cost can be used for optional tasks, that is, we provide some positive or negative reward if a task is scheduled:
from pyschedule import Scenario, solvers, plotters, alt S = Scenario('schedule_cost',horizon=10) R = S.Resource('R')not setting a schedule cost will set it to None
T0 = S.Task('T0',length=2,delay_cost=1)
setting the schedule cost of T1 to -1
T1 = S.Task('T1',length=2,delay_cost=1,schedule_cost=-1)
T0 += R T1 += R solvers.mip.solve(S,msg=1) print(S.solution())
INFO: execution time for solving mip (sec) = 0.016648054122924805 INFO: objective = 0.0 [(T0, R, 0, 2)]
In the schedule above, scheduling task
T1with schedule cost -1 would decrease the total cost by 1, but then we would have to schedule both tasks
T0and
T1, and hence one of them would have to start in period 2. This would result an additional delay cost of 2. Consequently, it makes more sense not to schedule
T1.
Using a resource for some periods might imply additional resource cost:
from pyschedule import Scenario, solvers, plotters, alt S = Scenario('resource_cost',horizon=10)assign a cost per period of 5
R = S.Resource('R',cost_per_period=5)
T = S.Task('T',length=2,delay_cost=1) T += R solvers.mip.solve(S,msg=1) print(S.solution())
INFO: execution time for solving mip (sec) = 0.01111602783203125 INFO: objective = 10.0 [(T, R, 0, 2)]
The total cost of the computed schedule is 5 although the single task is scheduled in the first period. This is due to the fact that scheduling any task costs 5 on resource
R.
To simplify the definition of tasks, it is possible to define task lists:
from pyschedule import Scenario, solvers, plotters, alt S = Scenario('many_tasks',horizon=10)create 5 tasks of the same type
T = S.Tasks('T',num=5,length=1,delay_cost=1)
print(T)
[T0, T1, T2, T3, T4]
We created 5 tasks of length 1 and delay cost 1. The index of the tasks is padded to the end of the given task name. Therefore, avoid task names ending with digits. Note that it would also be possible to create all tasks separately. But if they are similar, this simplifies the definition of scheduling problems. Finally, we can similarly define lists of resources:
from pyschedule import Scenario, solvers, plotters, alt S = Scenario('many_resources',horizon=10)create 5 resources of the same type
R = S.Resources('R',num=5)
print(R)
[R0, R1, R2, R3, R4]
It is possible to assign multiple resources to a task, either we define that one of these resources is required or all:
from pyschedule import Scenario, solvers, plotters, alt S = Scenario('resources_assignment',horizon=10)R = S.Resources('R',num=2) T = S.Tasks('T',num=2,delay_cost=1)
T0 requires either resource R0 or R1
T[0] += R[0] | R[1]
T1 requires resources R0 and R1
T[1] += R[0], R[1]
print the resources requirement
print(T[0].resources_req) print(T[1].resources_req)
[R0|R1] [R0, R1]
Note that if we have a list of resources, like above, we can also use the
alt-operator:
# T0 requires any of the resources from pyschedule import alt T[0] += alt(R)T1 requires all of the resources
T1 += R
Now we can solve this scenario:
python solvers.mip.solve(S,msg=1) print(S.solution())
INFO: execution time for solving mip (sec) = 0.05312514305114746 INFO: objective = 1.0 [(T0, R0, 0, 1), (T1, R0, 1, 2), (T1, R1, 1, 2)]Therefore,
T0is scheduled on resource
R0in period 0 and
T1on resources
R0and
R1in period 1.
It is often necessary to ensure that two tasks select the same resources:
from pyschedule import Scenario, solvers, plotters, alt S = Scenario('resources_dep',horizon=10)R = S.Resources('R',num=2) T = S.Tasks('T',num=2,delay_cost=1)
assign all resources to both resources
T += alt(R)
if T[1] is assigned to any resource in R, then also T[0]
T[0] += T[1] * R
plot the resource dependencies of task T0
print(T[0].tasks_req)
[T1R0, T1R1]
Now we can solve this scenario
python solvers.mip.solve(S,msg=1) print(S.solution())
INFO: execution time for solving mip (sec) = 0.02900981903076172 INFO: objective = 1.0 [(T1, R0, 0, 1), (T0, R0, 1, 2)]It would be better to distribute the two tasks to the two resources. However, due to the defined resource dependencies, they must be assigned to the same one.
We can restrict the periods when a job can be scheduled or when resource is available:
from pyschedule import Scenario, solvers, plotters, alt S = Scenario('periods',horizon=10)restrict the periods to 2 and 3
T = S.Task('T', length=1, periods=[3,4])
restrict the periods to the range 1..3
R = S.Resource('R', periods=range(1,4)) T += R solvers.mip.solve(S,msg=1) print(S.solution())
INFO: execution time for solving mip (sec) = 0.04649972915649414 INFO: objective = 0.0 [(T, R, 3, 4)]
Clearly, due to the periods restrictions, the only possible period to schedule task
Tis 3.
Another way to restrict the periods when a task can be scheduled are bounds:
from pyschedule import Scenario, solvers, plotters, alt S = Scenario('bounds',horizon=10) T = S.Task('T', length=1, delay_cost=1) R = S.Resource('R') T += Radd the constraints that T needs to get schedule after period 1 but before 5
S += T > 1, T < 5
solvers.mip.solve(S,msg=1) print(S.solution())
INFO: execution time for solving mip (sec) = 0.03368401527404785 INFO: objective = 1.0 [(T, R, 1, 2)]
This contraint is a lax bound, that is, task
Tcan be schedule in any point after period 1. If we want to enforce when exactly
Tis scheduled, we can use a tight bound. E.g. to force
Tto be schedule exactly after period 1, we can write:
S += T >= 1
Tasks often need to get scheduled in a certain order:
from pyschedule import Scenario, solvers, plotters S = Scenario('lax_precedence',horizon=10) R = S.Resource('R') T = S.Tasks('T',num=2,length=1,delay_cost=1) T += Rgive T0 a higher delay cost
T[0].delay_cost = 2
add a precedence constraint to ensure that it is still scheduled one period after T1 finishes
S += T[1] + 1 < T[0]
solvers.mip.solve(S,msg=1) print(S.solution())
INFO: execution time for solving mip (sec) = 0.019501924514770508 INFO: objective = 4.0 [(T1, R, 0, 1), (T0, R, 2, 3)]
Since task
T0is delayed two periods, we get a total delay cost of 4. If we would not have the precedence constraint, we could schedule
T0first and only get delay cost 1. Note that the
+ 1is optional.
We call this a lax precedence constraint. Similarly to tight bounds, tight precedence constraints additionally ensure that jobs are executed directly after each other:
from pyschedule import Scenario, solvers, plotters S = Scenario('tight_precedence',horizon=10) R = S.Resource('R') T = S.Tasks('T',num=2,length=1,delay_cost=2) T += Rgive T0 a negative delay cost
T[0].delay_cost = -1
ensure that T[0] is scheduled exactly two periods after T[1]
S += T[1] + 2 <= T[0]
solvers.mip.solve(S,msg=1) print(S.solution())
INFO: execution time for solving mip (sec) = 0.016117095947265625 INFO: objective = -3.0 [(T1, R, 0, 1), (T0, R, 3, 4)]
Since
T0has negative delay cost, it would be pushed to the end of the schedule, but the tight precedence constraint ensures that it is scheduled two periods after
T1finishes. If the delay cost of
T1would be smaller than
T0, than both tasks would be pushed to the end of the schedule.
It is often required that precedence constraints are only applied if two tasks are assigned to the same resource, e.g. if we want to ensure that a certain task is the last one that runs on some resource:
from pyschedule import Scenario, solvers, plotters, alt S = Scenario('cond_precedence',horizon=10) R = S.Resources('R',num=2)T = S.Task('T',length=1,delay_cost=1) T_final = S.Tasks('T_final',num=2,length=1,delay_cost=1) T_final[0] += R[0] T_final[1] += R[1] T += alt(R)
conditional precedences
S += T * R[0] < T_final[0] S += T * R[1] < T_final[1]
solvers.mip.solve(S,msg=1) print(S.solution())
INFO: execution time for solving mip (sec) = 0.040048837661743164 INFO: objective = 1.0 [(T, R0, 0, 1), (T_final1, R1, 0, 1), (T_final0, R0, 1, 2)]
The first conditional precedence implies that if task
Tis scheduled on
R[0], then
T_final[0]is scheduled afterwards. Therefore, it is allowed that
T_final[1]is scheduled in the same period as
Tsince
Tis not scheduled on
R[1].
Capacity constraints can be used to restrict the number tasks which are executed during a certain time period:
from pyschedule import Scenario, solvers, plotters S = Scenario('capacities',horizon=10) R = S.Resource('R') T = S.Tasks('T',num=4,length=1,delay_cost=1) T += Rcapacity constraint to limit the number of tasks until period 5 to 3
S += R[0:5] <= 3
solvers.mip.solve(S,msg=1) print(S.solution())
INFO: execution time for solving mip (sec) = 0.015291213989257812 INFO: objective = 8.0 [(T3, R, 0, 1), (T0, R, 1, 2), (T1, R, 2, 3), (T2, R, 5, 6)]
Due to the capacity constraint, one task is scheduled after period 5. If not defined otherwise, the capacity constraint is applied to the lengths of the task. That is, the sum of lengths of tasks before period 5 is required to be at most 3. We can make this more explicit by writing:
S += R['length'][0:5] <= 3
Finally, if we want to bound the maximum instead of the sum, we can write:
python S += R['length'][0:5].max <= 3
Cases where task lengths are larger than one deserve a special treatment:
from pyschedule import Scenario, solvers, plotters S = Scenario('capacities',horizon=10) R = S.Resource('R')task with non-unit length
T = S.Task('T',length=4,delay_cost=1) T += R
S += R[0:5] <= 3
solvers.mip.solve(S,msg=1) print(S.solution())
INFO: execution time for solving mip (sec) = 0.022754907608032227 INFO: objective = 2.0 [(T, R, 2, 6)]
Task
Thas to start in period 2 because of the capacity constraint. This is possible because the length of the part of this task which lies within the capacity constraint is 3. Specifically, the part scheduled in periods 2,3 and 4. This holds in general, a task contributes to a standard capacity constraint proportionally to how much it overlaps with the capacity constraint. This generalizes to user-defined task attributes as described in the next section.
We can apply capacity constraints to all task attributes, not just the task lengths, but also user-defined ones:
from pyschedule import Scenario, solvers, plotters S = Scenario('capacities_myattribute',horizon=10) R = S.Resource('R')define the additional property named myproperty uniformly as 1
T = S.Tasks('T',num=4,length=1,delay_cost=1,myattribute=1)
set it to 0 for the first task
T[0].myattribute = 0 T += R
the sum of myproperty must be smaller than 3 until period 5
S += R['myattribute'][0:5] <= 3
solvers.mip.solve(S,msg=1) print(S.solution())
INFO: execution time for solving mip (sec) = 0.018787145614624023 INFO: objective = 6.0 [(T3, R, 0, 1), (T0, R, 1, 2), (T1, R, 2, 3), (T2, R, 3, 4)]
Since
T[0]does not add anything to the sum of the myattribute-values before period 5, all tasks can be scheduled before this period.
The default way to aggregate within the range of a capacity constraint is to summarize. On the other hand, if we want to ensure that some attribute does not change too much over time, we can also restrict the sum of differences of this attribute:
from pyschedule import Scenario, solvers, plotters S = Scenario('capacities_diff',horizon=10) R = S.Resource('R')T = S.Tasks('T',num=4,length=1,delay_cost=1,myattribute=1) T[0].delay_cost = 2 T[0].myattribute = 0 T[1].delay_cost = 2 T[1].myattribute = 0 T += R
limit the sum of differences of myattribute to 1
S += R['myattribute'].diff <= 1
solvers.mip.solve(S,msg=1) print(S.solution())
INFO: execution time for solving mip (sec) = 0.05473184585571289 INFO: objective = 11.0 [(T3, R, 0, 1), (T2, R, 1, 2), (T0, R, 2, 3), (T1, R, 3, 4)]
Note that if we do not define the range of a capacity constraint like above, then the constraint is applied to the complete time horizon. In the scenario above, it would be advantageous to schedule tasks
T[0]and
T[1]as early as possible, since they have a higher delay cost. However, if would schedule them in periods 0 and 1, respectively, and directly afterwards
T[2]and
T[3], then myattribute would first increase by 1 in period 2 and afterwards decrease again by 1 in period 4, resulting in a sum of differences of 2.
The
.diff-capacity constraint limits the sum of increases and decreases. If we only want to limit the increases or decreases, then we can use
.diff_upor
.diff_down, respectively.
We can combine capacity constraints doing natural arithmetic:
from pyschedule import Scenario, solvers, plotters S = Scenario('capacities_arithmetic',horizon=10) R = S.Resource('R')T = S.Tasks('T',num=4,length=1,delay_cost=1,myattribute=1) T += R
add two capacities
S += R['myattribute'][:3] + R['myattribute'][5:7] <= 1
solvers.mip.solve(S,msg=1) print(S.solution())
INFO: execution time for solving mip (sec) = 0.03471493721008301 INFO: objective = 14.0 [(T3, R, 0, 1), (T2, R, 3, 4), (T1, R, 4, 5), (T0, R, 7, 8)]
Since one task is schedule in period 0, we cannot schedule any more tasks in periods 0 to 2 or in periods 5 to 6 . Therefore, we squeeze in two tasks in periods 3 and 4 and one task in period 7.
There are often task redundancies in a planning project, e.g. there might be a group of tasks which are interchangeable. That is, they could be swapped in the schedule without changing its cost or feasibility. Given this information in the
is_group-attribute, this can be exploited by the solver to often drastically speed-up the computation:
from pyschedule import Scenario, solvers, plotters S = Scenario('task_groups',horizon=10) R = S.Resource('R')these tasks are interchangeable
T = S.Tasks('T',num=10,length=1,delay_cost=1,is_group=True) T += R
solvers.mip.solve(S,msg=1) print(S.solution())
INFO: execution time for solving mip (sec) = 0.01534271240234375 INFO: objective = 45.0 [(T0, R, 0, 1), (T1, R, 1, 2), (T2, R, 2, 3), (T3, R, 3, 4), (T4, R, 4, 5), (T5, R, 5, 6), (T6, R, 6, 7), (T7, R, 7, 8), (T8, R, 8, 9), (T9, R, 9, 10)]
Running this with setting
is_group=Falseonly slightly increases the running time, but there are scenarios where this difference is much more significant:
INFO: execution time for solving mip (sec) = 0.025635719299316406 INFO: objective = 45.0 [(T7, R, 0, 1), (T0, R, 1, 2), (T4, R, 2, 3), (T2, R, 3, 4), (T5, R, 4, 5), (T6, R, 5, 6), (T1, R, 6, 7), (T9, R, 7, 8), (T8, R, 8, 9), (T3, R, 9, 10)]
CAUTION: combining task groups with capacities with resource dependencies might not work in some cases.
The default pyschedule backend is a time-indexed mixed integer formulation (MIP). There are the following parameters:
CBCwhich comes preinstalled with package
pulp. If SCIP is installed (command
scipmust be running), you can use
SCIP. Finally, if you have CPLEX installed (command
cplexmust be running), you can use
CPLEX
E.g. this could be used as follows:
solvers.mip.solve(S,kind='CPLEX',time_limit=60,random_seed=42,msg=1)
The default pyschedule backend to plot a schedule is matplotlib. Some parameters are:
.png-file (default is None)
E.g. this could be used as follows:
plotters.matplotlib.plot(S,img_filename='tmp.png',img_size=(5,5),hide_tasks=[T])
FINAL CAUTION: pyschedule is under active development, there might be non-backward-compatible changes.
import sys sys.path.append('src')