job_shop_lib.reinforcement_learning

Contains reinforcement learning components.

SingleJobShopGraphEnv	A Gymnasium environment for solving a specific instance of the Job Shop Scheduling Problem represented as a graph.
MultiJobShopGraphEnv	Gymnasium environment for solving multiple Job Shop Scheduling Problems using reinforcement learning and Graph Neural Networks.
ObservationDict	A dictionary containing the observation of the environment.
ObservationSpaceKey	Enumeration of the keys for the observation space dictionary.
ResourceTaskGraphObservation	Observation wrapper that converts an observation following the ObservationDict format to a format suitable to PyG's [HeteroData](https://pytorch-geometric.readthedocs.io/en/latest/generated/torch_geometric.data.HeteroData.html).
ResourceTaskGraphObservationDict	Represents a dictionary for resource task graph observations.
RewardObserver	Base class for all reward functions.
MakespanReward	Dense reward function based on the negative makespan of the schedule.
IdleTimeReward	Dense reward function based on the negative idle time of the schedule.
RewardWithPenalties	Reward function that adds penalties to another reward function.
RenderConfig	Configuration needed to initialize the GanttChartCreator class.
add_padding	Adds padding to the array.
create_edge_type_dict	Organizes edges based on node types.
map_values	Maps values in an array using a mapping.
get_optimal_actions	Indicates if each action is optimal according to a OptimalOperationsObserver instance.
get_deadline_violation_penalty	Compute the penalty for a scheduled operation that violates its deadline.
get_due_date_violation_penalty	Compute the penalty for a scheduled operation that violates its due date.

class SingleJobShopGraphEnv(job_shop_graph, feature_observer_configs, reward_function_config=DispatcherObserverConfig(class_type=<class 'job_shop_lib.reinforcement_learning._reward_observers.MakespanReward'>, kwargs={}), graph_updater_config=DispatcherObserverConfig(class_type=<class 'job_shop_lib.graphs.graph_updaters._residual_graph_updater.ResidualGraphUpdater'>, kwargs={}), ready_operations_filter=<function filter_dominated_operations>, render_mode=None, render_config=None, use_padding=True)

Bases: Env

A Gymnasium environment for solving a specific instance of the Job Shop Scheduling Problem represented as a graph.

This environment manages the scheduling process for a single Job Shop instance, using a graph representation and various observers to track the state and compute rewards.

Observation Space:

A dictionary with the following keys:

• "removed_nodes": Binary vector indicating removed graph nodes.

• "edge_list": Matrix of graph edges in COO format.

• Feature matrices: Keys corresponding to the composite observer features (e.g., "operations", "jobs", "machines").

Action Space:

MultiDiscrete space representing (job_id, machine_id) pairs.

Render Modes:

• "human": Displays the current Gantt chart.

• "save_video": Saves a video of the complete Gantt chart.

• "save_gif": Saves a GIF of the complete Gantt chart.

dispatcher

Manages the scheduling process. See Dispatcher.

composite_observer

A CompositeFeatureObserver which aggregates features from multiple observers.

graph_updater

Updates the graph representation after each action. See GraphUpdater.

reward_function

Computes rewards for actions taken. See RewardObserver.

action_space

Defines the action space. The action is a tuple of two integers (job_id, machine_id). The machine_id can be -1 if the selected operation can only be scheduled in one machine.

observation_space

Defines the observation space. The observation is a dictionary with the following keys:

• "removed_nodes": Binary vector indicating removed graph nodes.

• "edge_list": Matrix of graph edges in COO format.

• Feature matrices: Keys corresponding to the composite observer

  features (e.g., "operations", "jobs", "machines").

render_mode

The mode for rendering the environment ("human", "save_video", "save_gif").

gantt_chart_creator

Creates Gantt chart visualizations. See GanttChartCreator.

use_padding

Whether to use padding in observations. Padding maintains the observation space shape when the number of nodes changes.

Parameters:

• job_shop_graph (JobShopGraph) -- The JobShopGraph instance representing the job shop problem.

• feature_observer_configs (Sequence[str | FeatureObserverType | type[FeatureObserver] | DispatcherObserverConfig[type[FeatureObserver]] | DispatcherObserverConfig[FeatureObserverType] | DispatcherObserverConfig[str]]) -- A list of FeatureObserverConfig instances for the feature observers.

• reward_function_config (DispatcherObserverConfig[type[RewardObserver]]) -- The configuration for the reward function.

• graph_updater_config (DispatcherObserverConfig[type[GraphUpdater]]) -- The configuration for the graph updater.

• ready_operations_filter (Callable[[Dispatcher, list[Operation]], list[Operation]] | None) -- The function to use for pruning dominated operations.

• render_mode (str | None) -- The mode for rendering the environment ("human", "save_video", "save_gif").

• render_config (RenderConfig | None) -- Configuration for rendering (e.g., paths for saving videos or GIFs). See RenderConfig.

• use_padding (bool) -- Whether to use padding in observations. Padding maintains the observation space shape when the number of nodes changes.

metadata: dict[str, Any] = {'render_modes': ['human', 'save_video', 'save_gif']}

property instance: JobShopInstance

Returns the instance the environment is working on.

property job_shop_graph: JobShopGraph

Returns the job shop graph.

current_makespan()

Returns current makespan of partial schedule.

Return type:

int

machine_utilization()

Returns utilization percentage for each machine.

Returns:

Utilization percentage for each machine as a numpy array.

Return type:

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

Deprecated since version 1.1.2: This method is deprecated and will be removed in version 2.0.0.

reset(*, seed=None, options=None)

Resets the environment.

Parameters:

• seed (int | None) -- Added to match the signature of the parent class. It is not used in this method.

• options (dict[str, Any] | None) -- Additional options to pass to the environment. Not used in this method.

Returns:

• The observation of the environment.

• A dictionary with additional information, keys

  include: "feature_names", the names of the features in the observation; and "available_operations_with_ids", a list of available a list of available actions in the form of (operation_id, machine_id, job_id).

Return type:

A tuple containing the following elements

step(action)

Takes a step in the environment.

Parameters:

action (tuple[int, int]) -- The action to take. The action is a tuple of two integers (job_id, machine_id): the job ID and the machine ID in which to schedule the operation.

Returns:

• The observation of the environment.

• The reward obtained.

• Whether the environment is done.

• Whether the episode was truncated (always False).

• A dictionary with additional information. The dictionary contains the following keys: "feature_names", the names of the features in the observation; and "available_operations_with_ids", a list of available actions in the form of (operation_id, machine_id, job_id).

Return type:

A tuple containing the following elements

get_observation()

Returns the current observation of the environment.

Return type:

ObservationDict

render()

Renders the environment.

The rendering mode is set by the render_mode attribute:

• human: Renders the current Gannt chart.

• save_video: Saves a video of the Gantt chart. Used only if the

  schedule is completed.

• save_gif: Saves a GIF of the Gantt chart. Used only if the schedule

  is completed.

get_available_actions_with_ids()

Returns a list of available actions in the form of (operation_id, machine_id, job_id).

Return type:

list[tuple[int, int, int]]

validate_action(action)

Validates that the action is legal in the current state.

Parameters:

action (tuple[int, int]) -- The action to validate. The action is a tuple of two integers (job_id, machine_id).

Raises:

ValidationError -- If the action is invalid.

Return type:

None

class MultiJobShopGraphEnv(instance_generator, feature_observer_configs, graph_initializer=<function build_resource_task_graph>, graph_updater_config=DispatcherObserverConfig(class_type=<class 'job_shop_lib.graphs.graph_updaters._residual_graph_updater.ResidualGraphUpdater'>, kwargs={}), ready_operations_filter=<function filter_dominated_operations>, reward_function_config=DispatcherObserverConfig(class_type=<class 'job_shop_lib.reinforcement_learning._reward_observers.MakespanReward'>, kwargs={}), render_mode=None, render_config=None, use_padding=True)

Bases: Env

Gymnasium environment for solving multiple Job Shop Scheduling Problems using reinforcement learning and Graph Neural Networks.

This environment generates a new Job Shop Scheduling Problem instance for each reset, creates a graph representation, and manages the scheduling process using a Dispatcher.

The observation space includes:

• removed_nodes: Binary vector indicating removed nodes.

• edge_index: Edge list in COO format.

• operations: Matrix of operation features.

• jobs: Matrix of job features (if applicable).

• machines: Matrix of machine features (if applicable).

Internally, the class creates a SingleJobShopGraphEnv environment to manage the scheduling process for each JobShopInstance.

instance_generator

A InstanceGenerator that generates a new problem instance on each reset.

action_space

gymnasium.spaces.Discrete) action space with size equal to the maximum number of jobs.

observation_space

Dictionary of observation spaces. Keys are defined in ObservationSpaceKey.

single_job_shop_graph_env

Environment for a specific Job Shop Scheduling Problem instance. See SingleJobShopGraphEnv.

graph_initializer

Function to create the initial graph representation. It should take a JobShopInstance as input and return a JobShopGraph.

render_mode

Rendering mode for visualization. Supported modes are:

• human: Renders the current Gannt chart.

• save_video: Saves a video of the Gantt chart. Used only if the schedule is completed.

• save_gif: Saves a GIF of the Gantt chart. Used only if the schedule is completed.

render_config

Configuration for rendering. See RenderConfig.

feature_observer_configs

List of DispatcherObserverConfig for feature observers.

reward_function_config

Configuration for the reward function. See DispatcherObserverConfig and RewardObserver.

graph_updater_config

Configuration for the graph updater. The graph updater is used to update the graph representation after each action. See DispatcherObserverConfig and GraphUpdater.

Parameters:

• instance_generator (InstanceGenerator) -- A InstanceGenerator that generates a new problem instance on each reset.

• feature_observer_configs (Sequence[DispatcherObserverConfig[type[FeatureObserver]] | DispatcherObserverConfig[FeatureObserverType] | DispatcherObserverConfig[str]]) -- Configurations for feature observers. Each configuration should be a DispatcherObserverConfig with a class type that inherits from FeatureObserver or a string or enum that represents a built-in feature observer.

• graph_initializer (Callable[[JobShopInstance], JobShopGraph]) -- Function to create the initial graph representation. If None, the default graph initializer is used: build_resource_task_graph().

• graph_updater_config (DispatcherObserverConfig[type[GraphUpdater]]) -- Configuration for the graph updater. The graph updater is used to update the graph representation after each action. If None, the default graph updater is used: ResidualGraphUpdater.

• ready_operations_filter (Callable[[Dispatcher, list[Operation]], list[Operation]]) -- Function to filter ready operations. If None, the default filter is used: filter_dominated_operations().

• reward_function_config (DispatcherObserverConfig[type[RewardObserver]]) -- Configuration for the reward function. If None, the default reward function is used: MakespanReward.

• render_mode (str | None) --

  Rendering mode for visualization. Supported modes are:

  • human: Renders the current Gannt chart.

  • save_video: Saves a video of the Gantt chart. Used only if

    the schedule is completed.

  • save_gif: Saves a GIF of the Gantt chart. Used only if the

    schedule is completed.

• render_config (RenderConfig | None) -- Configuration for rendering. See RenderConfig.

• use_padding (bool) -- Whether to use padding in observations. If True, all matrices are padded to fixed sizes based on the maximum instance size. Values are padded with -1, except for the "removed_nodes" key, which is padded with True, indicating that the node is removed.

property dispatcher: Dispatcher

Returns the current dispatcher instance.

property reward_function: RewardObserver

Returns the current reward function instance.

property ready_operations_filter: Callable[[Dispatcher, list[Operation]], list[Operation]] | None

Returns the current ready operations filter.

property use_padding: bool

Returns whether the padding is used.

property job_shop_graph: JobShopGraph

Returns the current job shop graph.

property instance: JobShopInstance

Returns the current job shop instance.

reset(*, seed=None, options=None)

Resets the environment and returns the initial observation.

Parameters:

• seed (int | None) -- Random seed for reproducibility.

• options (dict[str, Any] | None) -- Additional options for reset (currently unused).

Returns:

• ObservationDict: The initial observation of the environment.

• dict: An info dictionary containing additional information about the reset state. This may include details about the generated instance or initial graph structure.

Return type:

tuple[ObservationDict, dict[str, Any]]

step(action)

Takes a step in the environment.

Parameters:

action (tuple[int, int]) -- The action to take. The action is a tuple of two integers (job_id, machine_id): the job ID and the machine ID in which to schedule the operation.

Returns:

• The observation of the environment.

• The reward obtained.

• Whether the environment is done.

• Whether the episode was truncated (always False).

• A dictionary with additional information. The dictionary contains the following keys: "feature_names", the names of the features in the observation; and "available_operations_with_ids", a list of available actions in the form of (operation_id, machine_id, job_id).

Return type:

tuple[ObservationDict, float, bool, bool, dict[str, Any]]

render()

Compute the render frames as specified by render_mode during the initialization of the environment.

The environment's metadata render modes (env.metadata["render_modes"]) should contain the possible ways to implement the render modes. In addition, list versions for most render modes is achieved through gymnasium.make which automatically applies a wrapper to collect rendered frames.

Note

As the render_mode is known during __init__, the objects used to render the environment state should be initialised in __init__.

By convention, if the render_mode is:

• None (default): no render is computed.

• "human": The environment is continuously rendered in the current display or terminal, usually for human consumption. This rendering should occur during step() and render() doesn't need to be called. Returns None.

• "rgb_array": Return a single frame representing the current state of the environment. A frame is a np.ndarray with shape (x, y, 3) representing RGB values for an x-by-y pixel image.

• "ansi": Return a strings (str) or StringIO.StringIO containing a terminal-style text representation for each time step. The text can include newlines and ANSI escape sequences (e.g. for colors).

• "rgb_array_list" and "ansi_list": List based version of render modes are possible (except Human) through the wrapper, gymnasium.wrappers.RenderCollection that is automatically applied during gymnasium.make(..., render_mode="rgb_array_list"). The frames collected are popped after render() is called or reset().

Note

Make sure that your class's metadata "render_modes" key includes the list of supported modes.

Changed in version 0.25.0: The render function was changed to no longer accept parameters, rather these parameters should be specified in the environment initialised, i.e., gymnasium.make("CartPole-v1", render_mode="human")

Return type:

None

get_available_actions_with_ids()

Returns a list of available actions in the form of (operation_id, machine_id, job_id).

Return type:

list[tuple[int, int, int]]

validate_action(action)

Validates the action.

Parameters:

action (tuple[int, int]) -- The action to validate. The action is a tuple of two integers (job_id, machine_id): the job ID and the machine ID in which to schedule the operation.

Raises:

ValidationError -- If the action is invalid.

Return type:

None

class ObservationDict

Bases: _ObservationDictRequired, _ObservationDictOptional

A dictionary containing the observation of the environment.

Required fields:

removed_nodes: Binary vector indicating removed nodes. edge_index: Edge list in COO format.

Optional fields:

operations: Matrix of operation features. jobs: Matrix of job features. machines: Matrix of machine features.

removed_nodes: ndarray[tuple[int, ...], dtype[bool]]

edge_index: ndarray[tuple[int, ...], dtype[int32]]

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

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

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

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

Bases: str, Enum

Enumeration of the keys for the observation space dictionary.

REMOVED_NODES = 'removed_nodes'

EDGE_INDEX = 'edge_index'

OPERATIONS = 'operations'

JOBS = 'jobs'

MACHINES = 'machines'

class ResourceTaskGraphObservation(env)

Bases: ObservationWrapper, Generic[EnvType]

Observation wrapper that converts an observation following the ObservationDict format to a format suitable to PyG's [HeteroData](https://pytorch-geometric.readthedocs.io/en/latest/generated/torch_geometric.data.HeteroData.html).

In particular, the edge_index is converted into a edge_index_dict with keys (node_type_i, "to", node_type_j). The node_type_i and node_type_j are the node types of the source and target nodes, respectively.

Additionally, the node features are stored in a dictionary with keys corresponding to the node type names under the node_features_dict key.

The node IDs are mapped to local IDs starting from 0. The original_ids_dict contains the original node IDs before removing nodes.

global_to_local_id

A dictionary mapping global node IDs to local node IDs for each node type.

type_ranges

A dictionary mapping node type names to (start, end) index ranges.

Parameters:

env (EnvType) -- The environment to wrap.

step(action)

Takes a step in the environment.

Parameters:

action (tuple[int, int]) -- The action to take. The action is a tuple of two integers (job_id, machine_id): the job ID and the machine ID in which to schedule the operation.

Returns:

• The observation of the environment.

• The reward obtained.

• Whether the environment is done.

• Whether the episode was truncated (always False).

• A dictionary with additional information. The dictionary contains the following keys: "feature_names", the names of the features in the observation; and "available_operations_with_ids", a list of available actions in the form of (operation_id, machine_id, job_id).

Return type:

A tuple containing the following elements

reset(*, seed=None, options=None)

Resets the environment.

Parameters:

• seed (int | None) -- Added to match the signature of the parent class. It is not used in this method.

• options (dict | None) -- Additional options to pass to the environment. Not used in this method.

Returns:

• The observation of the environment.

• A dictionary with additional information, keys

  include: "feature_names", the names of the features in the observation; and "available_operations_with_ids", a list of available a list of available actions in the form of (operation_id, machine_id, job_id).

Return type:

A tuple containing the following elements

observation(observation)

Processes the observation data into the resource task graph format.

Parameters:

observation (ObservationDict) -- The observation dictionary. It must NOT have padding.

Returns:

• "edge_index_dict": A dictionary mapping edge types to edge index arrays.

• "node_features_dict": A dictionary mapping node type names to

  node feature arrays.

• "original_ids_dict": A dictionary mapping node type names to the original node IDs before removing nodes.

Return type:

A dictionary containing the following keys

property unwrapped: EnvType

Returns the unwrapped environment.

class ResourceTaskGraphObservationDict

Bases: TypedDict

Represents a dictionary for resource task graph observations.

edge_index_dict: dict[tuple[str, str, str], ndarray[tuple[int, ...], dtype[int32]]]

node_features_dict: dict[str, ndarray[tuple[int, ...], dtype[float32]]]

original_ids_dict: dict[str, ndarray[tuple[int, ...], dtype[int32]]]

class RewardObserver(dispatcher, *, subscribe=True)

Bases: DispatcherObserver

Base class for all reward functions.

Parameters:

• dispatcher (Dispatcher)

• subscribe (bool)

rewards

List of rewards calculated for each operation scheduled by the dispatcher.

property last_reward: float

Returns the reward of the last step or 0 if no rewards have been calculated.

reset()

Sets rewards attribute to a new empty list.

Return type:

None

class MakespanReward(dispatcher, *, subscribe=True)

Bases: RewardObserver

Dense reward function based on the negative makespan of the schedule.

The reward is calculated as the difference between the makespan of the schedule before and after the last operation was scheduled. The makespan is the time at which the last operation is completed.

Parameters:

dispatcher (Dispatcher)

current_makespan

Makespan of the schedule after the last operation was scheduled.

reset()

Sets rewards attribute to a new empty list.

Return type:

None

update(scheduled_operation)

Called when an operation is scheduled on a machine.

Parameters:

scheduled_operation (ScheduledOperation)

class IdleTimeReward(dispatcher, *, subscribe=True)

Bases: RewardObserver

Dense reward function based on the negative idle time of the schedule.

The reward is calculated as the difference between the idle time of the schedule before and after the last operation was scheduled. The idle time is the sum of the time between the end of the last operation and the start of the next operation.

Parameters:

• dispatcher (Dispatcher)

• subscribe (bool)

update(scheduled_operation)

Called when an operation is scheduled on a machine.

Parameters:

scheduled_operation (ScheduledOperation)

class RewardWithPenalties(dispatcher, *, base_reward_observer, penalty_function, subscribe=True)

Bases: RewardObserver

Reward function that adds penalties to another reward function.

The reward is calculated as the sum of the reward from another reward function and a penalty for each constraint violation (due dates and deadlines).

base_reward_observer

The base reward observer to use for calculating the reward.

penalty_function

A function that takes a scheduled operation and the dispatcher as input and returns the penalty for that operation.

Parameters:

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

• base_reward_observer (RewardObserver) -- The base reward observer to use for calculating the reward. It must use the same dispatcher as this reward observer. If it is subscribed to the dispatcher, it will be unsubscribed.

• penalty_function (Callable[[ScheduledOperation, Dispatcher], float]) -- A function that takes a scheduled operation and the dispatcher as input and returns the penalty for that operation.

• subscribe (bool) -- Whether to subscribe to the dispatcher upon initialization.

Raises:

ValidationError -- If the base reward observer does not use the same dispatcher as this reward observer.

Added in version 1.7.0.

See also

The following functions (along with functools.partial) can be used to create penalty functions:

• get_deadline_violation_penalty()

• get_due_date_violation_penalty()

reset()

Sets rewards attribute to a new empty list.

Return type:

None

update(scheduled_operation)

Called when an operation is scheduled on a machine.

Parameters:

scheduled_operation (ScheduledOperation)

class RenderConfig

Bases: TypedDict

Configuration needed to initialize the GanttChartCreator class.

partial_gantt_chart_plotter_config: PartialGanttChartPlotterConfig

video_config: VideoConfig

gif_config: GifConfig

add_padding(array, output_shape, padding_value=-1, dtype=None)

Adds padding to the array.

Pads the input array to the specified output shape with a given padding value. If the dtype is not specified, the dtype of the input array is used.

Parameters:

• array (ndarray[tuple[int, ...], dtype[Any]]) -- The input array to be padded.

• output_shape (tuple[int, ...]) -- The desired shape of the output array.

• padding_value (float) -- The value to use for padding. Defaults to -1.

• dtype (type[T] | None) -- The data type for the output array. Defaults to None, in which case the dtype of the input array is used.

Returns:

The padded array with the specified output shape.

Raises:

ValidationError -- If the output shape is smaller than the input shape.

Return type:

ndarray[tuple[int, ...], dtype[T]]

Examples:

>>> array = np.array([[1, 2], [3, 4]])
>>> add_padding(array, (3, 3))
array([[ 1,  2, -1],
       [ 3,  4, -1],
       [-1, -1, -1]])

>>> add_padding(array, (3, 3), padding_value=0)
array([[1, 2, 0],
       [3, 4, 0],
       [0, 0, 0]])

>>> bool_array = np.array([[True, False], [False, True]])
>>> add_padding(bool_array, (3, 3), padding_value=False, dtype=int)
array([[1, 0, 0],
       [0, 1, 0],
       [0, 0, 0]])

>>> add_padding(bool_array, (3, 3), dtype=int)
array([[ 1,  0, -1],
       [ 0,  1, -1],
       [-1, -1, -1]])

create_edge_type_dict(edge_index, type_ranges, relationship='to')

Organizes edges based on node types.

Parameters:

• edge_index (ndarray[tuple[int, ...], dtype[T]]) -- numpy array of shape (2, E) where E is number of edges

• type_ranges (dict[str, tuple[int, int]]) -- dict[str, tuple[int, int]] Dictionary mapping type names to their corresponding index ranges [start, end) in the edge_index array.

• relationship (str) -- A string representing the relationship type between nodes.

Returns:

A dictionary with keys (type_i, relationship, type_j) and values as edge indices

Return type:

dict[tuple[str, str, str], ndarray[tuple[int, ...], dtype[T]]]

map_values(array, mapping)

Maps values in an array using a mapping.

Parameters:

• array (ndarray[tuple[int, ...], dtype[T]]) -- An NumPy array.

• mapping (dict[int, int])

Returns:

A NumPy array where each element has been replaced by its corresponding value from the mapping.

Raises:

ValidationError -- If the array contains values that are not in the mapping.

Return type:

ndarray[tuple[int, ...], dtype[T]]

Examples

>>> map_values(np.array([1, 2, 3]), {1: 10, 2: 20, 3: 30})
array([10, 20, 30])

>>> map_values(np.array([1, 2]), {1: 10, 2: 10, 3: 30})
array([10, 10])

get_optimal_actions(optimal_ops_observer, available_operations_with_ids)

Indicates if each action is optimal according to a OptimalOperationsObserver instance.

Parameters:

• optimal_ops_observer (OptimalOperationsObserver) -- The observer that provides optimal operations.

• available_operations_with_ids (list[tuple[int, int, int]]) -- List of available operations with their

• IDs (operation_id, machine_id, job_id)

Returns:

A dictionary mapping each tuple (operation_id, machine_id, job_id) in the available actions to a binary indicator (1 if optimal, 0 otherwise).

Return type:

dict[tuple[int, int, int], int]

get_deadline_violation_penalty(scheduled_operation, unused_dispatcher, deadline_penalty_factor=10000)

Compute the penalty for a scheduled operation that violates its deadline.

Parameters:

• scheduled_operation (ScheduledOperation) -- The scheduled operation to evaluate.

• unused_dispatcher (Dispatcher) -- This argument is unused but included for compatibility with the penalty function signature.

• deadline_penalty_factor (float) -- Cost added for each operation that finishes after its deadline. Defaults to 10_000.

Returns:

The penalty for the scheduled operation if it violates its deadline, otherwise 0.

Return type:

float

Added in version 1.7.0.

get_due_date_violation_penalty(scheduled_operation, unused_dispatcher, due_date_penalty_factor=100)

Compute the penalty for a scheduled operation that violates its due date.

Parameters:

• scheduled_operation (ScheduledOperation) -- The scheduled operation to evaluate.

• unused_dispatcher (Dispatcher) -- This argument is unused but included for compatibility with the penalty function signature.

• due_date_penalty_factor (float) -- Cost added for each operation that finishes after its due date. Defaults to 100.

Returns:

The penalty for the scheduled operation if it violates its due date, otherwise 0.

Return type:

float

Added in version 1.7.0.