job_shop_lib.dispatching.feature_observers

Contains FeatureObserver classes for observing features of the dispatcher.

FeatureObserver	Base class for feature observers.
FeatureType	Types of features that can be extracted.
CompositeFeatureObserver	Aggregates features from other FeatureObserver instances subscribed to the same Dispatcher by concatenating their feature matrices along the first axis (horizontal concatenation).
EarliestStartTimeObserver	Observer that adds a feature indicating the earliest start time of each operation, machine, and job in the graph.
IsReadyObserver	Feature creator that adds a binary feature indicating if the operation, machine or job is ready to be dispatched.
DurationObserver	Measures the remaining duration of operations, machines, and jobs.
IsScheduledObserver	Updates features based on scheduling operations.
PositionInJobObserver	Adds a feature indicating the position of operations in their respective jobs.
RemainingOperationsObserver	Adds a feature indicating the number of remaining operations for each job and machine.
IsCompletedObserver	Adds a binary feature indicating whether each operation, machine, or job has been completed.
DatesObserver	Observes time-related attributes of operations.
FeatureObserverType	Enumeration of the different feature observers.
feature_observer_factory	Creates and returns a FeatureObserver based on the specified FeatureObserver type.
FeatureObserverConfig	alias of DispatcherObserverConfig[type[FeatureObserver]] | DispatcherObserverConfig[FeatureObserverType] | DispatcherObserverConfig[str]

A FeatureObserver is a a subclass of DispatcherObserver that observes features related to operations, machines, or jobs in the dispatcher.

Attributes are stored in numpy arrays with a shape of (num_entities, feature_size), where num_entities is the number of entities being observed (e.g., operations, machines, or jobs) and feature_size is the number of values being observed for each entity.

The advantage of using arrays is that they can be easily updated in a vectorized manner, which is more efficient than updating each attribute individually. Furthermore, machine learning models can be trained on these arrays to predict the best dispatching decisions.

class FeatureObserver(dispatcher, *, subscribe=True, feature_types=None)

Bases: DispatcherObserver

Base class for feature observers.

A FeatureObserver is a a subclass of DispatcherObserver that observes features related to operations, machines, or jobs in the Dispatcher.

Attributes are stored in numpy arrays with a shape of (num_entities, feature_size), where num_entities is the number of entities being observed (e.g., operations, machines, or jobs) and feature_size is the number of values being observed for each entity.

The advantage of using arrays is that they can be easily updated in a vectorized manner, which is more efficient than updating each attribute individually. Furthermore, machine learning models can be trained on these arrays to predict the best dispatching decisions.

Arrays use the data type np.float32.

New FeatureObservers must inherit from this class, and re-define the class attributes _singleton (defualt ), _feature_size (default 1) and _supported_feature_types (default all feature types).

Feature observers are not singleton by default. This means that more than one instance of the same feature observer type can be subscribed to the dispatcher. This is useful when the first subscriber only observes a subset of the features, and the second subscriber observes a different subset of them. For example, the first subscriber could observe only the operation-related features, while the second subscriber could observe the jobs.

Parameters:

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

• subscribe (bool) -- If True, the observer is subscribed to the dispatcher upon initialization. Otherwise, the observer must be subscribed later or manually updated.

• feature_types (list[FeatureType] | FeatureType | None) -- A list of FeatureType or a single FeatureType that specifies the types of features to observe. They must be a subset of the class attribute supported_feature_types. If None, all supported feature types are tracked.

features

A dictionary of numpy arrays with the features. Each key is a FeatureType and each value is a numpy array with the features. The array has shape (num_entities, feature_size), where num_entities is the number of entities being observed (e.g., operations, machines, or jobs) and feature_size is the number of values being observed for each entity.

property feature_sizes: dict[FeatureType, int]

Returns the size of the features.

The size of the features is the number of values being observed for each entity. This corresponds to the second dimension of each array.

This number is typically one (e.g. measuring the duration of each operation), but some feature observers like the CompositeFeatureObserver may track more than one value.

property supported_feature_types: list[FeatureType]

Returns the supported feature types.

property feature_dimensions: dict[FeatureType, tuple[int, int]]

A dictionary containing the shape of each FeatureType.

initialize_features()

Initializes the features based on the current state of the dispatcher.

This method is automatically called after initializing the observer.

update(scheduled_operation)

Updates the features based on the scheduled operation.

By default, this method just calls initialize_features().

Parameters:

• ScheduledOperation -- The operation that has been scheduled.

• scheduled_operation (ScheduledOperation)

reset()

Sets features to zero and calls to :meth:initialize_features.

set_features_to_zero(exclude=None)

Sets all features to zero except for the ones specified in exclude.

Setting a feature to zero means that all values in the feature array are set to this value.

Parameters:

exclude (FeatureType | list[FeatureType] | None) -- A single FeatureType or a list of FeatureType that specifies the features that should not be set to zero. If None, all currently used features are set to zero.

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

Bases: str, Enum

Types of features that can be extracted.

OPERATIONS = 'operations'

MACHINES = 'machines'

JOBS = 'jobs'

class CompositeFeatureObserver(dispatcher, *, subscribe=True, feature_types=None, feature_observers=None)

Bases: FeatureObserver

Aggregates features from other FeatureObserver instances subscribed to the same Dispatcher by concatenating their feature matrices along the first axis (horizontal concatenation).

It provides also a custom __str__ method to display the features in a more readable way.

Parameters:

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

• subscribe (bool) -- If True, the observer is subscribed to the dispatcher upon initialization. Otherwise, the observer must be subscribed later or manually updated.

• feature_types (list[FeatureType] | FeatureType | None) -- A list of FeatureType or a single FeatureType that specifies the types of features to observe. They must be a subset of the class attribute supported_feature_types. If None, all supported feature types are tracked.

• feature_observers (list[FeatureObserver] | None) -- A list of FeatureObserver instances to aggregate features from. If None, all feature observers subscribed to the dispatcher are used.

See also

An example using this class can be found in the Feature Observers example.

Additionally, the class SingleJobShopGraphEnv uses this feature observer to aggregate features from multiple ones.

feature_observers

List of FeatureObserver instances to aggregate features from.

column_names: dict[FeatureType, list[str]]

Dictionary mapping FeatureType to a list of column names for the corresponding feature matrix. They are generated based on the class name of the FeatureObserver instance that produced the feature.

classmethod from_feature_observer_configs(dispatcher, feature_observer_configs, subscribe=True)

Creates the composite feature observer.

Parameters:

• dispatcher (Dispatcher) -- The dispatcher used to create the feature observers.

• feature_observer_configs (Sequence[str | FeatureObserverType | type[FeatureObserver] | DispatcherObserverConfig[type[FeatureObserver]] | DispatcherObserverConfig[FeatureObserverType] | DispatcherObserverConfig[str]]) -- The list of feature observer configuration objects.

• subscribe (bool) -- Whether to subscribe the CompositeFeatureObserver to the dispatcher.

Return type:

Self

property features_as_dataframe: dict[FeatureType, DataFrame]

Returns the features as a dictionary of pd.DataFrame instances.

initialize_features()

Initializes the features based on the current state of the dispatcher.

This method is automatically called after initializing the observer.

class EarliestStartTimeObserver(dispatcher, *, subscribe=True, feature_types=None)

Bases: FeatureObserver

Observer that adds a feature indicating the earliest start time of each operation, machine, and job in the graph.

The earliest start time of an operation refers to the earliest time at which the operation could potentially start without violating any constraints. This time is normalized by the current time (i.e., the difference between the earliest start time and the current time).

The earliest start time of a machine is the earliest start time of the next operation that can be scheduled on that machine.

Finally, the earliest start time of a job is the earliest start time of the next operation in the job.

Parameters:

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

• subscribe (bool) -- If True, the observer is subscribed to the dispatcher upon initialization. Otherwise, the observer must be subscribed later or manually updated.

• feature_types (list[FeatureType] | FeatureType | None) -- A list of FeatureType or a single FeatureType that specifies the types of features to observe. They must be a subset of the class attribute supported_feature_types. If None, all supported feature types are tracked.

earliest_start_times: ndarray[tuple[int, ...], dtype[float32]]

A 2D numpy array with the earliest start times of each operation. The array has shape (num_jobs, max_operations_per_job). The value at index (i, j) is the earliest start time of the j-th operation in the i-th job. If a job has fewer than the maximum number of operations in a job, the remaining values are set to np.nan. Similarly to JobShopInstance's durations_matrix_array() method.

update(scheduled_operation)

Recomputes the earliest start times and calls the initialize_features method.

The earliest start times is computed as the cumulative sum of the previous unscheduled operations in the job plus the maximum of the completion time of the last scheduled operation and the next available time of the machine(s) the operation is assigned.

After that, we substract the current time.

Parameters:

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

initialize_features()

Initializes the features based on the current state of the dispatcher.

class IsReadyObserver(dispatcher, *, subscribe=True, feature_types=None)

Bases: FeatureObserver

Feature creator that adds a binary feature indicating if the operation, machine or job is ready to be dispatched.

Parameters:

• dispatcher (Dispatcher)

• subscribe (bool)

• feature_types (list[FeatureType] | FeatureType | None)

initialize_features()

Initializes the features based on the current state of the dispatcher.

This method is automatically called after initializing the observer.

reset()

Sets features to zero and calls to :meth:initialize_features.

features

A dictionary of numpy arrays with the features. Each key is a FeatureType and each value is a numpy array with the features. The array has shape (num_entities, feature_size), where num_entities is the number of entities being observed (e.g., operations, machines, or jobs) and feature_size is the number of values being observed for each entity.

class DurationObserver(dispatcher, *, subscribe=True, feature_types=None)

Bases: FeatureObserver

Measures the remaining duration of operations, machines, and jobs.

The duration of an Operation is:

• if the operation has not been scheduled, it is the duration of the operation.

• if the operation has been scheduled, it is the remaining duration of the operation.

• if the operation has been completed, it is the last duration of the operation that has been computed. The duration must be set to 0 manually if needed. We do not update the duration of completed operations to save computation time.

The duration of a machine or job is the sum of the durations of the unscheduled operations that belong to the machine or job.

Parameters:

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

• subscribe (bool) -- If True, the observer is subscribed to the dispatcher upon initialization. Otherwise, the observer must be subscribed later or manually updated.

• feature_types (list[FeatureType] | FeatureType | None) -- A list of FeatureType or a single FeatureType that specifies the types of features to observe. They must be a subset of the class attribute supported_feature_types. If None, all supported feature types are tracked.

initialize_features()

Initializes the features based on the current state of the dispatcher.

This method is automatically called after initializing the observer.

update(scheduled_operation)

Updates the features based on the scheduled operation.

By default, this method just calls initialize_features().

Parameters:

• ScheduledOperation -- The operation that has been scheduled.

• scheduled_operation (ScheduledOperation)

features

A dictionary of numpy arrays with the features. Each key is a FeatureType and each value is a numpy array with the features. The array has shape (num_entities, feature_size), where num_entities is the number of entities being observed (e.g., operations, machines, or jobs) and feature_size is the number of values being observed for each entity.

class IsScheduledObserver(dispatcher, *, subscribe=True, feature_types=None)

Bases: FeatureObserver

Updates features based on scheduling operations.

This observer tracks which operations have been scheduled and updates feature matrices accordingly.

It updates a feature in the FeatureType.OPERATIONS() matrix to indicate that an operation has been scheduled.

Additionally, it counts the number of uncompleted but scheduled operations for each machine and job, updating the respective FeatureType.MACHINES() and FeatureType.JOBS() feature matrices.

Parameters:

• dispatcher (Dispatcher)

• subscribe (bool)

• feature_types (list[FeatureType] | FeatureType | None)

update(scheduled_operation)

Updates the features based on the scheduled operation.

By default, this method just calls initialize_features().

Parameters:

• ScheduledOperation -- The operation that has been scheduled.

• scheduled_operation (ScheduledOperation)

features

A dictionary of numpy arrays with the features. Each key is a FeatureType and each value is a numpy array with the features. The array has shape (num_entities, feature_size), where num_entities is the number of entities being observed (e.g., operations, machines, or jobs) and feature_size is the number of values being observed for each entity.

class PositionInJobObserver(dispatcher, *, subscribe=True, feature_types=None)

Bases: FeatureObserver

Adds a feature indicating the position of operations in their respective jobs.

Positions are adjusted dynamically as operations are scheduled. In other words, the position of an operation is the number of unscheduled operations that precede it in the job.

It only supports the OPERATIONS() feature type.

Parameters:

• dispatcher (Dispatcher)

• subscribe (bool)

• feature_types (list[FeatureType] | FeatureType | None)

initialize_features()

Initializes the features based on the current state of the dispatcher.

This method is automatically called after initializing the observer.

update(scheduled_operation)

Updates the features based on the scheduled operation.

By default, this method just calls initialize_features().

Parameters:

• ScheduledOperation -- The operation that has been scheduled.

• scheduled_operation (ScheduledOperation)

features

A dictionary of numpy arrays with the features. Each key is a FeatureType and each value is a numpy array with the features. The array has shape (num_entities, feature_size), where num_entities is the number of entities being observed (e.g., operations, machines, or jobs) and feature_size is the number of values being observed for each entity.

class RemainingOperationsObserver(dispatcher, *, subscribe=True, feature_types=None)

Bases: FeatureObserver

Adds a feature indicating the number of remaining operations for each job and machine.

It does not support FeatureType.OPERATIONS().

Parameters:

• dispatcher (Dispatcher)

• subscribe (bool)

• feature_types (list[FeatureType] | FeatureType | None)

initialize_features()

Initializes the features based on the current state of the dispatcher.

This method is automatically called after initializing the observer.

update(scheduled_operation)

Updates the features based on the scheduled operation.

By default, this method just calls initialize_features().

Parameters:

• ScheduledOperation -- The operation that has been scheduled.

• scheduled_operation (ScheduledOperation)

features

A dictionary of numpy arrays with the features. Each key is a FeatureType and each value is a numpy array with the features. The array has shape (num_entities, feature_size), where num_entities is the number of entities being observed (e.g., operations, machines, or jobs) and feature_size is the number of values being observed for each entity.

class IsCompletedObserver(dispatcher, *, feature_types=None, subscribe=True)

Bases: FeatureObserver

Adds a binary feature indicating whether each operation, machine, or job has been completed.

An operation is considered completed if it has been scheduled and the current time is greater than or equal to the sum of the operation's start time and duration.

A machine or job is considered completed if all of its operations have been completed.

Parameters:

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

• feature_types (list[FeatureType] | FeatureType | None) -- A list of FeatureType or a single FeatureType that specifies the types of features to observe. They must be a subset of the class attribute supported_feature_types. If None, all supported feature types are tracked.

• subscribe (bool) -- If True, the observer is subscribed to the dispatcher upon initialization. Otherwise, the observer must be subscribed later or manually updated.

initialize_features()

Initializes the features based on the current state of the dispatcher.

This method is automatically called after initializing the observer.

reset()

Sets features to zero and calls to :meth:initialize_features.

update(scheduled_operation)

Updates the features based on the scheduled operation.

By default, this method just calls initialize_features().

Parameters:

• ScheduledOperation -- The operation that has been scheduled.

• scheduled_operation (ScheduledOperation)

features

A dictionary of numpy arrays with the features. Each key is a FeatureType and each value is a numpy array with the features. The array has shape (num_entities, feature_size), where num_entities is the number of entities being observed (e.g., operations, machines, or jobs) and feature_size is the number of values being observed for each entity.

class DatesObserver(dispatcher, *, subscribe=True, attributes_to_observe=None, feature_types=None)

Bases: FeatureObserver

Observes time-related attributes of operations.

This observer tracks attributes like release date, deadline, and due date for each operation in the job shop instance. The attributes to be observed can be specified during initialization.

The values are stored in a numpy array of shape (num_operations, num_attributes), where num_attributes is the number of attributes being observed.

All attributes are updated based on the current time of the dispatcher so that they reflect the relative time remaining until the event occurs: (attribute - current_time). This means that the values will be negative if the event is in the past.

Parameters:

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

• subscribe (bool) -- If True, the observer is subscribed to the dispatcher upon initialization. Otherwise, the observer must be subscribed later or manually updated.

• attributes_to_observe (list[Literal['release_date', 'deadline', 'due_date']] | None) -- A list of attributes to observe. If None, all available time attributes (release_date, deadline, due_date) will be observed, provided they exist in the instance.

• feature_types (FeatureType | list[FeatureType] | None)

attributes_to_observe

List of attributes to observe.

property feature_sizes: dict[FeatureType, int]

Returns the size of the features.

The size of the features is the number of values being observed for each entity. This corresponds to the second dimension of each array.

This number is typically one (e.g. measuring the duration of each operation), but some feature observers like the CompositeFeatureObserver may track more than one value.

property attribute_map: dict[Literal['release_date', 'deadline', 'due_date'], int]

Maps attributes to their column index in the features array.

initialize_features()

Initializes the features for the operations.

This method sets up the features for the operations based on the attributes to observe. It creates a numpy array with the shape (num_operations, num_attributes) and fills it with the corresponding values from the job shop instance. Note that the matrix may contain np.nan values for operations that do not have deadlines or due dates.

See also

• job_shop_lib.JobShopInstance.release_dates_matrix_array()

• job_shop_lib.JobShopInstance.deadlines_matrix_array()

• job_shop_lib.JobShopInstance.due_dates_matrix_array()

update(scheduled_operation)

Updates the features based on the scheduled operation.

This method updates the features by subtracting the current time from the initial release date, deadline, and due date attributes of the operations.

Parameters:

scheduled_operation (ScheduledOperation) -- The scheduled operation that has just been processed. It is not used in this observer, but is required by the FeatureObserver.update() method.

reset()

Calls initialize_features()

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

Bases: str, Enum

Enumeration of the different feature observers.

Each FeatureObserver is associated with a string value that can be used to create the FeatureObserver using the factory function.

It does not include the CompositeFeatureObserver class since this observer is often managed separately from the others. For example, a common use case is to create a list of feature observers and pass them to

IS_READY = 'is_ready'

EARLIEST_START_TIME = 'earliest_start_time'

DURATION = 'duration'

IS_SCHEDULED = 'is_scheduled'

POSITION_IN_JOB = 'position_in_job'

REMAINING_OPERATIONS = 'remaining_operations'

IS_COMPLETED = 'is_completed'

feature_observer_factory(feature_observer_type, **kwargs)

Creates and returns a FeatureObserver based on the specified FeatureObserver type.

Parameters:

• feature_creator_type -- The type of FeatureObserver to create.

• **kwargs -- Additional keyword arguments to pass to the FeatureObserver constructor.

• feature_observer_type (str | FeatureObserverType | type[FeatureObserver] | DispatcherObserverConfig[type[FeatureObserver]] | DispatcherObserverConfig[FeatureObserverType] | DispatcherObserverConfig[str])

Returns:

A FeatureObserver instance.

Return type:

FeatureObserver