jobshoplab.utils.state_machine_utils package¶
Submodules¶
jobshoplab.utils.state_machine_utils.buffer_type_utils module¶
- get_output_buffers(instance)[source]¶
Get all output buffer configurations from the instance.
This function filters all buffers in the instance to return only those with the OUTPUT role, which are designated as final destinations for completed jobs in the job shop scheduling system.
- Parameters:¶
- instance : InstanceConfig¶
The instance configuration containing all buffers.
- Returns:¶
- A tuple of buffer configurations that have
the OUTPUT role. Used to determine where completed jobs should be transported to mark them as finished.
- Return type:¶
tuple[BufferConfig, …]
- get_buffer_config_by_id(buffers, buffer_id)[source]¶
Get a buffer config by its ID from a tuple of BufferConfig objects.
- Parameters:¶
- buffers : tuple[BufferConfig, ...]¶
A tuple of BufferConfig objects.
- buffer_id : str¶
The ID of the desired buffer.
- Returns:¶
The BufferConfig object with the specified ID.
- Return type:¶
- Raises:¶
InvalidValue – If the desired buffer is not found in the given tuple of buffers.
- is_job_in_buffer(buffer_state, job_id)[source]¶
Check if a job is in a buffer.
- put_in_buffer(buffer_state, buffer_config, job_state)[source]¶
Put a job in a buffer.
- Parameters:¶
- buffer_state : BufferState¶
The buffer to put the job in.
- job_state : JobState¶
The job to put in the buffer.
- buffer_config : BufferConfig¶
- Returns:¶
The updated buffer state.
- Return type:¶
- switch_buffer(instance, buffer_from_state, buffer_to_state, job_state)[source]¶
Switch a job from one buffer to another.
- Parameters:¶
- buffer_from_state : BufferState¶
The buffer to remove the job from.
- buffer_to_state : BufferState¶
The buffer to put the job in.
- job_state : JobState¶
The job to switch buffers.
- instance : InstanceConfig¶
- Returns:¶
The updated buffer states and job state.
- Return type:¶
- remove_from_buffer(buffer_state, job_id)[source]¶
Remove a job from a buffer.
- get_all_buffer_configs(instance)[source]¶
Get all buffer configurations from the instance configuration.
- Parameters:¶
- instance : InstanceConfig¶
The instance configuration.
- Returns:¶
A list of all buffer configurations.
- Return type:¶
- get_buffer_state_by_id(buffers, buffer_id)[source]¶
Get a buffer state by its ID from a tuple of BufferState objects.
- Parameters:¶
- buffers : tuple[BufferState, ...]¶
A tuple of BufferState objects.
- buffer_id : str¶
The ID of the desired buffer.
- Returns:¶
The BufferState object with the specified ID.
- Return type:¶
- Raises:¶
InvalidValue – If the desired buffer is not found in the given tuple of buffers.
- get_next_job_from_buffer(buffer_state, buffer_config)[source]¶
Get the next job ID to process based on buffer type.
- Parameters:¶
- buffer_state : BufferState¶
The current state of the buffer containing job IDs
- buffer_config : BufferConfig¶
The buffer configuration specifying type (FIFO, LIFO, etc.)
- Returns:¶
The job ID to process next, or None if no automatic ordering applies
- Return type:¶
str < /dev/null | None
- get_job_position_in_buffer(job_id, postbuffer_state)[source]¶
Get the position (index) of a job within a postbuffer.
- is_correct_position_for_buffer_type(job_position, buffer_length, buffer_type)[source]¶
Check if a job position is valid for pickup based on buffer type.
- job_in_correct_buffer_for_pickup(instance, buffer)[source]¶
Check if a job is in the correct buffer for pickup based on buffer type constraints. :param instance: The instance configuration containing buffer configurations :param buffer: The buffer to check against
- Parameters:¶
- instance : InstanceConfig¶
- buffer : BufferState¶
- Return type:¶
- is_job_ready_for_pickup_from_postbuffer(job_state, state, instance_config)[source]¶
Check if a job is ready for pickup from its postbuffer based on buffer type constraints.
jobshoplab.utils.state_machine_utils.component_type_utils module¶
jobshoplab.utils.state_machine_utils.core_utils module¶
- is_done(state, instance)[source]¶
Check if the state machine is done by verifying all jobs are in output buffers.
This function determines completion based on job locations rather than operation states. A job shop system is considered complete when all jobs have been transported to their final destination (output buffers), not just when their operations are finished. This ensures proper material flow and system closure.
- Parameters:¶
- state : State¶
The current state of the state machine containing all jobs.
- instance : InstanceConfig¶
The instance configuration containing buffer definitions.
- Returns:¶
- True if all jobs are located in output buffers, False otherwise.
This indicates whether the entire job shop scheduling process is complete.
- Return type:¶
- is_machine_transition_from_idle_to_setup(machine, transition)[source]¶
- Parameters:¶
- machine : MachineState¶
- transition : ComponentTransition¶
- Return type:¶
- is_machine_transition_from_setup_to_working(machine, transition)[source]¶
- Parameters:¶
- machine : MachineState¶
- transition : ComponentTransition¶
- Return type:¶
- is_machine_transition_from_working_to_outage(machine, transition)[source]¶
- Parameters:¶
- machine : MachineState¶
- transition : ComponentTransition¶
- Return type:¶
- is_machine_transition_from_outage_to_idle(machine, transition)[source]¶
- Parameters:¶
- machine : MachineState¶
- transition : ComponentTransition¶
- Return type:¶
- is_transport_transition_from_idle_to_working(transport, transition)[source]¶
- Parameters:¶
- transport : TransportState¶
- transition : ComponentTransition¶
- Return type:¶
- is_transport_transition_from_pickup_to_transit(transport, transition)[source]¶
- Parameters:¶
- transport : TransportState¶
- transition : ComponentTransition¶
- Return type:¶
- is_transport_transition_from_pickup_to_waitingpickup(transport, transition)[source]¶
- Parameters:¶
- transport : TransportState¶
- transition : ComponentTransition¶
- Return type:¶
- is_transport_transition_from_waitingpickup_to_transit(transport, transition)[source]¶
- Parameters:¶
- transport : TransportState¶
- transition : ComponentTransition¶
- Return type:¶
- is_transport_transition_from_working_to_outage(transport, transition)[source]¶
- Parameters:¶
- transport : TransportState¶
- transition : ComponentTransition¶
- Return type:¶
- is_transport_transition_from_transit_to_outage(transport, transition)[source]¶
- Parameters:¶
- transport : TransportState¶
- transition : ComponentTransition¶
- Return type:¶
- is_transport_transition_from_outage_to_idle(transport, transition)[source]¶
- Parameters:¶
- transport : TransportState¶
- transition : ComponentTransition¶
- Return type:¶
- is_transport_transition_from_waiting_pickup_waiting_pickup(transport, transition)[source]¶
- Parameters:¶
- transport : TransportState¶
- transition : ComponentTransition¶
- Return type:¶
- sorted_by_transport(transitions)[source]¶
- Parameters:¶
- transitions : tuple[ComponentTransition, ...]¶
- Return type:¶
- create_transport_location_from_job(current_dest, pickup_dest, dropoff_dest)[source]¶
- sorted_done_operations(job_states)[source]¶
jobshoplab.utils.state_machine_utils.dispatch_rules_uitls module¶
- get_inbuffer_jobs_for_machine(instance, state, machine_id)[source]¶
- get_prebuffer_operations_for_machine(instance, state, machine_id)[source]¶
- find_job_state_from_operation(state, operation_id)[source]¶
- find_job_config_from_operation(instance, operation_id)[source]¶
- Parameters:¶
- instance : InstanceConfig¶
- operation_id : str¶
-
get_machine_prebuffer_jobs_by_processing_time(instance, state, mode=
'SPT')[source]¶ Gets jobs for machines based on the processing time of their next operation.
- Parameters:¶
- instance : InstanceConfig¶
- state : State¶
-
get_machine_jobs_by_remaining_processing_time(instance, state, mode=
'srpt')[source]¶ Assigns jobs to machines based on Shortest/Longest Remaining Processing Time.
- Parameters:¶
- instance : InstanceConfig¶
- state : State¶
- get_all_outbuffer_jobs(instance, state)[source]¶
Retrieves all jobs in outbuffer and postbuffer across machines.
- Parameters:¶
- instance : InstanceConfig¶
- state : State¶
- get_job_with_most_remaining_processing_time(instance, job_states)[source]¶
Returns the job with the most remaining processing time.
-
get_job_by_operation_count(job_states, find_max=
True)[source]¶ Returns the job with the most or least remaining operations.
jobshoplab.utils.state_machine_utils.job_type_utils module¶
- get_job_config_by_id(jobs, job_id)[source]¶
Get a job config by its ID from a tuple of JobConfig objects.
- get_operation_state_by_id(jobs, operation_id)[source]¶
Get an operation by its ID from a tuple of JobState objects.
- get_operation_config_by_id(jobs, operation_id)[source]¶
Get an operation config by its ID from a tuple of JobConfig objects.
- get_next_not_done_operation(job)[source]¶
Get the next operation from a job.
- Parameters:¶
- Returns:¶
The next operation from the job.
- Return type:¶
- Raises:¶
InvalidValue – If the job has no more operations.
- get_next_idle_operation(job)[source]¶
Get the next idle operation from a job that needs to be executed.
This function searches for the first operation in the job that is in IDLE state, meaning it’s ready to be processed. Unlike get_next_not_done_operation, this function returns None if no idle operations exist, allowing graceful handling of jobs where all operations are complete or in progress.
- get_prior_executed_operation(job)[source]¶
Get the last operation from a job that is in the DONE state. :param job: The job to get the last operation from. :type job: JobState
- Returns:¶
The last operation from the job that is in the DONE state.
- Return type:¶
- Raises:¶
InvalidValue – If the job has no operations in the DONE state.
- Parameters:¶
- get_processing_operation(job)[source]¶
Get the active operation from a job.
- Parameters:¶
- Returns:¶
The active operation from the job.
- Return type:¶
- Raises:¶
InvalidValue – If the job has no active operations.
- group_operations_by_state(job_states)[source]¶
Group all operations from job states by their operation state state.
- is_done(job, instance)[source]¶
Check if a job is completely done in the job shop system.
A job is considered done when two conditions are met: 1. All operations are completed (DONE state) 2. The job has been transported to an output buffer
This dual-condition approach ensures that job completion includes both processing completion and proper material flow to the final destination.
- Parameters:¶
- job : JobState¶
The job to check for completion.
- instance : InstanceConfig¶
The instance configuration containing buffer definitions.
- Returns:¶
- True if the job is completely done (operations complete AND in output buffer),
False otherwise. This indicates full job shop process completion for the job.
- Return type:¶
- all_operations_done(job)[source]¶
Check if all operations in a single job are in DONE state.
This function verifies that every operation within a job has been completed by checking their state. It’s used as a prerequisite for job completion but doesn’t guarantee the job is fully done (transportation to output buffer is also required).
- no_operation_idle(job)[source]¶
Check if no operations in a job are in IDLE state.
This function determines if a job has any remaining operations that need to be executed. When this returns True, it means all operations are either in progress (PROCESSING) or completed (DONE), indicating the job may need transportation to the output buffer rather than to another machine for processing.
jobshoplab.utils.state_machine_utils.logisitcs_type_uitls module¶
jobshoplab.utils.state_machine_utils.machine_type_utils module¶
- get_machine_state_by_id(machines, machine_id)[source]¶
Get a machine by its ID from a tuple of MachineState objects.
- Parameters:¶
- machines : tuple[MachineState, ...]¶
A tuple of MachineState objects.
- machine_id : str¶
The ID of the desired machine.
- Returns:¶
The MachineState object with the specified ID.
- Return type:¶
- Raises:¶
InvalidValue – If the desired machine is not found in the given tuple of machines.
- get_machine_config_by_id(machine_configs, machine_id)[source]¶
Get a machine config by its ID from a tuple of MachineConfig objects.
- Parameters:¶
- machine_configs : tuple[MachineConfig, ...]¶
A tuple of MachineConfig objects.
- machine_id : str¶
The ID of the desired machine.
- Returns:¶
The MachineConfig object with the specified ID.
- Return type:¶
- Raises:¶
InvalidValue – If the desired machine config is not found in the given tuple of machine configs.
- get_machine_id_from_buffer(machine_configs, buffer_id)[source]¶
Get the Machine where the buffer is located.
Checks if buffer is prebuffer, postbuffer or buffer and returns the corresponding machine id.
- get_buffer_state_from_machine(machine_state, buffer_id)[source]¶
Get the buffer state from a machine state.
- replace_buffer_state_in_machine(machine_state, buffer_state)[source]¶
Update the buffer state in a machine state.
- Parameters:¶
- machine_state : MachineState¶
- buffer_state : BufferState¶
- Return type:¶
jobshoplab.utils.state_machine_utils.outage_utils module¶
- get_new_outage_states(component, instance, current_time)[source]¶
Get the time until the transport is available again.
- Parameters:¶
- component : TransportState | MachineState¶
- instance : InstanceConfig¶
- current_time : Time | NoTime¶
- Return type:¶
tuple[OutageState, …]
- get_occupied_time_from_outage_iterator(outage_states)[source]¶
Get the occupied time from the outage iterator.
jobshoplab.utils.state_machine_utils.possible_transition_utils module¶
- is_job_next_operation_free(job_state)[source]¶
Checks if a operation at a job is possible -> Job is not finished -> Job is not active
- is_job_at_machine(job_state, machine_state)[source]¶
Check if a job is at a machine.
- Parameters:¶
- job_state : JobState¶
The job to check.
- machine : MachineState
The machine to check.
- machine_state : MachineState¶
- Returns:¶
True if the job is at the machine, False otherwise.
- Return type:¶
- is_action_possible(job_state, state, instance)[source]¶
Checks if a action at a job is possible -> Machine action -> Transport action
Returns: True if the operation is possible, False otherwise
- get_possible_transports(transport_states, transport_configs)[source]¶
Checks if the transport is possible
Returns: True if the transport is possible, False otherwise
- Parameters:¶
- transport_states : tuple[TransportState, ...]¶
- transport_configs : tuple[TransportTypeConfig, ...]¶
- Return type:¶
Generator[TransportState, None, None]
- get_num_possible_events(state, instance, config)[source]¶
Get the number of possible events
Returns: The number of possible events
- sort_by_id(components)[source]¶
- Parameters:¶
- components : tuple[JobState | TransportState, ...]¶
- Return type:¶
tuple[JobState | TransportState, …]
- is_transportable(job_state, state, instance)[source]¶
Determine if a job needs transportation in the job shop system.
This function implements the core logic for deciding when jobs require transport: 1. Jobs already at output buffers don’t need transport (fully complete) 2. Jobs with all operations done but not at output need transport to finish 3. Jobs with remaining operations need transport if they’re not at the correct machine 4. Jobs already at their next operation’s machine don’t need transport
- Parameters:¶
- Returns:¶
- True if the job needs transport (either to next operation or output buffer),
False if the job is already at the correct location or fully complete.
- Return type:¶
- Raises:¶
InvalidValue – If job state is inconsistent (no operations but not at output).
- is_early_transport(job_state, state, instance)[source]¶
Determine if a job is eligible for early transport in the job shop system.
An early transport is triggered once the job (operation) has started its processing cycle on a machine. Some systems allow scheduling a transport “early” meaning the AGV is “called” early to avoid waiting times. This can lead to deadlock situations in AGV traffic management. Hence some systems disallow early transport.
In this function we are checking if a job is early meaning not in the correct buffer location.
- get_possible_transport_transition(state, instance, allow_early_transport)[source]¶
Get all available transports and mach each transport with each possible job Jobs are possible when:
Job is in state working -> when finised it need to be transported to next op
Job is in state idle and next op is not the current location
Jobs that are not getting are already assigned to an agv
Jobs where all operations are done but not at the output buffer
Returns: The tuple of possible ComponentTransitions
- get_possible_transitions(state, instance)[source]¶
Get all possible transitions for the state
- Return type:¶
- replace_machine_state(state, machine_state)[source]¶
- get_component_by_id(components, id)[source]¶
- replace_components(state, updated_components)[source]¶
jobshoplab.utils.state_machine_utils.time_utils module¶
jobshoplab.utils.state_machine_utils.transport_type_utils module¶
- get_transport_state_by_id(transports, transport_id)[source]¶
Get a transport by its ID from a tuple of TransportState objects.
- Parameters:¶
- transports : tuple[TransportState, ...]¶
A tuple of TransportState objects.
- transport_id : str¶
The ID of the desired transport.
- Returns:¶
The TransportState object with the specified ID.
- Return type:¶
- Raises:¶
InvalidValue – If the desired transport is not found in the given tuple of transports.
- get_transport_config_by_id(transports, transport_id)[source]¶
Get a transport by its ID from a tuple of TransportConfig objects.
- Parameters:¶
- transports : tuple[TransportConfig, ...]¶
A tuple of TransportConfig objects.
- transport_id : str¶
The ID of the desired transport.
- Returns:¶
The TransportConfig object with the specified ID.
- Return type:¶
- Raises:¶
InvalidValue – If the desired transport is not found in the given tuple of transports.
- group_transports_by_state(transports)[source]¶
Group all transports by their state.
- Parameters:¶
- transports : tuple[TransportState, ...]¶
A tuple of TransportState objects.
- Returns:¶
A dictionary with the transport state as the key and a tuple of transports as the value.
- Return type:¶
dict[str, tuple[TransportState, …]]
- get_transport_id_by_job_id(state, job_id)[source]¶
Find which transport is currently assigned to handle a specific job.
- get_transport_state_by_job_id(state, job_id)[source]¶
Find which transport state is currently assigned to handle a specific job.