job_shop_lib.dispatching

Contains classes and functions to solve the Job Shop Scheduling Problem step-by-step.

Dispatcher	Handles the logic of scheduling operations on machines.
DispatcherObserver	Abstract class that allows objects to observe and respond to changes within the Dispatcher.
HistoryObserver	Observer that stores the history of the dispatcher.
UnscheduledOperationsObserver	Stores the operations that have not been dispatched yet.
OptimalOperationsObserver	Observer that identifies which available operations are optimal based on a reference schedule.
DispatcherObserverConfig	Configuration for initializing any type of class.
ReadyOperationsFilter	alias of Callable[[Dispatcher, list[Operation]], list[Operation]]
ReadyOperationsFilterType	Enumeration of ready operations filter types.
ready_operations_filter_factory	Creates and returns a filter function based on the specified filter strategy name.
filter_dominated_operations	Filters out all the operations that are dominated.
filter_non_immediate_machines	Filters out all the operations associated with machines which earliest operation is not the current time.
filter_non_idle_machines	Filters out all the operations associated with non-idle machines.
filter_non_immediate_operations	Filters out all the operations that can't start immediately.
create_composite_operation_filter	Creates and returns a ReadyOperationsFilter function by combining multiple filter strategies.
StartTimeCalculator	alias of Callable[[Dispatcher, Operation, int], int]
no_setup_time_calculator	Default start time calculator that implements the standard behavior.
get_machine_dependent_setup_time_calculator	Returns a start time calculator that adds setup times based on machine IDs.
get_matrix_setup_time_calculator	Returns a start time calculator that adds setup times based on a matrix of setup times.
get_breakdown_calculator	Returns a start time calculator that accounts for machine breakdowns.
get_job_dependent_setup_calculator	Factory for a calculator with sequence-dependent setup times.
get_arrival_calculator	Returns a start time calculator that respects operation arrival times.

Dispatching refers to the decision-making process of selecting which job should be processed next on a particular machine when that machine becomes available.

class Dispatcher(instance, ready_operations_filter=None, start_time_calculator=<function no_setup_time_calculator>)

Bases: object

Handles the logic of scheduling operations on machines.

This class allow us to just define the order in which operations are sequenced and the machines in which they are processed. It is then responsible for scheduling the operations on the machines and keeping track of the next available time for each machine and job.

The core method of the class are:

dispatch(operation[, machine_id])	Schedules the given operation on the given machine.
reset()	Resets the dispatcher to its initial state.

It also provides methods to query the state of the schedule and the operations:

current_time()	Returns the current time of the schedule.
available_operations()	Returns a list of available operations for processing, optionally filtering out operations using the filter function.
available_machines()	Returns the list of ready machines.
available_jobs()	Returns the list of ready jobs.
unscheduled_operations()	Returns the list of operations that have not been scheduled.
scheduled_operations()	Returns the list of operations that have been scheduled.
ongoing_operations()	Returns the list of operations that are currently being processed.
completed_operations()	Returns the set of operations that have been completed.
uncompleted_operations()	Returns the list of operations that have not been completed yet.
is_scheduled(operation)	Checks if the given operation has been scheduled.
is_ongoing(scheduled_operation)	Checks if the given operation is currently being processed.
next_operation(job_id)	Returns the next operation to be scheduled for the given job.
earliest_start_time(operation)	Calculates the earliest start time for a given operation based on machine and job constraints.
remaining_duration(scheduled_operation)	Calculates the remaining duration of a scheduled operation.

The above methods which do not take any arguments are cached to improve performance. After each scheduling operation, the cache is cleared.

Parameters:

• instance (JobShopInstance) -- The instance of the job shop problem to be solved.

• ready_operations_filter (Callable[[Dispatcher, list[Operation]], list[Operation]] | None) -- A function that filters out operations that are not ready to be scheduled. The function should take the dispatcher and a list of operations as input and return a list of operations that are ready to be scheduled. If None, no filtering is done.

• start_time_calculator (StartTimeCalculator) -- A function that calculates the start time for a given operation on a given machine. The function should take a dispatcher, an operation and a machine id as input and return the start time for the operation. Defaults to the standard calculation which is the maximum of the machine's next available time and the job's next available time. This allows for implementing context-dependent setup times, downtime, and machine breakdowns.

instance

The instance of the job shop problem to be scheduled.

schedule

The schedule of operations on machines.

ready_operations_filter

A function that filters out operations that are not ready to be scheduled.

start_time_calculator

A function that calculates the start time for a given operation on a given machine.

subscribers: list[DispatcherObserver]

A list of observers subscribed to the dispatcher.

property machine_next_available_time: list[int]

Returns the next available time for each machine.

property job_next_operation_index: list[int]

Returns the index of the next operation to be scheduled for each job.

property job_next_available_time: list[int]

Returns the next available time for each job.

subscribe(observer)

Subscribes an observer to the dispatcher.

Parameters:

observer (DispatcherObserver)

unsubscribe(observer)

Unsubscribes an observer from the dispatcher.

Parameters:

observer (DispatcherObserver)

reset()

Resets the dispatcher to its initial state.

Return type:

None

dispatch(operation, machine_id=None)

Schedules the given operation on the given machine.

The start time of the operation is computed based on the next available time for the machine and the next available time for the job to which the operation belongs. The operation is then scheduled on the machine and the tracking attributes are updated.

Parameters:

• operation (Operation) -- The operation to be scheduled.

• machine_id (int | None) -- The id of the machine on which the operation is to be scheduled. If None, the Operation's machine_id attribute is used.

Raises:

• ValidationError -- If the operation is not ready to be scheduled.

• UninitializedAttributeError -- If the operation has multiple machines in its list and no machine_id is provided.

Return type:

None

is_operation_ready(operation)

Returns True if the given operation is ready to be scheduled.

An operation is ready to be scheduled if it is the next operation to be scheduled for its job.

Parameters:

operation (Operation) -- The operation to be checked.

Return type:

bool

start_time(operation, machine_id)

Computes the start time for the given operation on the given machine.

Uses the configured start time calculator to compute the start time. By default, this is the maximum of the next available time for the machine and the next available time for the job to which the operation belongs.

Parameters:

• operation (Operation) -- The operation to be scheduled.

• machine_id (int) -- The id of the machine on which the operation is to be scheduled.

Return type:

int

create_or_get_observer(observer, condition=<function Dispatcher.<lambda>>, **kwargs)

Creates a new observer of the specified type or returns an existing observer of the same type if it already exists in the dispatcher's list of observers.

Parameters:

• observer (type[ObserverType]) -- The type of observer to be created or retrieved.

• condition (Callable[[DispatcherObserver], bool]) -- A function that takes an observer and returns True if it is the observer to be retrieved. By default, it returns True for all observers.

• **kwargs -- Additional keyword arguments to be passed to the observer's constructor.

Return type:

ObserverType

current_time()

Returns the current time of the schedule.

The current time is the minimum start time of the available operations.

Return type:

int

min_start_time(operations)

Returns the minimum start time of the available operations.

Parameters:

operations (list[Operation])

Return type:

int

available_operations()

Returns a list of available operations for processing, optionally filtering out operations using the filter function.

This method first gathers all possible next operations from the jobs being processed. It then optionally filters these operations using the filter function.

Returns:

A list of Operation objects that are available for scheduling.

Return type:

list[Operation]

raw_ready_operations()

Returns a list of available operations for processing without applying the filter function.

Returns:

A list of Operation objects that are available for scheduling based on precedence and machine constraints only.

Return type:

list[Operation]

unscheduled_operations()

Returns the list of operations that have not been scheduled.

Return type:

list[Operation]

scheduled_operations()

Returns the list of operations that have been scheduled.

Return type:

list[ScheduledOperation]

available_machines()

Returns the list of ready machines.

Return type:

list[int]

available_jobs()

Returns the list of ready jobs.

Return type:

list[int]

earliest_start_time(operation)

Calculates the earliest start time for a given operation based on machine and job constraints.

This method is different from the start_time method in that it takes into account every machine that can process the operation, not just the one that will process it. However, it also assumes that the operation is ready to be scheduled in the job in favor of performance.

Parameters:

operation (Operation) -- The operation for which to calculate the earliest start time.

Returns:

The earliest start time for the operation.

Return type:

int

remaining_duration(scheduled_operation)

Calculates the remaining duration of a scheduled operation.

The method computes the remaining time for an operation to finish, based on the maximum of the operation's start time or the current time. This helps in determining how much time is left from 'now' until the operation is completed.

Parameters:

scheduled_operation (ScheduledOperation) -- The operation for which to calculate the remaining time.

Returns:

The remaining duration.

Return type:

int

completed_operations()

Returns the set of operations that have been completed.

This method returns the operations that have been scheduled and the current time is greater than or equal to the end time of the operation.

Return type:

set[ScheduledOperation]

uncompleted_operations()

Returns the list of operations that have not been completed yet.

This method checks for operations that either haven't been scheduled or have been scheduled but haven't reached their completion time.

Note: The behavior of this method changed in version 0.5.0. Previously, it only returned unscheduled operations. For the old behavior, use the unscheduled_operations method.

Return type:

list[Operation]

ongoing_operations()

Returns the list of operations that are currently being processed.

This method returns the operations that have been scheduled and are currently being processed by the machines.

Return type:

list[ScheduledOperation]

is_scheduled(operation)

Checks if the given operation has been scheduled.

Parameters:

operation (Operation)

Return type:

bool

is_ongoing(scheduled_operation)

Checks if the given operation is currently being processed.

Parameters:

scheduled_operation (ScheduledOperation)

Return type:

bool

next_operation(job_id)

Returns the next operation to be scheduled for the given job.

Parameters:

job_id (int) -- The id of the job for which to return the next operation.

Raises:

ValidationError -- If there are no more operations left for the job.

Return type:

Operation

class DispatcherObserver(dispatcher, *, subscribe=True)

Bases: ABC

Abstract class that allows objects to observe and respond to changes within the Dispatcher.

It follows the Observer design pattern, where observers subscribe to the dispatcher and receive updates when certain events occur, such as when an operation is scheduled or when the dispatcher is reset.

dispatcher

The Dispatcher instance to observe.

Parameters:

• dispatcher (Dispatcher) -- The Dispatcher instance to observe.

• subscribe (bool) -- If True, automatically subscribes the observer to the dispatcher when it is initialized. Defaults to True.

Raises:

ValidationError -- If is_singleton is True and an observer of the same type already exists in the dispatcher's list of subscribers.

Example:

from job_shop_lib.dispatching import DispatcherObserver, Dispatcher
from job_shop_lib import ScheduledOperation


class HistoryObserver(DispatcherObserver):
    def __init__(self, dispatcher: Dispatcher):
        super().__init__(dispatcher)
        self.history: list[ScheduledOperation] = []

    def update(self, scheduled_operation: ScheduledOperation):
        self.history.append(scheduled_operation)

    def reset(self):
        self.history = []

property is_singleton: bool

Returns whether this observer is a singleton.

This is a class attribute that determines whether only one instance of this observer type can be subscribed to the dispatcher.

abstract update(scheduled_operation)

Called when an operation is scheduled on a machine.

Parameters:

scheduled_operation (ScheduledOperation)

abstract reset()

Called when the dispatcher is reset.

class HistoryObserver(dispatcher, *, subscribe=True)

Bases: DispatcherObserver

Observer that stores the history of the dispatcher.

Parameters:

• dispatcher (Dispatcher)

• subscribe (bool)

update(scheduled_operation)

Called when an operation is scheduled on a machine.

Parameters:

scheduled_operation (ScheduledOperation)

reset()

Called when the dispatcher is reset.

class UnscheduledOperationsObserver(dispatcher, *, subscribe=True)

Bases: DispatcherObserver

Stores the operations that have not been dispatched yet.

This observer maintains a list of deques, each containing unscheduled operations for a specific job. It provides methods to access and manipulate unscheduled operations efficiently.

Parameters:

• dispatcher (Dispatcher)

• subscribe (bool)

property unscheduled_operations: Iterable[Operation]

An iterable of all unscheduled operations across all jobs.

property num_unscheduled_operations: int

The total number of unscheduled operations.

update(scheduled_operation)

Removes a scheduled operation from the unscheduled operations.

This method is called by the dispatcher when an operation is scheduled. It removes the operation from its job's deque of unscheduled operations.

Parameters:

scheduled_operation (ScheduledOperation) -- The operation that has been scheduled.

Return type:

None

reset()

Resets unscheduled operations to include all operations.

This method reinitializes the list of deques with all operations from all jobs in the instance.

Return type:

None

class OptimalOperationsObserver(dispatcher, reference_schedule, *, subscribe=True)

Bases: DispatcherObserver

Observer that identifies which available operations are optimal based on a reference schedule.

This observer compares the available operations at each step with a reference schedule to determine which operations would lead to the optimal solution. It can be used for training purposes or to analyze decision making in dispatching algorithms.

optimal_operations

Set of operations that are considered optimal based on the reference schedule.

reference_schedule

The reference schedule used to determine optimal operations.

Parameters:

• dispatcher (Dispatcher) -- The dispatcher instance to observe.

• reference_schedule (Schedule) -- A complete schedule that represents the optimal or reference solution.

• subscribe (bool) -- If True, automatically subscribes to the dispatcher.

Raises:

ValidationError -- If the reference schedule is incomplete or if it doesn't match the dispatcher's instance.

update(scheduled_operation)

Updates the optimal operations after an operation is scheduled.

Parameters:

scheduled_operation (ScheduledOperation) -- The operation that was just scheduled.

Return type:

None

reset()

Resets the observer to its initial state.

Return type:

None

class DispatcherObserverConfig(class_type, kwargs=<factory>)

Bases: Generic[T]

Configuration for initializing any type of class.

Useful for specifying the type of the DispatcherObserver and additional keyword arguments to pass to the dispatcher observer constructor while not containing the dispatcher argument.

Parameters:

• class_type (T) -- Type of the class to be initialized. It can be the class type, an enum value, or a string. This is useful for the creation of DispatcherObserver instances from the factory functions.

• kwargs (dict[str, Any]) -- Keyword arguments needed to initialize the class. It must not contain the dispatcher argument.

See also

• DispatcherObserver

• job_shop_lib.dispatching.feature_observers.feature_observer_factory()

class_type: T

Type of the class to be initialized. It can be the class type, an enum value, or a string. This is useful for the creation of DispatcherObserver instances from the factory functions.

kwargs: dict[str, Any]

Keyword arguments needed to initialize the class. It must not contain the dispatcher argument.

class ReadyOperationsFilterType(value, names=<not given>, *values, module=None, qualname=None, type=None, start=1, boundary=None)

Bases: str, Enum

Enumeration of ready operations filter types.

A filter function is used by the Dispatcher class to reduce the amount of available operations to choose from.

DOMINATED_OPERATIONS = 'dominated_operations'

NON_IMMEDIATE_MACHINES = 'non_immediate_machines'

NON_IDLE_MACHINES = 'non_idle_machines'

NON_IMMEDIATE_OPERATIONS = 'non_immediate_operations'

ready_operations_filter_factory(filter_name)

Creates and returns a filter function based on the specified filter strategy name.

The filter function filters operations based on certain criteria such as dominated operations, immediate machine operations, etc.

Parameters:

filter_name (str | ReadyOperationsFilterType | Callable[[Dispatcher, list[Operation]], list[Operation]]) -- The name of the filter function to be used. See ReadyOperationsFilterType for supported values. Alternatively, a custom filter function can be passed directly.

Returns:

A function that takes a Dispatcher instance and a list of Operation instances as input and returns a list of Operation instances based on the specified filter function.

Raises:

ValidationError -- If the filter_name argument is not recognized or is not supported.

Return type:

Callable[[Dispatcher, list[Operation]], list[Operation]]

filter_dominated_operations(dispatcher, operations)

Filters out all the operations that are dominated. An operation is dominated if there is another operation that ends before it starts on the same machine.

Parameters:

• dispatcher (Dispatcher)

• operations (list[Operation])

Return type:

list[Operation]

filter_non_immediate_machines(dispatcher, operations)

Filters out all the operations associated with machines which earliest operation is not the current time.

Parameters:

• dispatcher (Dispatcher)

• operations (list[Operation])

Return type:

list[Operation]

filter_non_idle_machines(dispatcher, operations)

Filters out all the operations associated with non-idle machines.

A machine is considered idle if there are no ongoing operations currently scheduled on it. This filter removes operations that are associated with machines that are busy (i.e., have at least one uncompleted operation).

Utilizes :meth:Dispatcher.ongoing_operations() to determine machine statuses.

Parameters:

• dispatcher (Dispatcher) -- The dispatcher object.

• operations (list[Operation]) -- The list of operations to filter.

Returns:

The list of operations that are associated with idle machines.

Return type:

list[Operation]

filter_non_immediate_operations(dispatcher, operations)

Filters out all the operations that can't start immediately.

An operation can start immediately if its earliest start time is the current time.

The current time is determined by the minimum start time of the operations.

Parameters:

• dispatcher (Dispatcher) -- The dispatcher object.

• operations (list[Operation]) -- The list of operations to filter.

Return type:

list[Operation]

create_composite_operation_filter(ready_operations_filters)

Creates and returns a ReadyOperationsFilter function by combining multiple filter strategies.

The composite filter function applies multiple filter strategies iteratively in the order they are specified in the list.

Parameters:

ready_operations_filters (Iterable[Callable[[Dispatcher, list[Operation]], list[Operation]] | str | ReadyOperationsFilterType]) -- A list of filter strategies to be used. Supported values are 'dominated_operations' and 'non_immediate_machines' or any Callable that takes a Dispatcher instance and a list of Operation instances as input and returns a list of Operation instances.

Returns:

A function that takes a Dispatcher instance and a list of Operation instances as input and returns a list of Operation instances based on the specified list of filter strategies.

Raises:

ValidationError -- If any of the filter strategies in the list are not recognized or are not supported.

Return type:

Callable[[Dispatcher, list[Operation]], list[Operation]]

no_setup_time_calculator(dispatcher, operation, machine_id)

Default start time calculator that implements the standard behavior.

The start time is the maximum of the machine's next available time, the job's next available time, and the operation's release date.

Parameters:

• dispatcher (Dispatcher) -- The dispatcher instance.

• operation (Operation) -- The operation to be scheduled.

• machine_id (int) -- The id of the machine on which the operation is to be scheduled.

Returns:

The start time for the operation on the given machine.

Return type:

int

get_machine_dependent_setup_time_calculator(setup_times, default=0)

Returns a start time calculator that adds setup times based on machine IDs.

Parameters:

• setup_times (dict[int, int]) -- Dictionary mapping machine_id to setup time in time units.

• default (int) -- Default setup time to use if a machine_id is not found in the setup_times dictionary.

Returns:

A start time calculator function that adds setup times.

Example

>>> setup_calc = get_machine_dependent_setup_time_calculator(
...     {0: 2, 1: 1, 2: 3}
... )
>>> dispatcher = Dispatcher(instance, start_time_calculator=setup_calc)

get_matrix_setup_time_calculator(setup_times)

Returns a start time calculator that adds setup times based on a matrix of setup times.

Parameters:

setup_times (Sequence[Sequence[int]] | ndarray[tuple[int, ...], dtype[integer]]) -- A 2D matrix where setup_times[i][j] is the setuptime for switching from operation i to operation j. Here, i and j are operation's IDs.

Returns:

A start time calculator function that adds setup times based on the matrix.

Return type:

Callable[[Dispatcher, Operation, int], int]

Example

>>> setup_calc = get_matrix_setup_time_calculator([[0, 2], [1, 0]])
>>> dispatcher = Dispatcher(instance, start_time_calculator=setup_calc)

get_breakdown_calculator(breakdowns)

Returns a start time calculator that accounts for machine breakdowns.

This calculator adjusts the start time of an operation based on when the machine is expected to be down due to breakdowns, maintenance, holidays, etc.

Parameters:

breakdowns (dict[int, list[tuple[int, int]]]) -- Dictionary mapping machine_id to list of (start_time, duration) tuples representing when the machine breaks down.

Returns:

A start time calculator function that accounts for breakdowns.

Example

>>> breakdown_calc = breakdown_calculator({0: [(5, 3)], 1: [(8, 2)]})
>>> dispatcher = Dispatcher(instance,
...                        start_time_calculator=breakdown_calc)

get_job_dependent_setup_calculator(same_job_setup=0, different_job_setup=4, initial_setup=0)

Factory for a calculator with sequence-dependent setup times.

This calculator determines setup time based on whether the current operation's job matches the job of the previously processed operation on the same machine.

Parameters:

• same_job_setup (int) -- Setup time when the previous operation on the machine was from the same job.

• different_job_setup (int) -- Setup time when the previous operation on the machine was from a different job.

• initial_setup (int) -- Setup time for the first operation on a machine or if the machine is currently idle.

Returns:

A start time calculator function that applies sequence-dependent setup times.

Example

>>> seq_dep_calc = sequence_dependent_setup_calculator(
...     same_job_setup=1, different_job_setup=4, initial_setup=1
... )
>>> # Assuming 'instance' is a JobShopInstance
>>> dispatcher = Dispatcher(
...     instance, start_time_calculator=seq_dep_calc
... )

get_arrival_calculator(arrival_times=None)

Returns a start time calculator that respects operation arrival times.

This calculator uses a predefined matrix of arrival times to ensure that no operation begins before its specified arrival time. If the arrival_times matrix isn't provided directly, the calculator attempts to retrieve it from the dispatcher's instance metadata using the key "arrival_times_matrix".

Parameters:

arrival_times (Sequence[Sequence[int]] | ndarray[tuple[int, ...], dtype[integer]] | None) -- A 2D matrix where arrival_times[i][j] is the arrival time for the operation at index j of job i. If None, the calculator will attempt to retrieve it from the dispatcher metadata.

Returns:

A start time calculator function that uses the arrival times.

Return type:

Callable[[Dispatcher, Operation, int], int]

Example

>>> arrival_calc = get_arrival_calculator([[0, 2], [1, 0]])
>>> dispatcher = Dispatcher(
...     instance, start_time_calculator=arrival_calc
... )

Modules

feature_observers	Contains FeatureObserver classes for observing features of the dispatcher.
rules	Contains the dispatching rules for the job shop scheduling problem.