labscheduler.solvers.priority_dispatching_framework module

TODO: add wikipedia links

This module is and implementation of the basic Priority Dispatcher (PD). It is the basis for all priority dispatching based scheduling algorithms. It provides the structures to schedule operations one after another (or multiple at a time), contains the logic which operations are allowed to be scheduled and manages this as actions (nomenclature from machine learning: action=act of adding a set of operations to the schedule). It is itself already a scheduling algorithm, but it is recommended to add some more intelligence by inheriting and changing the sort_actions() method.

The code excerpt is from the priority_dispatching_framework.py file and contains the implementation of the PDFramework class, which is a solver for the Job Shop Scheduling Problem (JSSP). The class inherits from the JSSPSolver class and implements the compute_schedule method, which takes an instance of the JSSP and computes a schedule for it. The method uses a priority dispatching heuristic algorithm to determine the order in which jobs should be processed on machines. The PDFramework class contains several instance variables, including the JSSP instance, a list of possible actions, a dictionary of machines used in the JSSP, and a list of job groups. The class also contains several methods, including sort_actions, which sorts the list of possible actions based on their priority, and step, which chooses and takes an action according to the current policies. The sort_actions method sorts the list of possible actions based on their priority. The method first sorts the list based on whether the action contains any dummy steps, with actions containing dummy steps having lower priority. The method then sorts the list based on whether the action is doable, with doable actions having higher priority. Finally, the method sorts the list based on whether the action has partially started, with actions that have partially started having the highest priority. The step method chooses and takes an action according to the current policies. The method first sorts the list of possible actions using the sort_actions method. The method then selects the first action in the sorted list and assigns it to a machine. The method updates the schedule and the list of possible actions based on the assigned action. The method continues to choose and take actions until there are no more possible actions. Overall, the PDFramework class implements a priority dispatching heuristic algorithm to solve the JSSP. The class uses several methods to sort the list of possible actions and choose the best action to assign to a machine. The algorithm is relatively simple and easy to implement, making it a popular choice for solving JSSP problems. However, the algorithm does not guarantee an optimal solution and may not be suitable for all JSSP problems.

(GitHub Copilot:)

class labscheduler.solvers.priority_dispatching_framework.Action(operations: list[~labscheduler.structures.Operation], machine: ~labscheduler.solvers.priority_dispatching_framework.UsedMachine, priority: float, assigned_tags: dict[str, ~labscheduler.solvers.priority_dispatching_framework.UsedMachine] = <factory>, min_start_job: ~datetime.datetime | None = None, min_start_machine: ~datetime.datetime | None = None, min_start: ~datetime.datetime | None = None, wait_job: ~datetime.timedelta = datetime.timedelta(0), wait_machine: ~datetime.timedelta = datetime.timedelta(0), doable: bool = True)

Bases: object

Represents an action in the sense of a scheduling agent: Deciding to add a certain (set of) operations to the schedule with intended/scheduled starting times on certain machines.

assigned_tags: dict[str, UsedMachine]

doable: bool = True

machine: UsedMachine

min_start: datetime | None = None

min_start_job: datetime | None = None

min_start_machine: datetime | None = None

operations: list[Operation]

priority: float

wait_job: timedelta = datetime.timedelta(0)

wait_machine: timedelta = datetime.timedelta(0)

class labscheduler.solvers.priority_dispatching_framework.JobGroup(in_moves: list[str], out_moves: list[str], priority_moves: list[str], started: bool = False, finished: bool = False)

Bases: object

Captures information on a special set of operations. Special means, that one operation has more than one direct predecessor and possibly multiple direct successors. The utility functions of this class help with scheduling such operation constellations correctly.

allowed(idx: str, possible_operation: list[str], past: list[str])

finished: bool = False

in_moves: list[str]

is_partially_done(past: list[str]) → bool

Checks, whether any but not all the moves in this group have been done

out_moves: list[str]

priority_moves: list[str]

remove(idx: str) → None

removes the given operation from this group’s lists and sets the finished-flag if none is left

started: bool = False

update(past: list[str]) → None

removes all steps from its lists that have been done (are in past).

class labscheduler.solvers.priority_dispatching_framework.PDFramework

Bases: JSSPSolver

Copilot: Priority Dispatching Framework (PDF) is a heuristic algorithm for solving the Job Shop Scheduling Problem (JSSP). This class implements the PDF algorithm as a JSSP solver. The usual PDF constructs a schedule by adding operations one by one. In each step all operations with their preceding operations already scheduled are collected. Then the one with the highest priority is chosen and added to the schedule with the earliest possible starting time. This adding is called action. In our case an action cal also consist of adding multiple operations at once or the decision to scip an operation. Our addable operations are further restricted by capacity constraints. To more likely succeed, this framework applies the following rule: If one operation O has multiple preceding operations O_p, none of O_p shall be added to the schedule, before all of them are addable. After one of O_p was added, all of O_p have to be added, before any other operation is added. Afterward, O has to be added. This rule uses the class JobGroup. If an operation has no specified executing machine (only a machine type) and there are multiple machines of that type the heuristic will create multiple actions (one for adding the operation on each machine) and one will be chosen.

_abc_impl = <_abc._abc_data object>

compute_schedule(inst: JSSP, time_limit: float, offset: float, **kwargs) → tuple[dict[str, ScheduledAssignment] | None, SolutionQuality]

Computes the schedule for the given JSSP instance using the PDF algorithm.

Parameters:

• inst – JSSP instance to solve.

• time_limit – Time limit for the solver.

• offset – Offset for the solver.

• kwargs – Additional arguments.

Returns:

Schedule for the given JSSP instance.

current_group: JobGroup | None

static get_algorithm_info() → AlgorithmInfo

Returns information about the PDF algorithm.

Returns:

Information about the PDF algorithm.

get_executor(op: Operation, lab: MachineCollection) → list[UsedMachine]

Determines which machine should execute this operation. For non move operations, we have to check what the last preceding move to device of the required type was. This should be the executor… also this is not failsafe

get_origin(operation: MoveOperation) → str

Determines the or at least a possible source machine for this move :param operation: :return:

get_target(operation: MoveOperation, action: Action) → str

Determines the or at least a possible target machine for this move :param operation: :return:

groups: list[JobGroup]

is_doable(action: Action) → bool

Checks whether the given action is also possible in the current state of the machines. :param action: :return:

is_solvable(inst: JSSP) → bool

Determines whether the given JSSP instance is solvable using the PDF algorithm.

Parameters:

inst – JSSP instance to check.

Returns:

True if the given JSSP instance is solvable using the PDF algorithm, False otherwise.

jssp: JSSP

load: dict[str, int]

machine_by_name: dict[str, UsedMachine]

min_start: datetime

operation_to_group: dict[str, JobGroup | None]

possible_actions: list[Action]

reagent_names: list[str]

reset(offset: float = 10)

Prepares to compute a new schedule by resetting all the inner structures. :param offset: :return:

schedule: dict[str, ScheduledAssignment]

schedule_started_operations()

Adds all already started operations to the schedule and sets up the inner variables to build the complete schedule from there

set_min_start(action: Action, J: dict[str, Operation])

Determines what the earliest starting time for this action is.

Parameters:

• action – Action to determine the earliest starting time for.

• J – Dictionary of operations by ID.

sort_actions(action_list: list[Action])

Sorts the given list of actions according to the current policies.

Parameters:

action_list – List of actions to sort.

Returns:

Sorted version of the input list.

step() → bool

Chooses and takes an action according to the current policies. :return: Whether the step was successfully done

take_action(action: Action, assigned_machines: dict[str, str])

Adds the operations of this action to the schedule on the chosen machine :param action: :param assigned_machines: :return:

update_groups(action_taken: Action)

Updates the intern operation group classes according to the taken action. :param action_taken: :return:

update_possible_actions(lab: MachineCollection, J: dict[str, Operation])

Updates the list of possible actions assuming the given action has been taken. :param lab: :param J: :return:

class labscheduler.solvers.priority_dispatching_framework.UsedMachine(*args, **kwargs)

Bases: Machine

Represents a machine on which operations get scheduled. It is meant to be a utility class for organizing the schedule on one particular machine.

add(operation: Operation, start) → None

Adds the given operation to the machines schedule at the given time :param operation: :param start: :return:

min_start() → datetime

Determines when this machine is able to do its next operation. :return: A datetime. If there are no restrictions from the machine side, it’s datetime.today()

min_transfer_time() → datetime

Determines, when the machine will be able to receive/give labware. Some machines can not receive labware while working. :return: A datetime. the default return value is datetime.today() if no restrictions are in place

waiting(start: datetime) → float

Computes the time, the machine will be idle until the given point in time :param start: :return: waiting time in seconds