ProcessScheduler

A a 100% pure python library to compute resource-constrained task schedules. It is distributed under the terms of the GNU General Public License v3 or later.

Introduction

ProcessScheduler is a versatile tool designed to streamline operations in various industrial domains, including manufacturing, construction, healthcare, and more. It serves as a solution for tackling intricate scheduling challenges that defy straightforward resolutions.

Within this toolkit, you’ll find a rich array of features, including:

• Task Definition: Define tasks with zero, fixed, or variable durations, along with work_amount specifications.

• Resource Management: Create and manage resources, complete with productivity and cost attributes. Efficiently assign resources to tasks.

• Temporal Task Constraints: Handle task temporal constraints such as precedence, fixed start times, and fixed end times.

• Resource Constraints: Manage resource availability and allocation.

• Logical Operations: Employ first-order logic operations to define relationships between tasks and resource constraints, including and/or/xor/not boolean operators, implications, if/then/else conditions.

• Multi-Objective Optimization: Optimize schedules across multiple objectives.

• Gantt Chart Visualization: Visualize schedules effortlessly with Gantt chart rendering, compatible with both matplotlib and plotly libraries.

• Export Capabilities: Seamlessly export solutions to JSON format and SMT problems to SMTLIB format.

This comprehensive guide will walk you through the process of model creation, solver execution, and solution analysis, making it a valuable resource for harnessing the full potential of ProcessScheduler.

What’s inside

ProcessScheduler operates on models written in the Python programming language, offering the flexibility to accommodate a wide range of scheduling requirements for tasks and resources.

To tackle scheduling challenges, ProcessScheduler leverages the power of the Microsoft SMT Z3 Prover, a licensed MIT SMT solver. For those eager to delve deeper into the optimization aspects of the solver, a comprehensive reference can be found in the paper “Bjorner et al. νZ - An Optimizing SMT Solver (2016).” Additionally, an introductory guide to programming with Z3 in Python is available at z3-py-tutorial. It’s worth noting that Z3 is the only mandatory dependency for ProcessScheduler.

Furthermore, the tool offers the flexibility to visualize scheduling solutions by rendering them into Gantt charts, which can be exported in common formats such as JPG, PNG, PDF, or SVG. Please note that the optional libraries, matplotlib and plotly, are not pre-installed but can be easily integrated based on your preferences and needs.

Download/install

Use pip to install the package and the required dependencies (Z3) on your machine:

pip install ProcessScheduler

and check the installation from a python3 prompt:

>>> import processscheduler as ps

Development version

Create a local copy of the github repository:

git clone https://github.com/tpaviot/ProcessScheduler

Then install the development version:

cd ProcessScheduler
pip install -e .

Scheduling problem

The SchedulingProblem class is the container for all modeling objects, such as tasks, resources and constraints.

Time slots as integers

A SchedulingProblem instance holds a time interval: the lower bound of this interval (the initial time) is always 0, the upper bound (the final time) can be set by passing the horizon attribute to the __init__() method:

my_problem = SchedulingProblem('MySchedulingProblem', horizon=20)

The interval’s duration is subdivided into discrete units called periods, each with a fixed duration of 1. If \(horizon\) is set to a value, then the number of periods is equal to \(horizon\), and the number of points within the interval \([0;horizon]\) is \(horizon+1\).

Warning

ProcessScheduler handles variables using integer values.

A period represents the finest granularity level for defining the timeline, task durations, and the schedule itself. This timeline is dimensionless, allowing you to map a period to your desired duration, be it in seconds, minutes, hours, or any other unit. For instance:

• If you aim to schedule tasks within a single day, say from 8 am to 6 pm (office hours), resulting in a 10-hour time interval, and you plan to schedule tasks in 1-hour intervals, then the horizon value should be set to 10 to achieve the desired number of periods:

\[horizon = \frac{18-8}{1}=10\]

• If your task scheduling occurs in the morning, from 8 am to 12 pm, resulting in a 4-hour time interval, and you intend to schedule tasks in 1-minute intervals, then the horizon value must be 240:

\[horizon = \frac{12-8}{1/60}=240\]

Note

The horizon attribute is optional. If it’s not explicitly provided during the __init__() method, the solver will determine an appropriate horizon value that complies with the defined constraints. In cases where the scheduling problem aims to optimize the horizon, such as achieving a specific makespan objective, manual setting of the horizon is not necessary.

Mapping integers to datetime objects

To enhance the readability of Gantt charts and make schedules more intuitive, ProcessScheduler allows you to represent time intervals in real dates and times rather than integers. You can explicitly set time values in seconds, minutes, hours, and more. The smallest time duration for a task, represented by the integer 1, can be mapped to a Python timedelta object. Similarly, any point in time can be mapped to a Python datetime object.

Creating Python timedelta objects can be achieved as follows:

from datetime import timedelta
delta = timedelta(days=50,
                  seconds=27,
                  microseconds=10,
                  milliseconds=29000,
                  minutes=5,
                  hours=8,
                  weeks=2)

For Python datetime objects, you can create them like this:

from datetime import datetime
now = datetime.now()

These attribute values can be provided to the SchedulingProblem initialization method as follows:

problem = ps.SchedulingProblem('DateTimeBase',
                                horizon=7,
                                delta_time=timedelta(minutes=15),
                                start_time=datetime.now())

Once the solver has completed its work and generated a solution, you can export the end times, start times, and durations to the Gantt chart or any other output format.

Note

For more detailed information on Python’s datetime package documentation and its capabilities, please refer to the datetime Python package documentation. This documentation provides comprehensive guidance on working with date and time objects in Python.

Task

According to the APICS dictionary, a task may either be:

1. In project management, the lowest level to which work can be divided on a project

2. In activity-based cost accounting, a task, a subdivision of an activity, is the least amount of work. Tasks are used to describe activities.

In the context of this software library, the concept of a task aligns with the first definition. ProcessScheduler’s primary objective is to compute a chronological sequence, or temporal order, for a collection of tasks while adhering to a specific set of constraints.

Common base Task model

The Task class and its derivatives represent any task. A Task instance is defined by the three following parameters:

• start: a point in the \([0, horizon]\) integer interval. If the task is scheduled, then \(start>=0\)

• end: a point in the \([0, horizon]\) integer interval. If the task is scheduled, then \(end>=start\) and \(end<=horizon\)

• duration: a integer number of periods, such as \(duration=end-start\)

# Example: The duration of this task depends on the number of workers handling boxes.
move_boxes = VariableDurationTask('MoveBoxesFromMachineAToInventory')

Warning

Each Task instance must have a unique name in the scheduling problem. To prevent that two tasks share the same name, ProcessScheduler raises an exception if ever a task with an existing name is already created.

Three Task derivative classes can be used to represent a task: FixedDurationTask, ZeroDurationTask, VariableDurationTask.

FixedDurationTask class

The duration of a FixedDurationTask is known a priori. You must pass the task name and duration arguments when creating the instance:

# I assume one period to be mapped to 15min, cooking will be 1.5 hour
# so the chicken requires 6*15mn=1.5h to be cooked
cook_chicken = FixedDurationTask('CookChicken', duration=6)

ZeroDurationTask class

A ZeroDurationTask is a FixedDurationTask where \(duration=0\), that is to say \(start=end\). Useful to represent project milestones, or other important points in time.

project_kickup = ZeroDurationTask('KickUp')

VariableDurationTask class

A VariableDurationTask represents a task for which the duration is not known. The solver is expected to find a duration that satisfies the constraints (the duration may depend on the number of resources assigned to the task). You can bound the duration by using max_duration and/or min_duration parameters.

# 48h max to get done
plant_wheat_seeds = VariableDurationTask('PlantWheatSeeds', max_duration=48)

A VariableDurationTask duration can be selected among a list of possible durations. The solver decides the duration.

# either 1 or 2 hour for an english lesson
english_lesson = VariableDurationTask('EnglishLesson', allowed_durations = [1, 2])

Advanced parameters

Work amount

The work_amount is the total amount of work that the Task must provide. It is set to 0 by default. The work_amount is a dimensionless positive integer value, it can be mapped to any unit according to the physical meaning of the work amount. For example, if the task target is to move small pieces of wood from one point to another, then the work_amount maybe 166000 if 166000 pieces of woods are to be moved. In a maintenance task, if there are 8 screws to unscrew, the UnScrew work_amount will be set to 8.

Temporal priority

The priority of a task is a positive integer that can take any value. It is not bounded. A task with a higher priority will be scheduled earlier than a task with a lower priority. If the solver is requested to optimize the global schedule in terms of task priorities (a “priority objective”) then a task with a high priority may be scheduled before a task with a lower priority.

Optional

All tasks instances are mandatory by default: the solver has to find a solution where all tasks are actually scheduled. However, tasks instances can be turned into optional tasks, by setting the optional flag to True:

# 10mn to clean the table. This is an optional task
clean_the_table_after_meal = FixedDurationTasks('CleanTable', duration=10, optional=True)

An optional task may or may not be scheduled by the solver. It depends on the constraints that bound the scheduling problem.

Task Constraints

ProcessScheduler offers a comprehensive set of predefined temporal task constraints to help you express common scheduling rules efficiently. These constraints allow you to define task-related rules such as “Task A must start exactly at time 4,” “Task B must end simultaneously with Task C,” “Task C should be scheduled precisely 3 periods after Task D,” and more.

Note

Constraints that start with Task* apply to a single task, while those starting with Task**s*** apply to two or more task instances.

Note

All Task constraints can be defined as either mandatory or optional. By default, constraints are mandatory (parameter optional=False). If you set the optional attribute to True, the constraint becomes optional and may or may not apply based on the solver’s discretion. You can force the schedule to adhere to an optional constraint using the task.applied attribute:

pb.add_constraint([task.applied == True])

Single task temporal constraints

These constraints apply to individual tasks.

TaskStartAt

Ensures that a tasks starts precisely at a specified time instant.

TaskStartAt: takes two parameters task and value such as the task starts exactly at the instant \(task.start = value\)

TaskStartAfter

Enforces that a task must start after a given time instant.

TaskStartAfterStrict can be strict lor lax.

TaskEndAt

Ensures that a task ends precisely at a specified time instant.

• TaskEndAt: takes two parameters task and value such as the task ends exactly at the instant value \(task.end = value\)

TaskEndBefore

Requires that a task ends before or at a given time instant.

• TaskEndBeforeStrict can be strict or lax.

Two tasks temporal constraints

These constraints apply to sets of two tasks.

TaskPrecedence

Ensures that one task is scheduled before another. The precedence can be either ‘lax,’ ‘strict,’ or ‘tight,’ and an optional offset can be applied.

The TaskPrecedence class takes two parameters task_1 and task_2 and constraints task_2 to be scheduled after task_1 is completed. The precedence type can either be 'lax' (default, task_2.start >= task_1.end)), 'strict' (task_2.start >= task_1.end)) or 'tight' (task_2.start >= task_1.end, task_2 starts immediately after task_1 is completed). An optional parameter offset can be additionally set.

task_1 = ps.FixedDurationTask('Task1', duration=3)
task_2 = ps.FixedVariableTask('Task2')
pc = TaskPrecedence(task1, task2, kind='tight', offset=2)

constraints the solver to schedule task_2 start exactly 2 periods after task_1 is completed.

TasksStartSynced

Specify that two tasks must start at the same time.

TasksStartSynced takes two parameters task_1 and task_2 such as the schedule must satisfy the constraint \(task_1.start = task_2.start\)

TasksEndSynced

Specify that two tasks must end at the same time.

TasksEndSynced takes two parameters task_1 and task_2 such as the schedule must satisfy the constraint \(task_1.end = task_2.end\)

TasksDontOverlap

Ensures that two tasks should not overlap in time.

TasksDontOverlap takes two parameters task_1 and task_2 such as the task_1 ends before the task_2 is started or the opposite (task_2 ends before task_1 is started)

$n$ tasks temporal constraints

TasksContiguous

Forces a set of tasks to be scheduled contiguously.

TasksContiguous takes a liste of tasks, force the schedule so that tasks are contiguous.

UnorderedTaskGroup

An UnorderedTaskGroup represents a collection of tasks that can be scheduled in any order. This means that the tasks within this group do not have a strict temporal sequence.

OrderedTaskGroup

A set of tasks that can be scheduled in any order, with time bounds

Advanced tasks constraints

ScheduleNTasksInTimeIntervals

Schedules a specific number of tasks within defined time intervals.

Given a list of \(m\) tasks, and a list of time intervals, ScheduleNTasksInTimeIntervals schedule \(N\) tasks among \(m\) in this time interval.

ResourceTasksDistance

Defines constraints on the temporal distance between tasks using a shared resource.

ResourceTasksDistance takes a mandatory attribute distance (integer), an optional time_periods (list of couples of integers e.g. [[0, 1], [5, 19]]). All tasks, that use the given resource, scheduled within the time_periods must have a maximal distance of distance (distance being considered as the time between two consecutive tasks).

Note

If the task(s) is (are) optional(s), all these constraints apply only if the task is scheduled. If the solver does not schedule the task, these constraints does not apply.

Logical task constraints

OptionalTaskConditionSchedule

Creates a constraint that schedules a task based on a specified Boolean condition.

OptionalTaskConditionSchedule creates a constraint that adds a condition for the task to be scheduled. The condition is a z3 BoolRef

OptionalTasksDependency

OptionalTasksDependency takes two optional tasks task_1 and task_2, and ensures that if task_1 is scheduled then that task_2 is forced to be scheduled as well.

ForceScheduleNOptionalTasks

Forces the scheduling of a specified number of optional tasks out of a larger set of optional tasks.

ForceScheduleNOptionalTasks forces \(m\) optional tasks among \(n\) to be scheduled, with \(m \leq n\).

Resource

According to the APICS dictionary, a resource is anything that adds value to a product or service in its creation, production, or delivery.

In the context of ProcessScheduler, a resource is anything that is needed by a task to be successfully processed. In a scheduling problem, resources can be human beings, machines, inventories, rooms or beds in an hotel or an hospital, elevator etc.

ProcessScheduler provides the following classes to deal with resources: Worker, CumulativeWorker

Worker

A Worker is an atomic, countable resource. Being atomic implies that it cannot be further divided into smaller parts, and being countable means it exists in a finite number, available during specific time intervals. The Worker class is ideal for representing entities like machines or humans. A Worker possesses the capacity to process tasks either individually or in collaboration with other workers or resources.

To create a Worker, you can use the following syntax:

john = Worker('JohnBenis')

CumulativeWorker

On the other hand, a CumulativeWorker can simultaneously handle multiple tasks in parallel. The maximum number of tasks that a CumulativeWorker can process concurrently is determined by the size parameter.

For example, you can define a CumulativeWorker like this:

# the machine A can process up to 4 tasks at the same time
machine_A = CumulativeWorker('MachineA', size=4)

Advanced parameters

Productivity

The productivity attribute of a worker represents the amount of work the worker can complete per period. By default, a worker’s productivity is set to 0.

For instance, if you have two drillers, with the first one capable of drilling 3 holes per period and the second one drilling 9 holes per period, you can define them as follows:

driller_1 = Worker('Driller1', productivity=3)
driller_2 = Worker('Driller1', productivity=9)

Note

The workers productivity is used by the solver to satisfy the targeted task work_amount parameter value.

Cost

You can associate cost information with any resource, enabling ProcessScheduler to compute the total cost of a schedule, the cost per resource, or optimize the schedule to minimize costs (see the Objective section for details). There are two ways to define resource costs:

1. Constant Cost Per Period: In this approach, the resource’s cost remains constant over time.

dev_1 = Worker('SeniorDeveloper', cost=ConstantCostPerPeriod(750))

2. Polynomial Cost Function: This method allows you to represent resource costs as a polynomial function of time. It’s particularly useful for modeling costs that are volatile (e.g., oil prices) or time-dependent (e.g., electricity costs). The cost parameter accepts any Python callable object.

def quadratic_time_function(t):
    return (t-20)**2 + 154

cost_function = PolynomialCostFunction(quadratic_time_function)
dev_1 = Worker('AWorker', cost=cost_function)

The worker cost is set to None by default.

You can visualize the cost function using Matplotlib, which provides insights into how the cost evolves over time:

cost_function.plot([0, 200])

Warning

Currently, ProcessScheduler can handle integer numbers only. Then, all the coefficients of the polynomial must be integer numbers. If ever there are floating point numbers, no exception will be raised, but you might face strange results in the cost computation.

Note

The worker cost_per_period is useful to measure the total cost of a resource/a set of resources/a schedule, or to find the schedule that minimizes the total cost of a resource/a set of resources/ a schedule.

Resource assignment

In the context of scheduling, resource assignment is the process of determining which resource or resources should be assigned to a task for its successful processing. ProcessScheduler provides flexible ways to specify resource assignments for tasks, depending on your scheduling needs. A Worker instance can process only one task per time period whereas a CumulativeWorker can process multiple tasks at the same time.

Single resource assignment

For assigning a single resource to a task, you can use the following syntax:

assemble_engine = FixedDurationTask('AssembleCarEngine', duration=10)
john = Worker('JohnBenis')

# the AssembleCarEngine can be processed by JohnBenis ONLY
assemble_engine.add_required_resource(john)

Multiple resources assignment

To assign multiple resources to a single task, you can use the following approach:

paint_car = FixedDurationTask('PaintCar', duration=13)

john = Worker('JohnBenis')
alice = Worker('AliceParker')

# the PaintCar task requires JohnBenis AND AliceParker
paint_engine.add_required_resources([john, alice])

Alternative resource assignment

ProcessScheduler introduces the SelectWorkers class, which allows the solver to decide which resource or resources to assign to a task from a collection of capable workers. You can specify whether the solver should assign exactly n resources, at most n resources, or at least n resources. Let’s consider the following example: 3 drillers are available, a drilling task can be processed by any of one of these 3 drillers. This can be represented as:

drilling_hole = FixedDurationTask('DrillHolePhi10mm', duration=10)
driller_1 = Worker('Driller1')
driller_2 = Worker('Driller2')
driller_3 = Worker('Driller3')
# the DrillHolePhi10mm task can be processed by the Driller1 OR the Driller2 OR the Driller 3
drilling_hole.add_required_resource(SelectWorkers([driller_1,
                                                   driller_2,
                                                   driller_3],
                                    nb_workers_to_select=1,
                                    kind='exact'))

In this case, the solver is instructed to assign exactly one resource from the list of three workers capable of performing the task. The kind parameter can be set to 'exact' (default), 'min', or 'max', depending on your requirements. Additionally, you can specify the number of workers to select with nb_workers_to_select, which can be any integer between 1 (default) and the total number of eligible workers in the list.

These resource assignment options provide flexibility and control over how tasks are allocated to available resources, ensuring efficient scheduling in various use cases.

Resource Constraints

ProcessScheduler provides a set of ready-to-use resource constraints. They allow expressing common rules such as “the resource A is available only from 8 am to 12” etc. There are a set of builtin ready-to-use constraints, listed below.

WorkLoad

The WorkLoad constraint can be used to restrict the number of tasks which are executed during certain time periods.

This constraint applies to one resource, whether it is a single worker or a cumulative worker. It takes the time periods as a python dictionary composed of time intervals (the keys) and an integer number (the capacity). The kind parameter allows to define which kind of restriction applies to the resource: 'exact', 'max' (default value) or 'min'.

c1 = ps.WorkLoad(worker_1, {(0, 6): 2})

In the previous example, the resource worker_1 cannot be scheduled into more than 2 timeslots between instants 0 and 6.

Any number of time intervals can be passed to this class, just extend the timeslots dictionary, e.g.:

c1 = ps.WorkLoad(worker_1, {(0, 6): 2,
                            (19, 21): 6})

The WorkLoad is not necessarily a limitation. Indeed you can specify that the integer number is actually an exact of minimal value to target. For example, if we need the resource worker_1 to be scheduled at least into three time slots between instants 0 and 10, then:

c1 = ps.WorkLoad(worker_1, {(0, 10): 3}, kind='min')

ResourceUnavailable

A ResourceUnavailable applies to a resource and prevent the solver to schedule this resource during certain time periods. This class takes a list of intervals:

worker_1 = ps.Worker('Sylvia')
ca = ps.ResourceUnavailable(worker_1, [(1,2), (6,8)])

The ca instance constraints the resource to be unavailable for 1 period between 1 and 2 instants, and for 2 periods between instants 6 and 8.

Note

This constraint is a special case for the WorkLoad where the number_of_time_slots is set to 0.

DistinctWorkers

A AllDifferentWorkers constraint applies to two SelectWorkers instances, used to assign alternative resources to a task. It constraints the solver to select different workers for each SelectWorkers. For instance:

s1 = ps.SelectWorkers([worker_1, worker_2])
s2 = ps.SelectWorkers([worker_1, worker_2])

could lead the solver to select worker_1 in both cases. Adding the following line:

cs = ps.DistinctWorkers(s1, s2)

let the solver selects the worker_1 for s1 and worker_2 for s2 or the opposite, worker_2 for s1 and worker_1 for s2. The cases where worker_1 is selected by both s1 and s2 or worker_2 by selected by both s1 and s2 are impossible.

SameWorkers

A AllSameWorkers constraint applies to two SelectWorkers instances. It constraints the solver to ensure both different SelectWorkers instances select the same worker. For example:

s1 = ps.SelectWorkers([worker_1, worker_2])
s2 = ps.SelectWorkers([worker_1, worker_2])

could lead the solver to select worker_1 for s1 and worker_2 for s2. Adding the following line:

cs = ps.SameWorkers(s1, s2)

ensures either worker_1 is selected by both s1 and s2, or worker_2 is selected by both s1 and s2.

ResourceTasksDistance

worker_1 = ps.Worker("Worker1")

ps.ResourceTasksDistance(
    worker_1,
    distance=4,
    mode="exact",
    list_of_time_intervals=[[10, 20], [30, 40]])

Buffer

A Buffer is an object where tasks can load or unload a finite number of items. A Buffer can be used to represent a tank, or a temporary buffer of a workshop where manufactured parts are temporarily stored.

A NonConcurrentBuffer is a specific buffers where tasks cannot load and/or unload at the same time. In other words, only one task can access the buffer at a given time. A NonConcurrentBuffer` has three main attributes:

1. the initial_level, i.e. the number of items in the buffer for time t=0,

2. the lower_bound, an optional parameter that sets the minimum number of items in the buffer during the schedule. If ever the solver cannot find a solution where the buffer level is always greater than the lower_bound, it will report an unsatisfiable problem,

3. the upper_bound, an optional parameter that sets the maximum number of items in the buffer during the schedule (in other words, the buffer capacity). If ever the solver cannot find a solution where the buffer level is always lower than the upper_bound, it will report an unsatisfiable problem.

Both initial_level, lower_bound and upper_bound are optional parameters. A NonConcurrentBuffer can be created as follows:

buff1 = ps.NonConcurrentBuffer("Buffer1")
buff2 = ps.NonConcurrentBuffer("Buffer2", initial_state=10)
buff3 = ps.NonConcurrentBuffer("Buffer3", lower_bound=0)
buff4 = ps.NonConcurrentBuffer("Buffer4", upper_bound=20)
buff5 = ps.NonConcurrentBuffer("Buffer5",
                               initial_state=3,
                               lower_bound=0,
                               upper_bound=10)

Buffer constraints

Buffers are loaded/unloaded by tasks. As a consequence, special tasks constraints are used to connect tasks to buffers: TaskUnloadBuffer and TaskLoadBuffer. Both classes take the task instance, the target buffer, and a quantity. Load/Unload constraints can be created as follows:

c1 = ps.TaskUnloadBuffer(task_1, buffer, quantity=3)
c2 = ps.TaskUnloadBuffer(task_2, buffer, quantity=6)
# etc.

Note

There is no limitation on the number of buffers and/or buffer constraints.

Example

Let’s take an example where a task T1 uses a machine M1 to manufacture a part (duration time for this task is 4). It takes one part in a Buffer1 and loads the Buffer2.

machine_1 = ps.Worker('M1')
task_1 = ps.FixedDurationTask('T1', duration=4)
task_1.add_required_resource(machine_1)
# the create buffers
buffer_1 = ps.NonConcurrentBuffer("Buffer1", initial_state=5)
buffer_2 = ps.NonConcurrentBuffer("Buffer2", initial_state=0)
# buffer constraints
c1 = ps.TaskUnloadBuffer(task_1, buffer_1, quantity=1)
c2 = ps.TaskLoadBuffer(task_1, buffer_2, quantity=1)

The graphical output shows the Gantt chart and the evolution of the buffer states along the time line.

First order logic constraints

Builtin constraints may not be sufficient to cover the large number of use-cases user may encounter. Rather than extending more and more the builtin constraints, ProcessScheduler lets you build your own constraints using logical operators, implications and if-then-else statement between builtin constraints or class attributes.

Logical operators

Logical operators and (\(\wedge\)), or (\(\lor\)), xor (\(\oplus\)), not (\(\lnot\)) are provided through the functions and_(), or_(), xor_() and not_().

Note

Take care of the trailing underscore character at the end of the function names. They are necessary because and, or, and not are python keywords that cannot be overloaded. This naming convention may conflict with functions from the operator standard module.

Using builtin task constraints in combination with logical operators enables a rich expressivity. Imagine that you need a task \(t_1\) to NOT start at time 3. At a first glance, you can expect a TaskDontStartAt to fit your needs, but it is not available from the builtin constraints library. The solution is to express this constraint in terms of first order logic, and state that you need the rule:

\[\lnot TaskStartAt(t_1, 3)\]

In python, this gives:

not_(TaskStartAt(t_1, 3)

You can combine/nest any of these operators to express a complex constraint. For example, if you don’t want the task to start at 3, and also you don’t want it to end at 9, then the rule to implement is:

\[\lnot TaskStartAt(t_1,3) \wedge \lnot TaskEndsAt(t_1, 9)\]

In python:

and_([not_(TaskStartAt(t_1, 3)),
      not_(TaskEndAt(t_1, 9))])

In a more general cas, those logical functions can take both task constraints or tasks attributes. For example, the following assertion is possible :

problem.add_constraint(t1.start == t_2.end + t_4.duration)

Logical Implication - Conditional expressions

The logical implication (\(\implies\)) is wrapped by the implies() function. It takes two parameters: a condition, that always has to be True or False, and a list of assertions that are to be implied if the condition is True. For example, the following logical implication:

\[t_2.start = 4 \implies TasksEndSynced(t_3, t_4)\]

is written in Python:

consequence = implies(t_2.start == 4,
                      [TasksEndSynced(t_3, t_4)]
problem.add_constraint(consequence)

Finally, an if/then/else statement is available through the function if_then_else() which takes 3 parameters: a condition and two lists of assertions that applies whether the condition is True or False.

ite = if_then_else(t_2.start == 4,  # condition
                   [TasksEndSynced(t_3, t_4)], # if the condition is True
                   [TasksStartSynced(t_3, t_4)]) # if the condition is False
problem.add_constraint(ite)

Note

The implies() and if_then_else() functions names do not conflict with any other function name from another package, thus dont have any underscore suffix.

Indicator

Indicator class

The Indicator class allows to define a criterion that quantifies the schedule so that it can be compared with other schedules. An Indicator instance results in a conclusion such a ‘the schedule A is better than schedule B because the indicator XXX is greater/lower’.

An Indicator instance is created by passing the indicator name as well as the mathematical expression to compute. For example:

flow_time = Indicator('FlowTime', task1.end + task2.end)
duration_square = Indicator('MinTaskDuration', task1.duration ** 2 + task2.duration ** 2)

Indicator values are computed by the solver, and are part of the solution. If the solution is rendered as a matplotlib Gantt chart, the indicator values are displayed on the upper right corner of the chart.

Indicators can also be bounded, although it is an optional feature. It is useful if the indicator is further be maximized (or minimized) by an optimization solver, in order to reduce the computation time. For example,

indicator1 = Indicator('Example1', task2.start - task1.end, bounds = (0,100)) # If lower and upper bounded
indicator2 = Indicator('Example2', task3.start - task2.end, bounds = (None,100)) # If only upper bounded
indicator3 = Indicator('Example3', task4.start - task3.end, bounds = (0,None)) # If only lower bounded

Note

There is no limit to the number of Indicators defined in the problem. The mathematical expression must be expressed in a polynomial form and using the Sqrt() function. Any other advanced mathematical functions such as exp(), sin(), etc. is not allowed because not supported by the solver.

Builtin indicators : Resource cost, utilization, number of tasks assigned

Two usual indicators are available : the utilization and cost of a resource (or a list of resources).

Use the add_indicator_resource_utilization() method to insert a cost computation to your schedule. This method takes a list of resources and compute the total cost (sum of productivity * duration for all resources). The result is a percentage: an utilization of 100% means that the resource is assigned 100% of the schedule timeline. In the following example, the indicator reports the utilization percentage of the worker_1().

utilization_res_1 = problem.add_indicator_resource_utilization(worker_1)

The add_indicator_resource_cost() method returns the total cost of a resource (or a list of resource). It is computed using each cost function defined for each resource. In the following example, the indicator cost_ind() is the total cost for both ressources worker_1() and worker_2().

cost_ind = problem.add_indicator_resource_cost([worker_1, worker_2])

At last, the add_indicator_number_tasks_assigned() method returns the number of tasks assigned to a resource after the schedule is completed.

problem.add_indicator_number_tasks_assigned(worker)

Optimization

ProcessScheduler is able to compute optimized schedules according to one (single optimization) or any number (multi-objectives) of objectives.

Objective

An objective is a target value for an Indicator or any of the variables defined in the scheduling problem:

• if the target value is known, then the objective can either be ExactObjective, MinimumObjective or MaximumObjective,

• it the target value is unknown but you want to find a minimal or maximal value, the the objective can be the result from an optimization resolution, MaximizeObjective or MinimizeObjective.

Warning

Using MaximizeObjective or MinimizeObjective classes turn the problem into an optimization problem. This will result in heavier computations and, thus, a longer time for the problem to be solved.

For example, if you need to optimize the utilization of a resource (bounded up to 100% within the problem horizon),

# first create the indicator
utilization_res_1 = problem.add_indicator_resource_utilization(worker_1)
# the tell the solver to optimize this value
ps.MaximizeObjective("MaximizeResource1Utilization", utilization_res_1)

Builtin optimization objectives

For any problem, the following builtin objectives are available:

• add_objective_makespan(): minimize the schedule horizon,

• add_objective_resource_utilization(): maximize resource occupation,

• add_objective_resource_cost(): minimize the total cost for selected resource(s),

• add_objective_priorities(): minimize total priority indicator (tasks with high priorities will be scheduled before tasks with lower priorities, under the condition however that all constraints are satisfied),

• add_objective_start_earliest(): minimize the start time of the last task to be scheduled,

• add_objective_start_latest(): maximize the start time of the first task to be scheduled,

• add_objective_flowtime(): minimize flowtime.

• add_objective_flowtime_single_resource(): minimize flowtime of a single resource on a specific time interval

Available solvers : incremental and optimize

The default optimization solver is incremental(). After a solution is found, the solver will run again and again to find a better solution untill the maximum allowed time is reached. If you provide a small max_time value, the solver will exit to the last found value, but there may be a better value. In that case, just increase the max_time and run again the solver.

solver = ps.SchedulingSolver(pb, max_time=300)  # 300s is 5 minutes
solution = solver.solve()

The other available solver is called optimize(), which use the builtin optsmt z3-solver. The computation cannot be interrupted, so be careful if the problem to solve involves many tasks/resources. However, the :func`optimize` is guaranteed to return the optimal value.

solver = ps.SchedulingSolver(pb, optimizer="optimize")  # 300s is 5 minutes
solution = solver.solve()

Single objective optimization

Imagine you need to schedule one specific task task_1 the later. After you defined the task as usual, then create the objective and set the optimization target:

pb = ps.SchedulingProblem('SingleObjective1', horizon=20)
task_1 = ps.FixedDurationTask('task1', duration = 3)
indicator_1 = ps.Indicator('Task1End', task_1.end)
ps.MaximizeObjective('MaximizeTask1End', indicator_1)
ps.SchedulingSolver(pb).solve()

The expected value for the indicator_1 maximization is 20. After running the script, you may get the following output:

Solver type:
===========
-> Standard SAT/SMT solver
Incremental optimizer:
======================
Found better value: 3 elapsed time:0.000s
Checking better value >3
Found better value: 4 elapsed time:0.071s
Checking better value >4
[...]
Checking better value >18
Found better value: 19 elapsed time:0.074s
Checking better value >19
Found better value: 20 elapsed time:0.074s
Checking better value >20
No solution can be found for problem MultiObjective2.
Reason: Unsatisfiable problem: no solution exists
Found optimum 20. Stopping iteration.
total number of iterations: 19
value: 20
MultiObjective2 satisfiability checked in 0.07s

The solver returns the expected result.

Multiple objective optimization using the incremental solver (default)

ProcessScheduler can deal with multiple objectives optimization. There is no limitation regarding the number of objectives.

Imagine you need to schedule two tasks task_1 and task_2 the later. After you defined the task as usual, then create the objective and set the optimization target:

pb = ps.SchedulingProblem('MultiObjective1', horizon=20)
task_1 = ps.FixedDurationTask('task1', duration = 3)
task_2 = ps.FixedDurationTask('task2', duration = 3)
indicator_1 = ps.Indicator('Task1End', task_1.end)
indicator_2 = ps.Indicator('Task2End', task_2.end)
ps.MaximizeObjective('MaximizeTask1End', indicator_1)
ps.MaximizeObjective('MaximizeTask2End', indicator_2)
solution = ps.SchedulingSolver(pb).solve()
print(solution)

After running the script, you may get the following output:

[...]
{
"horizon": 20,
"indicators": {
    "EquivalentIndicator": -40,
    "Task1End": 20,
    "Task2End": 20
},
[...]

The solver gives the expected result. Note that an EquivalentIndicator is built from both indicators. A maximization problem is always turned into a minimization problem (the equivalent indicator is negative).

Weighted objectives

In the previous example, if we add a constraint between tasks task_1 and task_2, then both tasks end may not be independent from each other. For example, let’s add the following constraint:

pb.add_constraint(task_1.end == 20 - task_2.start)

This looks like a kind of balance: the later task_1 is scheduled, the sooner task_2 is scheduled. We can leave both optimizations enabled, but the solver has to know what to do with these conflicting objectives, and especially what is there relative weight.

Note

MaimizeObjective and MinimizeObjective have an optional weight parameter set by default to 1.0. The higher this value, the more important the objective.

The python script will look like

import processscheduler as ps
pb = ps.SchedulingProblem('MultiObjective2', horizon=20)
task_1 = ps.FixedDurationTask('task1', duration = 3)
task_2 = ps.FixedDurationTask('task2', duration = 3)
pb.add_constraint(task_1.end == 20 - task_2.start)
indicator_1 = ps.Indicator('Task1End', task_1.end)
indicator_2 = ps.Indicator('Task2End', task_2.end)

ps.MaximizeObjective('MaximizeTask1End', indicator_1, weight=1.)
ps.MaximizeObjective('MaximizeTask2End', indicator_2, weight=1.)
solution = ps.SchedulingSolver(pb).solve()
print(solution)

"indicators": {
    "EquivalentIndicator": -23,
    "Task1End": 20,
    "Task2End": 3
},

The solver decides to schedule the Task1 at the end of the timeline. Let’s change the relative weights so that the second objective is considered as more important:

ps.MaximizeObjective('MaximizeTask1End', indicator_1, weight=1.)
# the second one is ten times more important
ps.MaximizeObjective('MaximizeTask2End', indicator_2, weight=10.)

This lead the solver to another solution:

"indicators": {
    "EquivalentIndicator": -203,
    "Task1End": 3,
    "Task2End": 20
},

Multiple objective optimization using the optimize solver (default)

Lexicon priority ('lex', default)

The solver optimizes the first objective, then the second one while keeping the first value, then the third one keeping both previous values etc.

In the previous example, the first objective to be optimized will be the end of task_1, obviously 20. Then, this value being fixed, there’s no other solution than start of the second task is 0, then task_2 end will be 3.

ps.SchedulingSolver(pb, optimizer=optimize, optimize_priority='lex').solve()

And the output

Optimization results:
=====================
    ->Objective priority specification: lex
    ->Objective values:
        ->Indicator_Task1End(max objective): 20
        ->Indicator_Task2End(max objective): 3

Lexicon priority ('box')

The optimization solver breaks the dependency between objectives and look for the maximum (resp. minimum) value that can be achieved for each objective.

In the previous example, the maximum of task_1end can be 20, and the maximum of task_2.end can also be 20, but not at the same time. The box priority then gives an information about the values that can be reached.

ps.SchedulingSolver(pb, optimizer=optimize, optimize_priority='box').solve()

And the output

Optimization results:
=====================
    ->Objective priority specification: lex
    ->Objective values:
        ->Indicator_Task1End(max objective): 20
        ->Indicator_Task2End(max objective): 20

Note

In the box mode, both objectives may not be reached simultaneously, the solver will give anyway a solution that satisfies all constraints (by default the solution obtained from the lexicon mode).

Pareto priority ('pareto')

The optimization solver suggests a new solution each time the solve() method is called. This allows traversing all solutions. Indeed we can have the task_1 end to 20 and task_2 end 3, but also the task_1 end to 19 and task_2 end to 4 etc. These all are solutions for the optimization problem.

The python code has to be slightly modified:

solver = ps.SchedulingSolver(pb, optimizer=optimize, optimize_priority='pareto')
while solver.solve():
    pass

And the output will be:

Optimization results:
=====================
    ->Objective priority specification: pareto
    ->Objective values:
        ->Indicator_Task1End(max objective): 20
        ->Indicator_Task2End(max objective): 3
SAT computation time:
=====================
    MultiObjective2 satisfiability checked in 0.00s
Optimization results:
=====================
    ->Objective priority specification: pareto
    ->Objective values:
        ->Indicator_Task1End(max objective): 19
        ->Indicator_Task2End(max objective): 4
SAT computation time:
=====================
    MultiObjective2 satisfiability checked in 0.00s
Optimization results:
=====================
    ->Objective priority specification: pareto
    ->Objective values:
        ->Indicator_Task1End(max objective): 18
        ->Indicator_Task2End(max objective): 5
[...]

Here you have 18 different solutions. You can add a test to the loop to stop the iteration as soon as you’re ok with the solution.

Problem solving

Solving a scheduling problem involves the SchedulingSolver class.

Define the solver

A SchedulingSolver instance takes a SchedulingProblem instance:

solver = SchedulingSolver(scheduling_problem_instance)

It takes the following optional arguments:

• debug: False by default, if set to True will output many useful information.

• max_time: in seconds, the maximal time allowed to find a solution. Default is 60s.

• parallel: boolean False by default, if True force the solver to be executed in multithreaded mode. It might be quicker. It might not.

• random_values: a boolean, default to False. If set to True, enable a builtin generator to set random initial values. By setting this attribute to True, one expects the solver to give a different solution each time it is called.

• logics: a string, None by default. Can be set to any of the supported z3 logics, “QF_IDL”, “QF_LIA”, etc. see https://smtlib.cs.uiowa.edu/logics.shtml. By default (logics set to None), the solver tries to find the best logics, but there can be significant improvements by setting a specific logics (“QF_IDL” or “QF_UFIDL” seems to give the best performances).

• verbosity: an integer, 0 by default. 1 or 2 increases the solver verbosity. TO be used in a debugging or inspection purpose.

Solve

Just call the solve() method. This method returns a Solution instance.

solution = solver.solve()

Running the solve() method returns can either fail or succeed, according to the 4 following cases:

1. The problem cannot be solved because some constraints are contradictory. It is called “Unsatisfiable”. The solve() method returns False. For example:

TaskStartAt(cook_the_chicken, 2)
TaskStartAt(cook_the_chicken, 3)

It is obvious that these constraints cannot be both satisfied.

2. The problem cannot be solved for an unknown reason (the satisfiability of the set of constraints cannot be computed). The solve() method returns False.

3. The solver takes too long to complete and exceeds the allowed max_time. The solve() method returns False.

4. The solver successes in finding a schedule that satisfies all the constraints. The solve() method returns the solution as a JSON dictionary.

Note

If the solver fails to give a solution, increase the max_time (case 3) or remove some constraints (cases 1 and 2).

Find another solution

The solver may schedule:

• one solution among many, in the case where there is no optimization,

• the best possible schedule in case of an optimization issue.

In both cases, you may need to check a different schedule that fits all the constraints. Use the find_another_solution() method and pass the variable you would want the solver to look for another solution.

Note

Before requesting another solution, the solve() method has first to be executed, i.e. there should already be a current solution.

You can pass any variable to the find_another_solution() method: a task start, a task end, a task duration, a resource productivity etc.

For example, there are 5 different ways to schedule a FixedDurationTask with a duration=2 in an horizon of 6. The default solution returned by the solver is:

problem = ps.SchedulingProblem('FindAnotherSolution', horizon=6)
solutions =[]
task_1 = ps.FixedDurationTask('task1', duration=2)
problem.add_task(task_1)
solver = ps.SchedulingSolver(problem)
solution = solver.solve()
print("Solution for task_1.start:", task_1.scheduled_start)

Solution for task_1.start: 0

Then, we can request for another solution:

solution = solver.find_another_solution(task_1.start)
if solution is not None:
    print("New solution for task_1.start:", solution.tasks[task_1.name].start)

Solution for task_1.start: 1

You can recursively call find_another_solution() to find all possible solutions, until the solver fails to return a new one.

Run in debug mode

If the debug attribute is set to True, the z3 solver is run with the unsat_core option. This will result in a much longer computation time, but this will help identifying the constraints that conflict. Because of this higher consumption of resources, the debug flag should be used only if the solver fails to find a solution.

Render to a Gantt chart

Call the render_gantt_matplotlib() to render the solution as a Gantt chart. The time line is from 0 to horizon value, you can choose to render either Task or Resource (default).

solution = solver.solve()
if solution is not None:
    solution.render_gantt_matplotlib()  # default render_mode is 'Resource'
    # a second gantt chart, in 'Task' mode
    solution.render_gantt_matplotlib(render_mode='Task')

Call the render_gantt_plotly() to render the solution as a Gantt chart using plotly. Take care that plotly rendering needs real timepoints (set at least delta_time at the problem creation).

solution = solver.solve()
if solution is not None:
    # default render_mode is 'Resource'
    solution.render_gantt_plotly(sort="Start", html_filename="index.html")
    # a second gantt chart, in 'Task' mode
    solution.render_gantt_plotly(render_mode='Task')

Use case: flowshop scheduling

This example is based on the paper from Tao et al. (2015), where authors present an introduction example. In a flow shop problem, a set of \(n\) jobs has to be processed on \(m\) different machines in the same order. Job \(j\), \(j=1,2,...,n\) is processed on machines \(i\), \(i=1,2,..,m\), with a nonnegative processing time \(p(i,j)\) and a release date \(r_j\), which is the earliest time when the job is permitted to process. Each machine can process at most one job and each job can be handled by at most one machine at any given time. The machine processes the jobs in a first come, first served manner. The goal is to determine a job sequence that minimizes the makespan.

The problem statement is:

The following solution is reported by the authors (order J1, J3, J4, J2, scheduled horizon: 29):

In this notebook, we try to reproduce the results reported by the authors.

Reference

Tao Ren, Meiting Guo, Lin Lin, Yunhui Miao, “A Local Search Algorithm for the Flow Shop Scheduling Problem with Release Dates”, Discrete Dynamics in Nature and Society, vol. 2015, Article ID 320140, 8 pages, 2015. https://doi.org/10.1155/2015/320140

Imports

import processscheduler as ps

%config InlineBackend.figure_formats = ['svg']

Create the scheduling problem

The total horizon is unknwown, leave it empty and only set the problem name.

flow_shop_problem = ps.SchedulingProblem("FlowShop")

Create the 3 machines M1, M2 and M3

M3 = ps.Worker("M3")
M2 = ps.Worker("M2")
M1 = ps.Worker("M1")

Create jobs J1, J2, J3 and J4 - related tasks

# J1
J11 = ps.FixedDurationTask("J11", duration=2)
J12 = ps.FixedDurationTask("J12", duration=5)
J13 = ps.FixedDurationTask("J13", duration=6)

# J2
J21 = ps.FixedDurationTask("J21", duration=1)
J22 = ps.FixedDurationTask("J22", duration=5)
J23 = ps.FixedDurationTask("J23", duration=7)

# J3
J31 = ps.FixedDurationTask("J31", duration=1)
J32 = ps.FixedDurationTask("J32", duration=4)
J33 = ps.FixedDurationTask("J33", duration=1)

# J4
J41 = ps.FixedDurationTask("J41", duration=3)
J42 = ps.FixedDurationTask("J42", duration=4)
J43 = ps.FixedDurationTask("J43", duration=7)

Assign resources

One machine per task.

J11.add_required_resource(M1)
J12.add_required_resource(M2)
J13.add_required_resource(M3)

J21.add_required_resource(M1)
J22.add_required_resource(M2)
J23.add_required_resource(M3)

J31.add_required_resource(M1)
J32.add_required_resource(M2)
J33.add_required_resource(M3)

J41.add_required_resource(M1)
J42.add_required_resource(M2)
J43.add_required_resource(M3)

Constraint: release dates

r1 = 0
r2 = 9
r3 = 2
r4 = 7

ps.TaskStartAfter(J11, r1)
ps.TaskStartAfter(J12, r1)
ps.TaskStartAfter(J13, r1)

ps.TaskStartAfter(J21, r2)
ps.TaskStartAfter(J22, r2)
ps.TaskStartAfter(J23, r2)

ps.TaskStartAfter(J31, r3)
ps.TaskStartAfter(J32, r3)
ps.TaskStartAfter(J33, r3)

ps.TaskStartAfter(J41, r4)
ps.TaskStartAfter(J42, r4)
ps.TaskStartAfter(J43, r4)

TaskStartAfter_be270e3a(<class 'processscheduler.task_constraint.TaskStartAfter'>)
1 assertion(s):
J43_start >= 7

Constraints: precedences

All jobs should be scheduled in the same ordre on each machine. The constraint is expressed as following: all J2 tasks must be scheduled before Or after J2 tasks, AND all J3 tasks must be scheduled before OR oafter J1 tasks etc.

J1 = [J11, J12, J13]
J2 = [J21, J22, J23]
J3 = [J31, J32, J33]
J4 = [J41, J42, J43]

# we need to combinations function of the itertools module,
# to compute all pairs from the list of jobs.
from itertools import combinations

for Ja, Jb in combinations([J1, J2, J3, J4], 2):
    befores = []
    afters = []
    for i in range(3):
        Ja_before_Jb = ps.TaskPrecedence(Ja[i], Jb[i])
        Ja_after_Jb = ps.TaskPrecedence(Jb[i], Ja[i])
        befores.append(Ja_before_Jb)
        afters.append(Ja_after_Jb)
    xor_operation = ps.xor_([ps.and_(befores), ps.and_(afters)])
    flow_shop_problem.add_constraint(xor_operation)

Add a makespan objective

makespan_obj = flow_shop_problem.add_objective_makespan()

Solution, plot the schedule

solver = ps.SchedulingSolver(flow_shop_problem)
solution = solver.solve()
solution.render_gantt_matplotlib(fig_size=(10, 5), render_mode="Resource")

Solver type:
===========
        -> Standard SAT/SMT solver
Incremental optimizer:
======================
        Found value: 28 elapsed time:0.009s
        Checking better value < 28
        Found value: 22 elapsed time:0.009s
        Checking better value < 22
        Found value: 21 elapsed time:0.009s
        Checking better value < 21
        Can't find a better solution for problem FlowShop.

        Found optimum 21. Stopping iteration.
        total number of iterations: 4
        value: 21
        FlowShop satisfiability checked in 0.01s

We confirm the job sort from Tao et al. (2015) (J1 then J3, J4 and finally J2). The horizon is here only 21.

Use case: formula one pitstop

This example is based on the DailyMail blog entry https://www.dailymail.co.uk/sport/formulaone/article-4401632/Formula-One-pit-stop-does-crew-work.html where a nice image shows 21 people changing the 4 tires of a Formula 1 Ferrari. In this example, only 16 out 21 people are represented. This notebook can be tested online at mybinder.org

from IPython.display import YouTubeVideo

YouTubeVideo("aHSUp7msCIE", width=800, height=300)

Imports

import processscheduler as ps

%config InlineBackend.figure_formats = ['svg']

Create the scheduling problem

The total horizon is not knwown, leave it empty and only set the problem name.

change_tires_problem = ps.SchedulingProblem("ChangeTires")

Create the 16 available resources

Each people in and around the car is represented as a worker.

nb_lifters = 2
nb_gunners = 4
nb_tyre_handlers = 8
nb_stabilizers = 2

# Lift tasks
lifters = [ps.Worker("JackOperator%i" % (i + 1)) for i in range(nb_lifters)]
gunners = [ps.Worker("Gunner%i" % (i + 1)) for i in range(nb_gunners)]
tyre_handlers = [ps.Worker("Handler%i" % (i + 1)) for i in range(nb_tyre_handlers)]
stabilizers = [ps.Worker("Stabilizer%i" % (i + 1)) for i in range(nb_stabilizers)]

Create tasks and assign resources

One period is mapped to one second. For example, if lifting the rear take 2sec then the duration will be set to 2.

# lift tasks and lifters
# both lift tasks can be processed by any one of the lifters
lift_rear_up = ps.FixedDurationTask("LiftRearUp", duration=2)
lift_front_up = ps.FixedDurationTask("LiftFrontUp", duration=2)
lift_rear_up.add_required_resource(lifters[0])
lift_front_up.add_required_resource(lifters[1])

lift_rear_down = ps.FixedDurationTask("LiftRearDown", duration=2)
lift_front_down = ps.FixedDurationTask("LiftFrontDown", duration=2)
lift_rear_down.add_required_resource(lifters[0])
lift_front_down.add_required_resource(lifters[1])

# unscrew tasks
unscrew_front_left_tyre = ps.FixedDurationTask("UnScrewFrontLeftTyre", duration=2)
unscrew_front_right_tyre = ps.FixedDurationTask("UnScrewFrontRightTyre", duration=2)
unscrew_rear_left_tyre = ps.FixedDurationTask("UnScrewRearLeftTyre", duration=2)
unscrew_rear_right_tyre = ps.FixedDurationTask("UnScrewRearRightTyre", duration=2)

gunner_unscrew_front_left_tyre = ps.SelectWorkers(gunners, 1)
unscrew_front_left_tyre.add_required_resource(gunner_unscrew_front_left_tyre)

gunner_unscrew_front_right_tyre = ps.SelectWorkers(gunners, 1)
unscrew_front_right_tyre.add_required_resource(gunner_unscrew_front_right_tyre)

gunner_unscrew_rear_left_tyre = ps.SelectWorkers(gunners, 1)
unscrew_rear_left_tyre.add_required_resource(gunner_unscrew_rear_left_tyre)

gunner_unscrew_rear_right_tyre = ps.SelectWorkers(gunners, 1)
unscrew_rear_right_tyre.add_required_resource(gunner_unscrew_rear_right_tyre)

# screw tasks and gunners
screw_front_left_tyre = ps.FixedDurationTask("ScrewFrontLeftTyre", duration=2)
screw_front_right_tyre = ps.FixedDurationTask("ScrewFrontRightTyre", duration=2)
screw_rear_left_tyre = ps.FixedDurationTask("ScrewRearLeftTyre", duration=2)
screw_rear_right_tyre = ps.FixedDurationTask("ScrewRearRightTyre", duration=2)

gunner_screw_front_left_tyre = ps.SelectWorkers(gunners)
screw_front_left_tyre.add_required_resource(gunner_screw_front_left_tyre)

gunner_screw_front_right_tyre = ps.SelectWorkers(gunners)
screw_front_right_tyre.add_required_resource(gunner_screw_front_right_tyre)

gunner_screw_rear_left_tyre = ps.SelectWorkers(gunners)
screw_rear_left_tyre.add_required_resource(gunner_screw_rear_left_tyre)

gunner_screw_rear_right_tyre = ps.SelectWorkers(gunners)
screw_rear_right_tyre.add_required_resource(gunner_screw_rear_right_tyre)

# tires OFF and handlers
front_left_tyre_off = ps.FixedDurationTask("FrontLeftTyreOff", duration=2)
front_right_tyre_off = ps.FixedDurationTask("FrontRightTyreOff", duration=2)
rear_left_tyre_off = ps.FixedDurationTask("RearLeftTyreOff", duration=2)
rear_right_tyre_off = ps.FixedDurationTask("RearRightTyreOff", duration=2)

for tyre_off_task in [
    front_left_tyre_off,
    front_right_tyre_off,
    rear_left_tyre_off,
    rear_right_tyre_off,
]:
    tyre_off_task.add_required_resource(ps.SelectWorkers(tyre_handlers))

# tires ON and handlers, same as above
front_left_tyre_on = ps.FixedDurationTask("FrontLeftTyreOn", duration=2)
front_right_tyre_on = ps.FixedDurationTask("FrontRightTyreOn", duration=2)
rear_left_tyre_on = ps.FixedDurationTask("RearLeftTyreOn", duration=2)
rear_right_tyre_on = ps.FixedDurationTask("RearRightTyreOn", duration=2)

for tyre_on_task in [
    front_left_tyre_on,
    front_right_tyre_on,
    rear_left_tyre_on,
    rear_right_tyre_on,
]:
    tyre_on_task.add_required_resource(ps.SelectWorkers(tyre_handlers))

Stabilizers start their job as soon as the car is stopped until the end of the whole activity.

stabilize_left = ps.VariableDurationTask("StabilizeLeft")
stabilize_right = ps.VariableDurationTask("StabilizeRight")

stabilize_left.add_required_resource(stabilizers[0])
stabilize_right.add_required_resource(stabilizers[1])

ps.TaskStartAt(stabilize_left, 0)
ps.TaskStartAt(stabilize_right, 0)

ps.TaskEndAt(stabilize_left, change_tires_problem.horizon)
ps.TaskEndAt(stabilize_right, change_tires_problem.horizon)

TaskEndAt_1147c857(<class 'processscheduler.task_constraint.TaskEndAt'>)
1 assertion(s):
StabilizeRight_end == horizon

Task precedences

# front left tyre operations
fr_left = [
    unscrew_front_left_tyre,
    front_left_tyre_off,
    front_left_tyre_on,
    screw_front_left_tyre,
]
for i in range(len(fr_left) - 1):
    ps.TaskPrecedence(fr_left[i], fr_left[i + 1])
# front right tyre operations
fr_right = [
    unscrew_front_right_tyre,
    front_right_tyre_off,
    front_right_tyre_on,
    screw_front_right_tyre,
]
for i in range(len(fr_right) - 1):
    ps.TaskPrecedence(fr_right[i], fr_right[i + 1])
# rear left tyre operations
re_left = [
    unscrew_rear_left_tyre,
    rear_left_tyre_off,
    rear_left_tyre_on,
    screw_rear_left_tyre,
]
for i in range(len(re_left) - 1):
    ps.TaskPrecedence(re_left[i], re_left[i + 1])
# front left tyre operations
re_right = [
    unscrew_rear_right_tyre,
    rear_right_tyre_off,
    rear_right_tyre_on,
    screw_rear_right_tyre,
]
for i in range(len(re_right) - 1):
    ps.TaskPrecedence(re_right[i], re_right[i + 1])

# all un screw operations must start after the car is lift by both front and rear jacks
for unscrew_tasks in [
    unscrew_front_left_tyre,
    unscrew_front_right_tyre,
    unscrew_rear_left_tyre,
    unscrew_rear_right_tyre,
]:
    ps.TaskPrecedence(lift_rear_up, unscrew_tasks)
    ps.TaskPrecedence(lift_front_up, unscrew_tasks)

# lift down operations must occur after each screw task is completed
for screw_task in [
    screw_front_left_tyre,
    screw_front_right_tyre,
    screw_rear_left_tyre,
    screw_rear_right_tyre,
]:
    ps.TaskPrecedence(screw_task, lift_rear_down)
    ps.TaskPrecedence(screw_task, lift_front_down)

First solution, plot the schedule

solver = ps.SchedulingSolver(change_tires_problem)
solution_1 = solver.solve()
solution_1.render_gantt_matplotlib(fig_size=(10, 5), render_mode="Resource")

Solver type:
===========
        -> Standard SAT/SMT solver
Total computation time:
=====================
        ChangeTires satisfiability checked in 0.09s

Second solution: add a makespan objective

Obviously, the former solution is not the best solution, not sure Ferrari will win this race ! The whole “change tires” activity must be as short as possible, so let’s add a makespan objective, i.e. a constraint that minimizes the schedule horizon.

# add makespan objective
change_tires_problem.add_objective_makespan()

solver_2 = ps.SchedulingSolver(change_tires_problem)
solution_2 = solver_2.solve()
solution_2.render_gantt_matplotlib(fig_size=(9, 5), render_mode="Task")

Solver type:
===========
        -> Standard SAT/SMT solver
Incremental optimizer:
======================
        Found value: 18 elapsed time:0.080s
        Checking better value < 18
        Found value: 17 elapsed time:0.172s
        Checking better value < 17
        Found value: 16 elapsed time:0.175s
        Checking better value < 16
        Found value: 15 elapsed time:0.181s
        Checking better value < 15
        Found value: 14 elapsed time:0.185s
        Checking better value < 14
        Found value: 13 elapsed time:0.206s
        Checking better value < 13
        Found value: 12 elapsed time:0.211s
        Checking better value < 12
        Can't find a better solution for problem ChangeTires.

        Found optimum 12. Stopping iteration.
        total number of iterations: 8
        value: 12
        ChangeTires satisfiability checked in 0.21s

Third solution: constraint workers

This is not the best possible solution. Indeed, we can notice that the Gunner2 unscrews the RearRightTyre and screw the RearLeft tyre. We cannot imagine that a solution where gunners turn around the car is acceptable. There are two solutions to fix the schedule: - let the gunner be able to turn around the car, and add a “Move” task with a duration that represent the time necessary to move from one tyre to the other, - constraint the worker to screw the same tyre he unscrewed. Let’s go this way

ps.SameWorkers(gunner_unscrew_front_left_tyre, gunner_screw_front_left_tyre)
ps.SameWorkers(gunner_unscrew_front_right_tyre, gunner_screw_front_right_tyre)
ps.SameWorkers(gunner_unscrew_rear_left_tyre, gunner_screw_rear_left_tyre)
ps.SameWorkers(gunner_unscrew_rear_right_tyre, gunner_screw_rear_right_tyre)

solver_3 = ps.SchedulingSolver(change_tires_problem)
solution_3 = solver_3.solve()
solution_3.render_gantt_matplotlib(fig_size=(9, 5), render_mode="Task")

Solver type:
===========
        -> Standard SAT/SMT solver
Incremental optimizer:
======================
        Found value: 22 elapsed time:0.076s
        Checking better value < 22
        Found value: 21 elapsed time:0.139s
        Checking better value < 21
        Found value: 20 elapsed time:0.158s
        Checking better value < 20
        Found value: 19 elapsed time:0.195s
        Checking better value < 19
        Found value: 18 elapsed time:0.198s
        Checking better value < 18
        Found value: 17 elapsed time:0.201s
        Checking better value < 17
        Found value: 16 elapsed time:0.204s
        Checking better value < 16
        Found value: 15 elapsed time:0.212s
        Checking better value < 15
        Found value: 14 elapsed time:0.216s
        Checking better value < 14
        Found value: 13 elapsed time:0.222s
        Checking better value < 13
        Found value: 12 elapsed time:0.226s
        Checking better value < 12
        Can't find a better solution for problem ChangeTires.

        Found optimum 12. Stopping iteration.
        total number of iterations: 12
        value: 12
        ChangeTires satisfiability checked in 0.23s

This is much better !

Use case: software development

To illustrate the way to use ProcessScheduler, let’s imagine the simple following use case: the developmenent of a scheduling software intended for end-user. The software is developed using Python, and provides a modern Qt GUI. Three junior developers are in charge (Elias, Louis, Elise), under the supervision of their project manager Justine. The objective of this document is to generate a schedule of the different developmenent tasks to go rom the early design stages to the first software release. This notebook can tested online at mybinder.org

Step 1. Import the module

The best way to import the processscheduler module is to choose an alias import. Indeed, a global import should generate name conflicts. Here, the ps alias is used.

import processscheduler as ps
from datetime import timedelta, datetime

%config InlineBackend.figure_formats = ['svg']

Step 2. Create the scheduling problem

The SchedulingProblem has to be defined. The problem must have a name (it is a mandatory argument). Of course you can create as many problems (i.e; SchedulingProblem instances), for example if you need to compare two or more different schedules.

problem = ps.SchedulingProblem(
    "SoftwareDevelopment", delta_time=timedelta(days=1), start_time=datetime.now()
)

Step 3. Create tasks instances

The SchedulingProblem has to be defined. The problem must have a name (it is a mandatory argument). Of course you can create as many problems (i.e SchedulingProblem instances) as needed, for example if you need to compare two or more different schedules. In this example, one period is one day.

preliminary_design = ps.FixedDurationTask("PreliminaryDesign", duration=1)  # 1 day
core_development = ps.VariableDurationTask("CoreDevelopmenent", work_amount=10)
gui_development = ps.VariableDurationTask("GUIDevelopment", work_amount=15)
integration = ps.VariableDurationTask("Integration", work_amount=3)
tests_development = ps.VariableDurationTask("TestDevelopment", work_amount=8)
release = ps.ZeroDurationTask("ReleaseMilestone")

Step 4. Create tasks time constraints

Define precedences or set start and end times

ps.TaskStartAt(preliminary_design, 0)
ps.TaskPrecedence(preliminary_design, core_development)
ps.TaskPrecedence(preliminary_design, gui_development)
ps.TaskPrecedence(gui_development, tests_development)
ps.TaskPrecedence(core_development, tests_development)
ps.TaskPrecedence(tests_development, integration)
ps.TaskPrecedence(integration, release)

TaskPrecedence_d7a12226(<class 'processscheduler.task_constraint.TaskPrecedence'>)
1 assertion(s):
Integration_end <= ReleaseMilestone_start

Step 5. Create resources

Define all resources required for all tasks to be processed, including productivity and cost_per_period.

elias = ps.Worker(
    "Elias", productivity=2, cost=ps.ConstantCostPerPeriod(600)
)  # cost in $/day
louis = ps.Worker("Louis", productivity=2, cost=ps.ConstantCostPerPeriod(600))
elise = ps.Worker("Elise", productivity=3, cost=ps.ConstantCostPerPeriod(800))
justine = ps.Worker("Justine", productivity=2, cost=ps.ConstantCostPerPeriod(1200))

Step 6. Assign resources to tasks

preliminary_design.add_required_resources([elias, louis, elise, justine])
core_development.add_required_resources([louis, elise])
gui_development.add_required_resources([elise])
tests_development.add_required_resources([elias, louis])
integration.add_required_resources([justine])
release.add_required_resources([justine])

Step 7. Add a total cost indicator

This resource cost indicator computes the total cost of selected resources.

cost_ind = problem.add_indicator_resource_cost([elias, louis, elise, justine])

Step 8. Solve and plot using plotly

# solve
solver = ps.SchedulingSolver(problem)
solution = solver.solve()

Solver type:
===========
        -> Standard SAT/SMT solver
Total computation time:
=====================
        SoftwareDevelopment satisfiability checked in 0.01s

if solution:
    solution.render_gantt_plotly()

API documentation

Problem

Problem definition and related classes.

class processscheduler.problem.SchedulingProblem(name, horizon=None, delta_time=None, start_time=None, end_time=None)

Bases: _NamedUIDObject

A scheduling problem

Parameters:

• name (str) – the problem name, a string type

• horizon (Optional[int]) – an optional integer, the final instant of the timeline

• delta_time (Optional[timedelta]) – an optional timedelta object

• start_time (Optional[datetime]) – an optional datetime object

• end_time (Optional[datetime]) – an optional datetime object

• datetime_format – an optional string

add_indicator_number_tasks_assigned(resource)

compute the number of tasks as resource is assigned

Parameters:

resource (Resource) –

add_indicator_resource_cost(list_of_resources)

compute the total cost of a set of resources

Parameters:

list_of_resources (List[Resource]) –

Return type:

Indicator

add_indicator_resource_utilization(resource)

Compute the total utilization of a single resource.

The percentage is rounded to an int value.

Parameters:

resource (Resource) –

Return type:

Indicator

add_objective_flowtime(weight=1)

the flowtime is the sum of all ends, minimize. Be carful that it is contradictory with makespan

Return type:

Union[ArithRef, Indicator]

add_objective_flowtime_single_resource(resource, time_interval=None, weight=1)

Optimize flowtime for a single resource, for all the tasks scheduled in the time interval provided. Is ever no time interval is passed to the function, the flowtime is minimized for all the tasks scheduled in the workplan.

Return type:

Union[ArithRef, Indicator]

add_objective_makespan(weight=1)

makespan objective

Return type:

Union[ArithRef, Indicator]

add_objective_priorities(weight=1)

optimize the solution such that all task with a higher priority value are scheduled before other tasks

Return type:

Union[ArithRef, Indicator]

add_objective_resource_cost(list_of_resources, weight=1)

minimise the cost of selected resources

Parameters:

list_of_resources (List[Resource]) –

Return type:

Union[ArithRef, Indicator]

add_objective_resource_utilization(resource, weight=1)

Maximize resource occupation.

Parameters:

resource (Resource) –

Return type:

Union[ArithRef, Indicator]

add_objective_start_earliest(weight=1)

minimize the greatest start time, i.e. tasks are schedules as early as possible

Return type:

Union[ArithRef, Indicator]

add_objective_start_latest(weight=1)

maximize the minimum start time, i.e. all the tasks are scheduled as late as possible

Return type:

Union[ArithRef, Indicator]

append_z3_assertion(z3_assertion)

Add a z3 assertion to the list of assertions to be satisfied.

Parameters:

z3_assertion (BoolRef) – the z3 assertion

Return type:

bool

get_z3_assertions()

Return the assertions list

Return type:

List[BoolRef]

maximize_indicator(indicator)

Maximize indicator

Parameters:

indicator (Indicator) –

Return type:

MaximizeObjective

minimize_indicator(indicator)

Minimize indicator

Parameters:

indicator (Indicator) –

Return type:

MinimizeObjective

Tasks

digraph inheritance4f9972f595 { bgcolor=transparent; rankdir=LR; size="8.0, 12.0"; "task.FixedDurationTask" [URL="#processscheduler.task.FixedDurationTask",fillcolor=white,fontname="Vera Sans, DejaVu Sans, Liberation Sans, Arial, Helvetica, sans",fontsize=10,height=0.25,shape=box,style="setlinewidth(0.5),filled",target="_top",tooltip="Task with constant duration."]; "task.Task" -> "task.FixedDurationTask" [arrowsize=0.5,style="setlinewidth(0.5)"]; "task.Task" [fillcolor=white,fontname="Vera Sans, DejaVu Sans, Liberation Sans, Arial, Helvetica, sans",fontsize=10,height=0.25,shape=box,style="setlinewidth(0.5),filled",tooltip="a Task object"]; "task.UnavailabilityTask" [fillcolor=white,fontname="Vera Sans, DejaVu Sans, Liberation Sans, Arial, Helvetica, sans",fontsize=10,height=0.25,shape=box,style="setlinewidth(0.5),filled",tooltip="A task that tells that a resource is unavailable during this period. This"]; "task.FixedDurationTask" -> "task.UnavailabilityTask" [arrowsize=0.5,style="setlinewidth(0.5)"]; "task.VariableDurationTask" [URL="#processscheduler.task.VariableDurationTask",fillcolor=white,fontname="Vera Sans, DejaVu Sans, Liberation Sans, Arial, Helvetica, sans",fontsize=10,height=0.25,shape=box,style="setlinewidth(0.5),filled",target="_top",tooltip="The duration can take any value, computed by the solver."]; "task.Task" -> "task.VariableDurationTask" [arrowsize=0.5,style="setlinewidth(0.5)"]; "task.ZeroDurationTask" [URL="#processscheduler.task.ZeroDurationTask",fillcolor=white,fontname="Vera Sans, DejaVu Sans, Liberation Sans, Arial, Helvetica, sans",fontsize=10,height=0.25,shape=box,style="setlinewidth(0.5),filled",target="_top",tooltip="Task with zero duration, an instant in the schedule."]; "task.Task" -> "task.ZeroDurationTask" [arrowsize=0.5,style="setlinewidth(0.5)"]; }

The module that defines the task related classes.

class processscheduler.task.FixedDurationTask(name, duration, work_amount=0, priority=1, optional=False)

Bases: Task

Task with constant duration.

Parameters:

• name (str) – the task name. It must be unique

• duration (int) – the task duration as a number of periods

• work_amount (Optional[int]) – represent the quantity of work this task must produce

• priority (Optional[int]) – the task priority. The greater the priority, the sooner it will be scheduled

• optional (Optional[bool]) – True if task schedule is optional, False otherwise (default)

add_required_resource(resource, dynamic=False)

Add a required resource to the current task.

Parameters:

resource (Resource) – any of one of the Resource derivatives class (Worker, SelectWorkers etc.)

Return type:

None

If dynamic flag (False by default) is set to True, then the resource is dynamic and can join the task any time between its start and end times.

add_required_resources(list_of_resources, dynamic=False)

Add a set of required resources to the current task.

This method calls the add_required_resource method for each resource of the list. As a consequence, be aware this is not an atomic transaction.

Parameters:

list_of_resources (List[Resource]) – the list of resources to add.

Return type:

None

append_z3_assertion(z3_assertion)

Add a z3 assertion to the list of assertions to be satisfied.

Parameters:

z3_assertion (BoolRef) – the z3 assertion

Return type:

bool

get_z3_assertions()

Return the assertions list

Return type:

List[BoolRef]

set_assertions(list_of_z3_assertions)

Take a list of constraint to satisfy. Create two cases: if the task is scheduled, nothing is done; if the task is optional, move task to the past

Parameters:

list_of_z3_assertions (List[BoolRef]) –

Return type:

None

class processscheduler.task.VariableDurationTask(name, min_duration=0, max_duration=None, allowed_durations=None, work_amount=0, priority=1, optional=False)

Bases: Task

The duration can take any value, computed by the solver.

Parameters:

• name (str) –

• min_duration (Optional[int]) –

• max_duration (Optional[int]) –

• allowed_durations (Optional[List[int]]) –

• work_amount (Optional[int]) –

• priority (Optional[int]) –

• optional (Optional[bool]) –

add_required_resource(resource, dynamic=False)

Add a required resource to the current task.

Parameters:

resource (Resource) – any of one of the Resource derivatives class (Worker, SelectWorkers etc.)

Return type:

None

If dynamic flag (False by default) is set to True, then the resource is dynamic and can join the task any time between its start and end times.

add_required_resources(list_of_resources, dynamic=False)

Add a set of required resources to the current task.

This method calls the add_required_resource method for each resource of the list. As a consequence, be aware this is not an atomic transaction.

Parameters:

list_of_resources (List[Resource]) – the list of resources to add.

Return type:

None

append_z3_assertion(z3_assertion)

Add a z3 assertion to the list of assertions to be satisfied.

Parameters:

z3_assertion (BoolRef) – the z3 assertion

Return type:

bool

get_z3_assertions()

Return the assertions list

Return type:

List[BoolRef]

set_assertions(list_of_z3_assertions)

Take a list of constraint to satisfy. Create two cases: if the task is scheduled, nothing is done; if the task is optional, move task to the past

Parameters:

list_of_z3_assertions (List[BoolRef]) –

Return type:

None

class processscheduler.task.ZeroDurationTask(name, optional=False)

Bases: Task

Task with zero duration, an instant in the schedule.

The task end and start are constrained to be equal.

Parameters:

• name (str) – the task name. It must be unique

• optional (Optional[bool]) –

add_required_resource(resource, dynamic=False)

Add a required resource to the current task.

Parameters:

resource (Resource) – any of one of the Resource derivatives class (Worker, SelectWorkers etc.)

Return type:

None

If dynamic flag (False by default) is set to True, then the resource is dynamic and can join the task any time between its start and end times.

add_required_resources(list_of_resources, dynamic=False)

Add a set of required resources to the current task.

This method calls the add_required_resource method for each resource of the list. As a consequence, be aware this is not an atomic transaction.

Parameters:

list_of_resources (List[Resource]) – the list of resources to add.

Return type:

None

append_z3_assertion(z3_assertion)

Add a z3 assertion to the list of assertions to be satisfied.

Parameters:

z3_assertion (BoolRef) – the z3 assertion

Return type:

bool

get_z3_assertions()

Return the assertions list

Return type:

List[BoolRef]

set_assertions(list_of_z3_assertions)

Take a list of constraint to satisfy. Create two cases: if the task is scheduled, nothing is done; if the task is optional, move task to the past

Parameters:

list_of_z3_assertions (List[BoolRef]) –

Return type:

None

Resources

digraph inheritance184ad63a39 { bgcolor=transparent; rankdir=LR; size="8.0, 12.0"; "resource.CumulativeWorker" [URL="#processscheduler.resource.CumulativeWorker",fillcolor=white,fontname="Vera Sans, DejaVu Sans, Liberation Sans, Arial, Helvetica, sans",fontsize=10,height=0.25,shape=box,style="setlinewidth(0.5),filled",target="_top",tooltip="A cumulative worker can process multiple tasks in parallel."]; "resource.Resource" -> "resource.CumulativeWorker" [arrowsize=0.5,style="setlinewidth(0.5)"]; "resource.Resource" [URL="#processscheduler.resource.Resource",fillcolor=white,fontname="Vera Sans, DejaVu Sans, Liberation Sans, Arial, Helvetica, sans",fontsize=10,height=0.25,shape=box,style="setlinewidth(0.5),filled",target="_top",tooltip="base class for the representation of a resource"]; "resource.SelectWorkers" [URL="#processscheduler.resource.SelectWorkers",fillcolor=white,fontname="Vera Sans, DejaVu Sans, Liberation Sans, Arial, Helvetica, sans",fontsize=10,height=0.25,shape=box,style="setlinewidth(0.5),filled",target="_top",tooltip="Class representing the selection of n workers chosen among a list"]; "resource.Resource" -> "resource.SelectWorkers" [arrowsize=0.5,style="setlinewidth(0.5)"]; "resource.Worker" [URL="#processscheduler.resource.Worker",fillcolor=white,fontname="Vera Sans, DejaVu Sans, Liberation Sans, Arial, Helvetica, sans",fontsize=10,height=0.25,shape=box,style="setlinewidth(0.5),filled",target="_top",tooltip="A worker is an atomic resource that cannot be split into smaller parts."]; "resource.Resource" -> "resource.Worker" [arrowsize=0.5,style="setlinewidth(0.5)"]; }

class processscheduler.resource.CumulativeWorker(name, size, productivity=1, cost=None)

Bases: Resource

A cumulative worker can process multiple tasks in parallel.

Parameters:

• name (str) –

• size (int) –

• productivity (Optional[int]) –

• cost (Optional[int]) –

add_busy_interval(task, interval)

add an interval in which the resource is busy

Parameters:

interval (Tuple[ArithRef, ArithRef]) –

Return type:

None

append_z3_assertion(z3_assertion)

Add a z3 assertion to the list of assertions to be satisfied.

Parameters:

z3_assertion (BoolRef) – the z3 assertion

Return type:

bool

get_busy_intervals()

returns the list of all busy intervals

Return type:

List[Tuple[ArithRef, ArithRef]]

get_select_workers()

Each time the cumulative resource is assigned to a task, a SelectWorker instance is assigned to the task.

get_z3_assertions()

Return the assertions list

Return type:

List[BoolRef]

class processscheduler.resource.Resource(name)

Bases: _NamedUIDObject

base class for the representation of a resource

Parameters:

name (str) –

add_busy_interval(task, interval)

add an interval in which the resource is busy

Parameters:

interval (Tuple[ArithRef, ArithRef]) –

Return type:

None

append_z3_assertion(z3_assertion)

Add a z3 assertion to the list of assertions to be satisfied.

Parameters:

z3_assertion (BoolRef) – the z3 assertion

Return type:

bool

get_busy_intervals()

returns the list of all busy intervals

Return type:

List[Tuple[ArithRef, ArithRef]]

get_z3_assertions()

Return the assertions list

Return type:

List[BoolRef]

class processscheduler.resource.SelectWorkers(list_of_workers, nb_workers_to_select=1, kind='exact')

Bases: Resource

Class representing the selection of n workers chosen among a list of possible workers

Parameters:

• list_of_workers (List[Resource]) –

• nb_workers_to_select (Optional[int]) –

• kind (Optional[str]) –

add_busy_interval(task, interval)

add an interval in which the resource is busy

Parameters:

interval (Tuple[ArithRef, ArithRef]) –

Return type:

None

append_z3_assertion(z3_assertion)

Add a z3 assertion to the list of assertions to be satisfied.

Parameters:

z3_assertion (BoolRef) – the z3 assertion

Return type:

bool

get_busy_intervals()

returns the list of all busy intervals

Return type:

List[Tuple[ArithRef, ArithRef]]

get_z3_assertions()

Return the assertions list

Return type:

List[BoolRef]

class processscheduler.resource.Worker(name, productivity=1, cost=None)

Bases: Resource

A worker is an atomic resource that cannot be split into smaller parts. Typical workers are human beings, machines etc.

Parameters:

• name (str) –

• productivity (Optional[int]) –

• cost (Optional[int]) –

add_busy_interval(task, interval)

add an interval in which the resource is busy

Parameters:

interval (Tuple[ArithRef, ArithRef]) –

Return type:

None

append_z3_assertion(z3_assertion)

Add a z3 assertion to the list of assertions to be satisfied.

Parameters:

z3_assertion (BoolRef) – the z3 assertion

Return type:

bool

get_busy_intervals()

returns the list of all busy intervals

Return type:

List[Tuple[ArithRef, ArithRef]]

get_z3_assertions()

Return the assertions list

Return type:

List[BoolRef]

Constraints

digraph inheritance3c3c1452f4 { bgcolor=transparent; rankdir=LR; size="8.0, 12.0"; "constraint.Constraint" [fillcolor=white,fontname="Vera Sans, DejaVu Sans, Liberation Sans, Arial, Helvetica, sans",fontsize=10,height=0.25,shape=box,style="setlinewidth(0.5),filled",tooltip="The base class for all constraints, including Task and Resource constraints."]; "constraint.ResourceConstraint" [fillcolor=white,fontname="Vera Sans, DejaVu Sans, Liberation Sans, Arial, Helvetica, sans",fontsize=10,height=0.25,shape=box,style="setlinewidth(0.5),filled",tooltip="Constraint that applies on a Resource (typically a Worker)"]; "constraint.Constraint" -> "constraint.ResourceConstraint" [arrowsize=0.5,style="setlinewidth(0.5)"]; "constraint.TaskConstraint" [fillcolor=white,fontname="Vera Sans, DejaVu Sans, Liberation Sans, Arial, Helvetica, sans",fontsize=10,height=0.25,shape=box,style="setlinewidth(0.5),filled",tooltip="Constraint that applies on a Task"]; "constraint.Constraint" -> "constraint.TaskConstraint" [arrowsize=0.5,style="setlinewidth(0.5)"]; "resource_constraint.DistinctWorkers" [URL="#processscheduler.resource_constraint.DistinctWorkers",fillcolor=white,fontname="Vera Sans, DejaVu Sans, Liberation Sans, Arial, Helvetica, sans",fontsize=10,height=0.25,shape=box,style="setlinewidth(0.5),filled",target="_top",tooltip="This constraint ensures that workers selected by two SelectWorker instances are distinct."]; "constraint.ResourceConstraint" -> "resource_constraint.DistinctWorkers" [arrowsize=0.5,style="setlinewidth(0.5)"]; "resource_constraint.ResourceTasksDistance" [URL="#processscheduler.resource_constraint.ResourceTasksDistance",fillcolor=white,fontname="Vera Sans, DejaVu Sans, Liberation Sans, Arial, Helvetica, sans",fontsize=10,height=0.25,shape=box,style="setlinewidth(0.5),filled",target="_top",tooltip="This constraint enforces a specific number of time unitary periods between tasks for a single resource."]; "constraint.ResourceConstraint" -> "resource_constraint.ResourceTasksDistance" [arrowsize=0.5,style="setlinewidth(0.5)"]; "resource_constraint.ResourceUnavailable" [URL="#processscheduler.resource_constraint.ResourceUnavailable",fillcolor=white,fontname="Vera Sans, DejaVu Sans, Liberation Sans, Arial, Helvetica, sans",fontsize=10,height=0.25,shape=box,style="setlinewidth(0.5),filled",target="_top",tooltip="This constraint sets the unavailability of a resource in terms of time intervals."]; "constraint.ResourceConstraint" -> "resource_constraint.ResourceUnavailable" [arrowsize=0.5,style="setlinewidth(0.5)"]; "resource_constraint.SameWorkers" [URL="#processscheduler.resource_constraint.SameWorkers",fillcolor=white,fontname="Vera Sans, DejaVu Sans, Liberation Sans, Arial, Helvetica, sans",fontsize=10,height=0.25,shape=box,style="setlinewidth(0.5),filled",target="_top",tooltip="This constraint ensures that workers selected by two SelectWorker instances are the same."]; "constraint.ResourceConstraint" -> "resource_constraint.SameWorkers" [arrowsize=0.5,style="setlinewidth(0.5)"]; "resource_constraint.WorkLoad" [URL="#processscheduler.resource_constraint.WorkLoad",fillcolor=white,fontname="Vera Sans, DejaVu Sans, Liberation Sans, Arial, Helvetica, sans",fontsize=10,height=0.25,shape=box,style="setlinewidth(0.5),filled",target="_top",tooltip="A WorkLoad constraint restricts the number of tasks that can be executed on a resource"]; "constraint.ResourceConstraint" -> "resource_constraint.WorkLoad" [arrowsize=0.5,style="setlinewidth(0.5)"]; "task_constraint.ForceScheduleNOptionalTasks" [URL="#processscheduler.task_constraint.ForceScheduleNOptionalTasks",fillcolor=white,fontname="Vera Sans, DejaVu Sans, Liberation Sans, Arial, Helvetica, sans",fontsize=10,height=0.25,shape=box,style="setlinewidth(0.5),filled",target="_top",tooltip="Given a set of m different optional tasks, force the solver to schedule"]; "constraint.TaskConstraint" -> "task_constraint.ForceScheduleNOptionalTasks" [arrowsize=0.5,style="setlinewidth(0.5)"]; "task_constraint.OptionalTaskConditionSchedule" [URL="#processscheduler.task_constraint.OptionalTaskConditionSchedule",fillcolor=white,fontname="Vera Sans, DejaVu Sans, Liberation Sans, Arial, Helvetica, sans",fontsize=10,height=0.25,shape=box,style="setlinewidth(0.5),filled",target="_top",tooltip="An optional task that is scheduled only if a condition is fulfilled."]; "constraint.TaskConstraint" -> "task_constraint.OptionalTaskConditionSchedule" [arrowsize=0.5,style="setlinewidth(0.5)"]; "task_constraint.OptionalTasksDependency" [URL="#processscheduler.task_constraint.OptionalTasksDependency",fillcolor=white,fontname="Vera Sans, DejaVu Sans, Liberation Sans, Arial, Helvetica, sans",fontsize=10,height=0.25,shape=box,style="setlinewidth(0.5),filled",target="_top",tooltip="task_2 is scheduled if and only if task_1 is scheduled"]; "constraint.TaskConstraint" -> "task_constraint.OptionalTasksDependency" [arrowsize=0.5,style="setlinewidth(0.5)"]; "task_constraint.OrderedTaskGroup" [URL="#processscheduler.task_constraint.OrderedTaskGroup",fillcolor=white,fontname="Vera Sans, DejaVu Sans, Liberation Sans, Arial, Helvetica, sans",fontsize=10,height=0.25,shape=box,style="setlinewidth(0.5),filled",target="_top",tooltip="A set of tasks that can be scheduled in a specified order, with time bounds."]; "constraint.TaskConstraint" -> "task_constraint.OrderedTaskGroup" [arrowsize=0.5,style="setlinewidth(0.5)"]; "task_constraint.ScheduleNTasksInTimeIntervals" [URL="#processscheduler.task_constraint.ScheduleNTasksInTimeIntervals",fillcolor=white,fontname="Vera Sans, DejaVu Sans, Liberation Sans, Arial, Helvetica, sans",fontsize=10,height=0.25,shape=box,style="setlinewidth(0.5),filled",target="_top",tooltip="Given a set of m different tasks, and a list of time intervals, schedule N tasks among m"]; "constraint.TaskConstraint" -> "task_constraint.ScheduleNTasksInTimeIntervals" [arrowsize=0.5,style="setlinewidth(0.5)"]; "task_constraint.TaskEndAt" [URL="#processscheduler.task_constraint.TaskEndAt",fillcolor=white,fontname="Vera Sans, DejaVu Sans, Liberation Sans, Arial, Helvetica, sans",fontsize=10,height=0.25,shape=box,style="setlinewidth(0.5),filled",target="_top",tooltip="On task must complete at the desired time"]; "constraint.TaskConstraint" -> "task_constraint.TaskEndAt" [arrowsize=0.5,style="setlinewidth(0.5)"]; "task_constraint.TaskEndBefore" [URL="#processscheduler.task_constraint.TaskEndBefore",fillcolor=white,fontname="Vera Sans, DejaVu Sans, Liberation Sans, Arial, Helvetica, sans",fontsize=10,height=0.25,shape=box,style="setlinewidth(0.5),filled",target="_top",tooltip="task.end < value"]; "constraint.TaskConstraint" -> "task_constraint.TaskEndBefore" [arrowsize=0.5,style="setlinewidth(0.5)"]; "task_constraint.TaskLoadBuffer" [URL="#processscheduler.task_constraint.TaskLoadBuffer",fillcolor=white,fontname="Vera Sans, DejaVu Sans, Liberation Sans, Arial, Helvetica, sans",fontsize=10,height=0.25,shape=box,style="setlinewidth(0.5),filled",target="_top",tooltip="A task that loads a buffer"]; "constraint.TaskConstraint" -> "task_constraint.TaskLoadBuffer" [arrowsize=0.5,style="setlinewidth(0.5)"]; "task_constraint.TaskPrecedence" [URL="#processscheduler.task_constraint.TaskPrecedence",fillcolor=white,fontname="Vera Sans, DejaVu Sans, Liberation Sans, Arial, Helvetica, sans",fontsize=10,height=0.25,shape=box,style="setlinewidth(0.5),filled",target="_top",tooltip="Task precedence relation"]; "constraint.TaskConstraint" -> "task_constraint.TaskPrecedence" [arrowsize=0.5,style="setlinewidth(0.5)"]; "task_constraint.TaskStartAfter" [URL="#processscheduler.task_constraint.TaskStartAfter",fillcolor=white,fontname="Vera Sans, DejaVu Sans, Liberation Sans, Arial, Helvetica, sans",fontsize=10,height=0.25,shape=box,style="setlinewidth(0.5),filled",target="_top"]; "constraint.TaskConstraint" -> "task_constraint.TaskStartAfter" [arrowsize=0.5,style="setlinewidth(0.5)"]; "task_constraint.TaskStartAt" [URL="#processscheduler.task_constraint.TaskStartAt",fillcolor=white,fontname="Vera Sans, DejaVu Sans, Liberation Sans, Arial, Helvetica, sans",fontsize=10,height=0.25,shape=box,style="setlinewidth(0.5),filled",target="_top",tooltip="One task must start at the desired time"]; "constraint.TaskConstraint" -> "task_constraint.TaskStartAt" [arrowsize=0.5,style="setlinewidth(0.5)"]; "task_constraint.TaskUnloadBuffer" [URL="#processscheduler.task_constraint.TaskUnloadBuffer",fillcolor=white,fontname="Vera Sans, DejaVu Sans, Liberation Sans, Arial, Helvetica, sans",fontsize=10,height=0.25,shape=box,style="setlinewidth(0.5),filled",target="_top",tooltip="A tasks that unloads a buffer"]; "constraint.TaskConstraint" -> "task_constraint.TaskUnloadBuffer" [arrowsize=0.5,style="setlinewidth(0.5)"]; "task_constraint.TasksContiguous" [URL="#processscheduler.task_constraint.TasksContiguous",fillcolor=white,fontname="Vera Sans, DejaVu Sans, Liberation Sans, Arial, Helvetica, sans",fontsize=10,height=0.25,shape=box,style="setlinewidth(0.5),filled",target="_top",tooltip="A list of tasks are scheduled contiguously."]; "constraint.TaskConstraint" -> "task_constraint.TasksContiguous" [arrowsize=0.5,style="setlinewidth(0.5)"]; "task_constraint.TasksDontOverlap" [URL="#processscheduler.task_constraint.TasksDontOverlap",fillcolor=white,fontname="Vera Sans, DejaVu Sans, Liberation Sans, Arial, Helvetica, sans",fontsize=10,height=0.25,shape=box,style="setlinewidth(0.5),filled",target="_top",tooltip="Two tasks must not overlap, i.e. one needs to be completed before"]; "constraint.TaskConstraint" -> "task_constraint.TasksDontOverlap" [arrowsize=0.5,style="setlinewidth(0.5)"]; "task_constraint.TasksEndSynced" [URL="#processscheduler.task_constraint.TasksEndSynced",fillcolor=white,fontname="Vera Sans, DejaVu Sans, Liberation Sans, Arial, Helvetica, sans",fontsize=10,height=0.25,shape=box,style="setlinewidth(0.5),filled",target="_top",tooltip="Two tasks that must complete at the same time"]; "constraint.TaskConstraint" -> "task_constraint.TasksEndSynced" [arrowsize=0.5,style="setlinewidth(0.5)"]; "task_constraint.TasksStartSynced" [URL="#processscheduler.task_constraint.TasksStartSynced",fillcolor=white,fontname="Vera Sans, DejaVu Sans, Liberation Sans, Arial, Helvetica, sans",fontsize=10,height=0.25,shape=box,style="setlinewidth(0.5),filled",target="_top",tooltip="Two tasks that must start at the same time"]; "constraint.TaskConstraint" -> "task_constraint.TasksStartSynced" [arrowsize=0.5,style="setlinewidth(0.5)"]; "task_constraint.UnorderedTaskGroup" [URL="#processscheduler.task_constraint.UnorderedTaskGroup",fillcolor=white,fontname="Vera Sans, DejaVu Sans, Liberation Sans, Arial, Helvetica, sans",fontsize=10,height=0.25,shape=box,style="setlinewidth(0.5),filled",target="_top",tooltip="A set of tasks that can be scheduled in any order, with time bounds."]; "constraint.TaskConstraint" -> "task_constraint.UnorderedTaskGroup" [arrowsize=0.5,style="setlinewidth(0.5)"]; }

Task constraints and related classes.

class processscheduler.task_constraint.ForceScheduleNOptionalTasks(list_of_optional_tasks, nb_tasks_to_schedule=1, kind='exact', optional=False)

Bases: TaskConstraint

Given a set of m different optional tasks, force the solver to schedule at at least/at most/exactly n tasks, with 0 < n <= m.

Parameters:

• nb_tasks_to_schedule (Optional[int]) –

• kind (Optional[str]) –

• optional (Optional[bool]) –

append_z3_assertion(z3_assertion)

Add a z3 assertion to the list of assertions to be satisfied.

Parameters:

z3_assertion (BoolRef) – the z3 assertion

Return type:

bool

get_z3_assertions()

Return the assertions list

Return type:

List[BoolRef]

set_created_from_assertion()

Set the flag created_from_assertion True. This flag must be set to True if, for example, a constraint is defined from the expression: ps.not_(ps.TaskStartAt(task_1, 0)) thus, the Task task_1 assertions must not be add to the z3 solver.

Return type:

None

set_z3_assertions(list_of_z3_assertions)

Each constraint comes with a set of z3 assertions to satisfy.

Parameters:

list_of_z3_assertions (List[BoolRef]) –

Return type:

None

class processscheduler.task_constraint.OptionalTaskConditionSchedule(task, condition, optional=False)

Bases: TaskConstraint

An optional task that is scheduled only if a condition is fulfilled.

Parameters:

• condition (BoolRef) –

• optional (Optional[bool]) –

append_z3_assertion(z3_assertion)

Add a z3 assertion to the list of assertions to be satisfied.

Parameters:

z3_assertion (BoolRef) – the z3 assertion

Return type:

bool

get_z3_assertions()

Return the assertions list

Return type:

List[BoolRef]

set_created_from_assertion()

Set the flag created_from_assertion True. This flag must be set to True if, for example, a constraint is defined from the expression: ps.not_(ps.TaskStartAt(task_1, 0)) thus, the Task task_1 assertions must not be add to the z3 solver.

Return type:

None

set_z3_assertions(list_of_z3_assertions)

Each constraint comes with a set of z3 assertions to satisfy.

Parameters:

list_of_z3_assertions (List[BoolRef]) –

Return type:

None

class processscheduler.task_constraint.OptionalTasksDependency(task_1, task_2, optional=False)

Bases: TaskConstraint

task_2 is scheduled if and only if task_1 is scheduled

Parameters:

optional (Optional[bool]) –

append_z3_assertion(z3_assertion)

Add a z3 assertion to the list of assertions to be satisfied.

Parameters:

z3_assertion (BoolRef) – the z3 assertion

Return type:

bool

get_z3_assertions()

Return the assertions list

Return type:

List[BoolRef]

set_created_from_assertion()

Set the flag created_from_assertion True. This flag must be set to True if, for example, a constraint is defined from the expression: ps.not_(ps.TaskStartAt(task_1, 0)) thus, the Task task_1 assertions must not be add to the z3 solver.

Return type:

None

set_z3_assertions(list_of_z3_assertions)

Each constraint comes with a set of z3 assertions to satisfy.

Parameters:

list_of_z3_assertions (List[BoolRef]) –

Return type:

None

class processscheduler.task_constraint.OrderedTaskGroup(list_of_tasks, time_interval=None, kind='lax', time_interval_length=0, optional=False)

Bases: TaskConstraint

A set of tasks that can be scheduled in a specified order, with time bounds.

Parameters:

• kind (Optional[str]) –

• time_interval_length (Optional[int]) –

• optional (Optional[bool]) –

append_z3_assertion(z3_assertion)

Add a z3 assertion to the list of assertions to be satisfied.

Parameters:

z3_assertion (BoolRef) – the z3 assertion

Return type:

bool

get_z3_assertions()

Return the assertions list

Return type:

List[BoolRef]

set_created_from_assertion()

Set the flag created_from_assertion True. This flag must be set to True if, for example, a constraint is defined from the expression: ps.not_(ps.TaskStartAt(task_1, 0)) thus, the Task task_1 assertions must not be add to the z3 solver.

Return type:

None

set_z3_assertions(list_of_z3_assertions)

Each constraint comes with a set of z3 assertions to satisfy.

Parameters:

list_of_z3_assertions (List[BoolRef]) –

Return type:

None

class processscheduler.task_constraint.ScheduleNTasksInTimeIntervals(list_of_tasks, nb_tasks_to_schedule, list_of_time_intervals, kind='exact', optional=False)

Bases: TaskConstraint

Given a set of m different tasks, and a list of time intervals, schedule N tasks among m in this time interval

Parameters:

• kind (Optional[str]) –

• optional (Optional[bool]) –

append_z3_assertion(z3_assertion)

Add a z3 assertion to the list of assertions to be satisfied.

Parameters:

z3_assertion (BoolRef) – the z3 assertion

Return type:

bool

get_z3_assertions()

Return the assertions list

Return type:

List[BoolRef]

set_created_from_assertion()

Set the flag created_from_assertion True. This flag must be set to True if, for example, a constraint is defined from the expression: ps.not_(ps.TaskStartAt(task_1, 0)) thus, the Task task_1 assertions must not be add to the z3 solver.

Return type:

None

set_z3_assertions(list_of_z3_assertions)

Each constraint comes with a set of z3 assertions to satisfy.

Parameters:

list_of_z3_assertions (List[BoolRef]) –

Return type:

None

class processscheduler.task_constraint.TaskEndAt(task, value, optional=False)

Bases: TaskConstraint

On task must complete at the desired time

Parameters:

• value (int) –

• optional (Optional[bool]) –

append_z3_assertion(z3_assertion)

Add a z3 assertion to the list of assertions to be satisfied.

Parameters:

z3_assertion (BoolRef) – the z3 assertion

Return type:

bool

get_z3_assertions()

Return the assertions list

Return type:

List[BoolRef]

set_created_from_assertion()

Set the flag created_from_assertion True. This flag must be set to True if, for example, a constraint is defined from the expression: ps.not_(ps.TaskStartAt(task_1, 0)) thus, the Task task_1 assertions must not be add to the z3 solver.

Return type:

None

set_z3_assertions(list_of_z3_assertions)

Each constraint comes with a set of z3 assertions to satisfy.

Parameters:

list_of_z3_assertions (List[BoolRef]) –

Return type:

None

class processscheduler.task_constraint.TaskEndBefore(task, value, kind='lax', optional=False)

Bases: TaskConstraint

task.end < value

Parameters:

• value (int) –

• kind (Optional[str]) –

• optional (Optional[bool]) –

append_z3_assertion(z3_assertion)

Add a z3 assertion to the list of assertions to be satisfied.

Parameters:

z3_assertion (BoolRef) – the z3 assertion

Return type:

bool

get_z3_assertions()

Return the assertions list

Return type:

List[BoolRef]

set_created_from_assertion()

Set the flag created_from_assertion True. This flag must be set to True if, for example, a constraint is defined from the expression: ps.not_(ps.TaskStartAt(task_1, 0)) thus, the Task task_1 assertions must not be add to the z3 solver.

Return type:

None

set_z3_assertions(list_of_z3_assertions)

Each constraint comes with a set of z3 assertions to satisfy.

Parameters:

list_of_z3_assertions (List[BoolRef]) –

Return type:

None

class processscheduler.task_constraint.TaskLoadBuffer(task, buffer, quantity, optional=False)

Bases: TaskConstraint

A task that loads a buffer

Parameters:

optional (Optional[bool]) –

append_z3_assertion(z3_assertion)

Add a z3 assertion to the list of assertions to be satisfied.

Parameters:

z3_assertion (BoolRef) – the z3 assertion

Return type:

bool

get_z3_assertions()

Return the assertions list

Return type:

List[BoolRef]

set_created_from_assertion()

Set the flag created_from_assertion True. This flag must be set to True if, for example, a constraint is defined from the expression: ps.not_(ps.TaskStartAt(task_1, 0)) thus, the Task task_1 assertions must not be add to the z3 solver.

Return type:

None

set_z3_assertions(list_of_z3_assertions)

Each constraint comes with a set of z3 assertions to satisfy.

Parameters:

list_of_z3_assertions (List[BoolRef]) –

Return type:

None

class processscheduler.task_constraint.TaskPrecedence(task_before, task_after, offset=0, kind='lax', optional=False)

Bases: TaskConstraint

Task precedence relation

Parameters:

• kind (Optional[str]) –

• optional (Optional[bool]) –

append_z3_assertion(z3_assertion)

Add a z3 assertion to the list of assertions to be satisfied.

Parameters:

z3_assertion (BoolRef) – the z3 assertion

Return type:

bool

get_z3_assertions()

Return the assertions list

Return type:

List[BoolRef]

set_created_from_assertion()

Set the flag created_from_assertion True. This flag must be set to True if, for example, a constraint is defined from the expression: ps.not_(ps.TaskStartAt(task_1, 0)) thus, the Task task_1 assertions must not be add to the z3 solver.

Return type:

None

set_z3_assertions(list_of_z3_assertions)

Each constraint comes with a set of z3 assertions to satisfy.

Parameters:

list_of_z3_assertions (List[BoolRef]) –

Return type:

None

class processscheduler.task_constraint.TaskStartAfter(task, value, kind='lax', optional=False)

Bases: TaskConstraint

Parameters:

• value (int) –

• kind (Optional[str]) –

• optional (Optional[bool]) –

append_z3_assertion(z3_assertion)

Add a z3 assertion to the list of assertions to be satisfied.

Parameters:

z3_assertion (BoolRef) – the z3 assertion

Return type:

bool

get_z3_assertions()

Return the assertions list

Return type:

List[BoolRef]

set_created_from_assertion()

Set the flag created_from_assertion True. This flag must be set to True if, for example, a constraint is defined from the expression: ps.not_(ps.TaskStartAt(task_1, 0)) thus, the Task task_1 assertions must not be add to the z3 solver.

Return type:

None

set_z3_assertions(list_of_z3_assertions)

Each constraint comes with a set of z3 assertions to satisfy.

Parameters:

list_of_z3_assertions (List[BoolRef]) –

Return type:

None

class processscheduler.task_constraint.TaskStartAt(task, value, optional=False)

Bases: TaskConstraint

One task must start at the desired time

Parameters:

• value (int) –

• optional (Optional[bool]) –

append_z3_assertion(z3_assertion)

Add a z3 assertion to the list of assertions to be satisfied.

Parameters:

z3_assertion (BoolRef) – the z3 assertion

Return type:

bool

get_z3_assertions()

Return the assertions list

Return type:

List[BoolRef]

set_created_from_assertion()

Set the flag created_from_assertion True. This flag must be set to True if, for example, a constraint is defined from the expression: ps.not_(ps.TaskStartAt(task_1, 0)) thus, the Task task_1 assertions must not be add to the z3 solver.

Return type:

None

set_z3_assertions(list_of_z3_assertions)

Each constraint comes with a set of z3 assertions to satisfy.

Parameters:

list_of_z3_assertions (List[BoolRef]) –

Return type:

None

class processscheduler.task_constraint.TaskUnloadBuffer(task, buffer, quantity, optional=False)

Bases: TaskConstraint

A tasks that unloads a buffer

Parameters:

optional (Optional[bool]) –

append_z3_assertion(z3_assertion)

Add a z3 assertion to the list of assertions to be satisfied.

Parameters:

z3_assertion (BoolRef) – the z3 assertion

Return type:

bool

get_z3_assertions()

Return the assertions list

Return type:

List[BoolRef]

set_created_from_assertion()

Set the flag created_from_assertion True. This flag must be set to True if, for example, a constraint is defined from the expression: ps.not_(ps.TaskStartAt(task_1, 0)) thus, the Task task_1 assertions must not be add to the z3 solver.

Return type:

None

set_z3_assertions(list_of_z3_assertions)

Each constraint comes with a set of z3 assertions to satisfy.

Parameters:

list_of_z3_assertions (List[BoolRef]) –

Return type:

None

class processscheduler.task_constraint.TasksContiguous(list_of_tasks, optional=False)

Bases: TaskConstraint

A list of tasks are scheduled contiguously.

Parameters:

optional (Optional[bool]) –

append_z3_assertion(z3_assertion)

Add a z3 assertion to the list of assertions to be satisfied.

Parameters:

z3_assertion (BoolRef) – the z3 assertion

Return type:

bool

get_z3_assertions()

Return the assertions list

Return type:

List[BoolRef]

set_created_from_assertion()

Set the flag created_from_assertion True. This flag must be set to True if, for example, a constraint is defined from the expression: ps.not_(ps.TaskStartAt(task_1, 0)) thus, the Task task_1 assertions must not be add to the z3 solver.

Return type:

None

set_z3_assertions(list_of_z3_assertions)

Each constraint comes with a set of z3 assertions to satisfy.

Parameters:

list_of_z3_assertions (List[BoolRef]) –

Return type:

None

class processscheduler.task_constraint.TasksDontOverlap(task_1, task_2, optional=False)

Bases: TaskConstraint

Two tasks must not overlap, i.e. one needs to be completed before the other can be processed

Parameters:

optional (Optional[bool]) –

append_z3_assertion(z3_assertion)

Add a z3 assertion to the list of assertions to be satisfied.

Parameters:

z3_assertion (BoolRef) – the z3 assertion

Return type:

bool

get_z3_assertions()

Return the assertions list

Return type:

List[BoolRef]

set_created_from_assertion()

Set the flag created_from_assertion True. This flag must be set to True if, for example, a constraint is defined from the expression: ps.not_(ps.TaskStartAt(task_1, 0)) thus, the Task task_1 assertions must not be add to the z3 solver.

Return type:

None

set_z3_assertions(list_of_z3_assertions)

Each constraint comes with a set of z3 assertions to satisfy.

Parameters:

list_of_z3_assertions (List[BoolRef]) –

Return type:

None

class processscheduler.task_constraint.TasksEndSynced(task_1, task_2, optional=False)

Bases: TaskConstraint

Two tasks that must complete at the same time

Parameters:

optional (Optional[bool]) –

append_z3_assertion(z3_assertion)

Add a z3 assertion to the list of assertions to be satisfied.

Parameters:

z3_assertion (BoolRef) – the z3 assertion

Return type:

bool

get_z3_assertions()

Return the assertions list

Return type:

List[BoolRef]

set_created_from_assertion()

Set the flag created_from_assertion True. This flag must be set to True if, for example, a constraint is defined from the expression: ps.not_(ps.TaskStartAt(task_1, 0)) thus, the Task task_1 assertions must not be add to the z3 solver.

Return type:

None

set_z3_assertions(list_of_z3_assertions)

Each constraint comes with a set of z3 assertions to satisfy.

Parameters:

list_of_z3_assertions (List[BoolRef]) –

Return type:

None

class processscheduler.task_constraint.TasksStartSynced(task_1, task_2, optional=False)

Bases: TaskConstraint

Two tasks that must start at the same time

Parameters:

optional (Optional[bool]) –

append_z3_assertion(z3_assertion)

Add a z3 assertion to the list of assertions to be satisfied.

Parameters:

z3_assertion (BoolRef) – the z3 assertion

Return type:

bool

get_z3_assertions()

Return the assertions list

Return type:

List[BoolRef]

set_created_from_assertion()

Set the flag created_from_assertion True. This flag must be set to True if, for example, a constraint is defined from the expression: ps.not_(ps.TaskStartAt(task_1, 0)) thus, the Task task_1 assertions must not be add to the z3 solver.

Return type:

None

set_z3_assertions(list_of_z3_assertions)

Each constraint comes with a set of z3 assertions to satisfy.

Parameters:

list_of_z3_assertions (List[BoolRef]) –

Return type:

None

class processscheduler.task_constraint.UnorderedTaskGroup(list_of_tasks, time_interval=None, time_interval_length=0, optional=False)

Bases: TaskConstraint

A set of tasks that can be scheduled in any order, with time bounds.

Parameters:

• time_interval_length (Optional[int]) –

• optional (Optional[bool]) –

append_z3_assertion(z3_assertion)

Add a z3 assertion to the list of assertions to be satisfied.

Parameters:

z3_assertion (BoolRef) – the z3 assertion

Return type:

bool

get_z3_assertions()

Return the assertions list

Return type:

List[BoolRef]

set_created_from_assertion()

Set the flag created_from_assertion True. This flag must be set to True if, for example, a constraint is defined from the expression: ps.not_(ps.TaskStartAt(task_1, 0)) thus, the Task task_1 assertions must not be add to the z3 solver.

Return type:

None

set_z3_assertions(list_of_z3_assertions)

Each constraint comes with a set of z3 assertions to satisfy.

Parameters:

list_of_z3_assertions (List[BoolRef]) –

Return type:

None

Resource constraints and related classes.

class processscheduler.resource_constraint.DistinctWorkers(select_workers_1, select_workers_2, optional=False)

Bases: ResourceConstraint

This constraint ensures that workers selected by two SelectWorker instances are distinct.

Parameters: - select_workers_1: The first set of selected workers. - select_workers_2: The second set of selected workers. - optional (bool, optional): Whether the constraint is optional (default is False).

Parameters:

optional (Optional[bool]) –

append_z3_assertion(z3_assertion)

Add a z3 assertion to the list of assertions to be satisfied.

Parameters:

z3_assertion (BoolRef) – the z3 assertion

Return type:

bool

get_z3_assertions()

Return the assertions list

Return type:

List[BoolRef]

set_created_from_assertion()

Set the flag created_from_assertion True. This flag must be set to True if, for example, a constraint is defined from the expression: ps.not_(ps.TaskStartAt(task_1, 0)) thus, the Task task_1 assertions must not be add to the z3 solver.

Return type:

None

set_z3_assertions(list_of_z3_assertions)

Each constraint comes with a set of z3 assertions to satisfy.

Parameters:

list_of_z3_assertions (List[BoolRef]) –

Return type:

None

class processscheduler.resource_constraint.ResourceTasksDistance(resource, distance, list_of_time_intervals=None, optional=False, mode='exact')

Bases: ResourceConstraint

This constraint enforces a specific number of time unitary periods between tasks for a single resource. The constraint can be applied within specified time intervals.

Parameters: - resource: The resource to which the constraint applies. - distance (int): The desired number of time unitary periods between tasks. - list_of_time_intervals (list, optional): A list of time intervals within which the constraint is restricted. - optional (bool, optional): Whether the constraint is optional (default is False). - mode (str, optional): The mode for enforcing the constraint - “exact” (default), “min”, or “max”.

Parameters:

• distance (int) –

• list_of_time_intervals (Optional[list]) –

• optional (Optional[bool]) –

• mode (Optional[str]) –

append_z3_assertion(z3_assertion)

Add a z3 assertion to the list of assertions to be satisfied.

Parameters:

z3_assertion (BoolRef) – the z3 assertion

Return type:

bool

get_z3_assertions()

Return the assertions list

Return type:

List[BoolRef]

set_created_from_assertion()

Set the flag created_from_assertion True. This flag must be set to True if, for example, a constraint is defined from the expression: ps.not_(ps.TaskStartAt(task_1, 0)) thus, the Task task_1 assertions must not be add to the z3 solver.

Return type:

None

set_z3_assertions(list_of_z3_assertions)

Each constraint comes with a set of z3 assertions to satisfy.

Parameters:

list_of_z3_assertions (List[BoolRef]) –

Return type:

None

class processscheduler.resource_constraint.ResourceUnavailable(resource, list_of_time_intervals, optional=False)

Bases: ResourceConstraint

This constraint sets the unavailability of a resource in terms of time intervals.

Parameters: - resource: The resource for which unavailability is defined. - list_of_time_intervals: A list of time intervals during which the resource is unavailable for any task.

For example, [(0, 2), (5, 8)] represents time intervals from 0 to 2 and from 5 to 8.

• optional (bool, optional): Whether the constraint is optional (default is False).

Parameters:

optional (Optional[bool]) –

append_z3_assertion(z3_assertion)

Add a z3 assertion to the list of assertions to be satisfied.

Parameters:

z3_assertion (BoolRef) – the z3 assertion

Return type:

bool

get_z3_assertions()

Return the assertions list

Return type:

List[BoolRef]

set_created_from_assertion()

Set the flag created_from_assertion True. This flag must be set to True if, for example, a constraint is defined from the expression: ps.not_(ps.TaskStartAt(task_1, 0)) thus, the Task task_1 assertions must not be add to the z3 solver.

Return type:

None

set_z3_assertions(list_of_z3_assertions)

Each constraint comes with a set of z3 assertions to satisfy.

Parameters:

list_of_z3_assertions (List[BoolRef]) –

Return type:

None

class processscheduler.resource_constraint.SameWorkers(select_workers_1, select_workers_2, optional=False)

Bases: ResourceConstraint

This constraint ensures that workers selected by two SelectWorker instances are the same.

Parameters: - select_workers_1: The first set of selected workers. - select_workers_2: The second set of selected workers. - optional (bool, optional): Whether the constraint is optional (default is False).

Parameters:

optional (Optional[bool]) –

append_z3_assertion(z3_assertion)

Add a z3 assertion to the list of assertions to be satisfied.

Parameters:

z3_assertion (BoolRef) – the z3 assertion

Return type:

bool

get_z3_assertions()

Return the assertions list

Return type:

List[BoolRef]

set_created_from_assertion()

Set the flag created_from_assertion True. This flag must be set to True if, for example, a constraint is defined from the expression: ps.not_(ps.TaskStartAt(task_1, 0)) thus, the Task task_1 assertions must not be add to the z3 solver.

Return type:

None

set_z3_assertions(list_of_z3_assertions)

Each constraint comes with a set of z3 assertions to satisfy.

Parameters:

list_of_z3_assertions (List[BoolRef]) –

Return type:

None

class processscheduler.resource_constraint.WorkLoad(resource, dict_time_intervals_and_bound, kind='max', optional=False)

Bases: ResourceConstraint

A WorkLoad constraint restricts the number of tasks that can be executed on a resource during a certain time period. The resource can be a single Worker or a CumulativeWorker.

Parameters: - resource: The resource for which the workload constraint is defined. - dict_time_intervals_and_bound: A dictionary that maps interval tuples to integer bounds.

For example, {(1,20):6, (50,60):2} specifies that the resource can use no more than 6 slots in the interval (1,20), and no more than 2 slots in the interval (50,60).

• kind (str, optional): Specifies whether the constraint is exact, a minimum, or a maximum (default is “max”).

• optional (bool, optional): Whether the constraint is optional (default is False).

Parameters:

• kind (Optional[str]) –

• optional (Optional[bool]) –

append_z3_assertion(z3_assertion)

Add a z3 assertion to the list of assertions to be satisfied.

Parameters:

z3_assertion (BoolRef) – the z3 assertion

Return type:

bool

get_z3_assertions()

Return the assertions list

Return type:

List[BoolRef]

set_created_from_assertion()

Set the flag created_from_assertion True. This flag must be set to True if, for example, a constraint is defined from the expression: ps.not_(ps.TaskStartAt(task_1, 0)) thus, the Task task_1 assertions must not be add to the z3 solver.

Return type:

None

set_z3_assertions(list_of_z3_assertions)

Each constraint comes with a set of z3 assertions to satisfy.

Parameters:

list_of_z3_assertions (List[BoolRef]) –

Return type:

None

Solver

Solver definition and related classes/functions.

class processscheduler.solver.SchedulingSolver(problem, debug=False, max_time=10, parallel=False, random_values=False, logics=None, verbosity=0, optimizer='incremental', optimize_priority='pareto')

Bases: object

A solver class

Parameters:

• debug (Optional[bool]) –

• max_time (Optional[int]) –

• parallel (Optional[bool]) –

• random_values (Optional[bool]) –

• logics (Optional[str]) –

• verbosity (Optional[int]) –

• optimizer (Optional[str]) –

• optimize_priority (Optional[str]) –

build_solution(z3_sol)

create and return a SchedulingSolution instance

check_sat(find_better_value=False)

check satisfiability. find_beter_value: the check_sat method is called from the incremental solver. Then we should not prompt that no solution can be found, but that no better solution can be found. Return * result as True (sat) or False (unsat, unknown). * the computation time.

Parameters:

find_better_value (Optional[bool]) –

create_objective()

create optimization objectives

Return type:

bool

export_to_smt2(smt_filename)

export the model to a smt file to be processed by another SMT solver

Parameters:

smt_filename (str) –

find_another_solution(variable)

let the solver find another solution for the variable

Parameters:

variable (ArithRef) –

Return type:

bool

get_parameters_description()

return the solver parameter names and values as a dict

print_assertions()

A utility method to display solver assertions

print_solution()

A utility method that displays all internal variables for the current solution

print_statistics()

A utility method that displays solver statistics

solve()

call the solver and returns the solution, if ever

Return type:

Union[bool, SchedulingSolution]

solve_optimize_incremental(variable, max_recursion_depth=None, kind='min')

target a min or max for a variable, without the Optimize solver. The loop continues ever and ever until the next value is more than 90%

Parameters:

• variable (ArithRef) –

• max_recursion_depth (Optional[int]) –

• kind (Optional[str]) –

Return type:

int

Solution

Solution definition and Gantt export.

class processscheduler.solution.BufferSolution(name)

Bases: object

Class to represent the solution for a Buffer.

Parameters:

name (str) –

class processscheduler.solution.ResourceSolution(name)

Bases: object

Class to represent the solution for the resource assignments.

Parameters:

name (str) –

class processscheduler.solution.SchedulingSolution(problem)

Bases: object

A class that represent the solution of a scheduling problem. Can be rendered to a matplotlib Gantt chart, or exported to json

add_indicator_solution(indicator_name, indicator_value)

Add indicator solution.

Parameters:

• indicator_name (str) –

• indicator_value (int) –

Return type:

None

add_resource_solution(resource_solution)

Add resource solution.

Parameters:

resource_solution (ResourceSolution) –

Return type:

None

add_task_solution(task_solution)

Add task solution.

Parameters:

task_solution (TaskSolution) –

Return type:

None

get_all_tasks_but_unavailable()

Return all tasks except those of the type UnavailabilityTask used to represent a ResourceUnavailable constraint.

get_scheduled_tasks()

Return scheduled tasks.

render_gantt_matplotlib(fig_size=(9, 6), show_plot=True, show_indicators=True, render_mode='Resource', fig_filename=None)

generate a gantt diagram using matplotlib. Inspired by https://www.geeksforgeeks.org/python-basic-gantt-chart-using-matplotlib/

Parameters:

• fig_size (Optional[Tuple[int, int]]) –

• show_plot (Optional[bool]) –

• show_indicators (Optional[bool]) –

• render_mode (Optional[str]) –

• fig_filename (Optional[str]) –

Return type:

None

render_gantt_plotly(fig_size=None, show_plot=True, show_indicators=True, render_mode='Resource', sort=None, fig_filename=None, html_filename=None)

Use plotly.create_gantt method, see https://plotly.github.io/plotly.py-docs/generated/plotly.figure_factory.create_gantt.html

Parameters:

• fig_size (Optional[Tuple[int, int]]) –

• show_plot (Optional[bool]) –

• show_indicators (Optional[bool]) –

• render_mode (Optional[str]) –

• sort (Optional[str]) –

• fig_filename (Optional[str]) –

• html_filename (Optional[str]) –

Return type:

None

to_json_string()

Export the solution to a json string.

Return type:

str

class processscheduler.solution.TaskSolution(name)

Bases: object

Class to represent the solution for a scheduled Task.

Parameters:

name (str) –

Indices and tables

• Index

• Module Index

• Search Page