Skip to content

[Feature] Add largest_processing_time dispatching rule #56

Open
@Pabloo22

Description

@Pabloo22

What is a Dispatching Rule?

In job_shop_lib, dispatching rules are functions that determine which operation to schedule next from a list of available operations. These rules are the core decision-making component in many job shop scheduling algorithms.

A dispatching rule takes a Dispatcher instance as input and returns the selected Operation to be scheduled next. The Dispatcher contains information about the current state of the schedule, available operations, and machine availabilities.

You can see examples of existing dispatching rules in the job_shop_lib/dispatching/rules/_dispatching_rules_functions.py file, such as shortest_processing_time_rule, most_work_remaining_rule, and others.

The Goal: A Largest Processing Time Rule

We need a new dispatching rule called largest_processing_time_rule that selects the operation with the longest duration from the available operations. This is essentially the opposite of the existing shortest_processing_time_rule.

The Largest Processing Time (LPT) rule is a well-known heuristic in scheduling theory. While it may seem counterintuitive, prioritizing longer operations first can sometimes lead to better overall schedules, particularly in scenarios where you want to handle the most time-consuming tasks early to avoid bottlenecks later.

The logic should be as follows:

  1. Get all available operations from the dispatcher
  2. Select the operation with the maximum duration
  3. If multiple operations have the same maximum duration, return any of them (consistent with how Python's max() function works)

Implementation Steps

Here's a clear path to get this done:

  1. Create the Dispatching Rule Function

    In the file job_shop_lib/dispatching/rules/_dispatching_rules_functions.py, create a new function with the following signature:

    def largest_processing_time_rule(dispatcher: Dispatcher) -> Operation:
        """Dispatches the operation with the longest duration."""
        # ... your implementation will go here
  2. Implement the Logic

    The implementation should be very similar to shortest_processing_time_rule, but instead of using min(), you'll use max(). Look at the existing function for reference:

    def largest_processing_time_rule(dispatcher: Dispatcher) -> Operation:
        """Dispatches the operation with the longest duration."""
        return max(
            dispatcher.available_operations(),
            key=lambda operation: operation.duration,
        )

    Note: The docstring should follow the existing style. Look at other dispatching rule functions for reference.

  3. Add to the Factory System

    Update job_shop_lib/dispatching/rules/_dispatching_rule_factory.py:

    • Add LARGEST_PROCESSING_TIME = "largest_processing_time" to the DispatchingRuleType enum
    • Import your new function at the top of the file
    • Add an entry to the dispatching_rules dictionary in the dispatching_rule_factory function
  4. Expose the New Function

    To make your new function easily importable, add it to:

    • The import statement in job_shop_lib/dispatching/rules/__init__.py
    • The __all__ list in the same file
    • The ".. autosummary" section in the module's docstring
  5. Write Tests

    Add tests to tests/dispatching/rules/test_dispatching_rules.py. Your test should cover:

    • The rule correctly selects the operation with the longest duration
    • The factory works as expected
    • It can be used in DispatchingRuleSolver by calling it's string representation defined in the Enum

    Example test structure:

    def test_largest_processing_time_rule(dispatcher: Dispatcher):
        """Tests that the largest_processing_time_rule selects the operation
        with the longest duration."""
        selected_operation = largest_processing_time_rule(dispatcher)
        available_operations = dispatcher.available_operations()
        max_duration = max(op.duration for op in available_operations)
        assert selected_operation.duration == max_duration
  6. Optional: Add Scoring Function

    For consistency with other rules, you might also want to add a corresponding scoring function:

    def largest_processing_time_score(dispatcher: Dispatcher) -> list[int]:
        """Scores each job based on the duration of the next operation.
        
        The score is the duration of the next operation in each job.
        This means that jobs with longer next operations will have higher scores.
        """
        # Implementation here

Expected Behavior

The largest_processing_time_rule should:

  • Select the operation with the maximum duration from dispatcher.available_operations()
  • Work seamlessly with the existing DispatchingRuleSolver
  • Be accessible via DispatchingRuleSolver("largest_processing_time")
  • Follow the same patterns as existing dispatching rules

Files to Modify

  1. job_shop_lib/dispatching/rules/_dispatching_rules_functions.py - Add the main function
  2. job_shop_lib/dispatching/rules/_dispatching_rule_factory.py - Add to enum and factory
  3. job_shop_lib/dispatching/rules/__init__.py - Export the function
  4. tests/dispatching/rules/test_dispatching_rules.py - Add comprehensive tests

Other Considerations

  • Ensure that you follow the CONTRIBUTING.md guidelines, including the instructions for commit messages.
  • Create a new branch for this issue from the latest version of the main branch.
  • Look at how shortest_processing_time_rule is implemented and tested as your template.
  • Make sure all existing tests still pass after your changes. We use Poetry 1.8.2.
  • Ask this issue to be assigned to you

Metadata

Metadata

Assignees

Labels

Projects

No projects

Milestone

No milestone

Relationships

None yet

Development

No branches or pull requests

Issue actions