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.
ReadyOperationsFilter	alias of Callable[[Dispatcher, list[Operation]], list[Operation]]
DispatcherObserverConfig	Configuration for initializing any type of class.
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.
create_composite_operation_filter	Creates and returns a composite filter based on the specified list of filters.
ReadyOperationsFilterType	Enumeration of ready operations filter types.
ready_operations_filter_factory	Creates and returns a filter function based on the specified filter strategy name.

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)

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.

Parameters:

• instance (JobShopInstance)

• ready_operations_filter (Callable[[Dispatcher, list[Operation]], list[Operation]] | None)

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.

__init__(instance, ready_operations_filter=None)

Initializes the object with the given instance.

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.

Return type:

None

instance

schedule

ready_operations_filter

subscribers: list[DispatcherObserver]

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)

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) -- The id of the machine on which the operation is to be scheduled.

Raises:

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

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.

The start time 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. If None, the start time is computed based on the next available time for the operation on any machine.

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

ready_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[Operation]

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[Operation]

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

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]

create_composite_operation_filter(ready_operations_filters)

Creates and returns a composite filter based on the specified list of filters.

The composite filter function filters operations based on the specified list of filter strategies.

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 :class:`~job_shop_lib.Operation`instances.

Returns:

A function that takes a Dispatcher instance and a list of Operation instances as input and returns a list of :class:`~job_shop_lib.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]]

class ReadyOperationsFilterType(value, names=None, *, 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'

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. Supported values are 'dominated_operations' and 'immediate_machine_operations'.

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]]

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.

Parameters:

• dispatcher (Dispatcher)

• subscribe (bool)

dispatcher

The Dispatcher instance to observe.

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 = []

__init__(dispatcher, *, subscribe=True)

Initializes the observer with the Dispatcher and subscribes to it.

Parameters:

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

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

Raises:

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

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 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)

• kwargs (dict[str, Any])

class_type

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.

Type:

job_shop_lib.dispatching._dispatcher_observer_config.T

kwargs

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

Type:

dict[str, Any]

class_type: T

kwargs: dict[str, Any]

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

Modules

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