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.
FeatureObserverType	Enumeration of the different feature observers.
feature_observer_factory	Creates and returns a node feature creator based on the specified node feature creator 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. This is because most machine

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=None, *, 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[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[Any, 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.

remaining_ops_per_machine

The number of unscheduled operations per machine.

remaining_ops_per_job

The number of unscheduled operations per job.

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)

class FeatureObserverType(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)

Bases: str, Enum

Enumeration of the different feature observers.

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

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'

COMPOSITE = 'composite'

feature_observer_factory(feature_creator_type, **kwargs)

Creates and returns a node feature creator based on the specified node feature creator type.

Parameters:

• feature_creator_type (str | FeatureObserverType | type[FeatureObserver] | DispatcherObserverConfig[type[FeatureObserver]] | DispatcherObserverConfig[FeatureObserverType] | DispatcherObserverConfig[str]) -- The type of node feature creator to create.

• **kwargs -- Additional keyword arguments to pass to the node feature creator constructor.

Returns:

A node feature creator instance.

Return type:

FeatureObserver