Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

feat: Added Weighted Interval Scheduling #11765

Open
wants to merge 6 commits into
base: master
Choose a base branch
from
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions DIRECTORY.md
Original file line number Diff line number Diff line change
Expand Up @@ -1195,6 +1195,7 @@
* [Non Preemptive Shortest Job First](scheduling/non_preemptive_shortest_job_first.py)
* [Round Robin](scheduling/round_robin.py)
* [Shortest Job First](scheduling/shortest_job_first.py)
* [Weighted Interval Scheduling](scheduling/weighted_interval_scheduling.py)

## Searches
* [Binary Search](searches/binary_search.py)
Expand Down
81 changes: 81 additions & 0 deletions scheduling/weighted_interval_scheduling.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,81 @@
# Implementation of Weighted Interval Scheduling algorithm
# In this algorithm, we are given a list of jobs with start and end times,
# and each job has a specific weight.
# The goal is to find the maximum weight subset of non-overlapping jobs.
# https://en.wikipedia.org/wiki/Interval_scheduling

from __future__ import annotations


def latest_non_conflict(jobs: list[tuple[int, int, int]], indx: int) -> int:
"""
This function finds the latest job that does not conflict with
the current job at index `n`.
The jobs are given as (start_time, end_time, weight), and the
jobs should be sorted by end time.
It returns the index of the latest job that finishes before the
current job starts.
Return: The index of the latest non-conflicting job.
>>> latest_non_conflict([(1, 3, 50), (2, 5, 20), (4, 6, 30)], 2)
0
>>> latest_non_conflict([(1, 3, 50), (3, 4, 60), (5, 9, 70)], 2)
1
"""
for j in range(indx - 1, -1, -1):
if jobs[j][1] <= jobs[indx][0]:
return j
return -1


def find_max_weight(jobs: list[tuple[int, int, int]]) -> int:
"""
This function calculates the maximum weight of non-overlapping jobs
using dynamic programming.
Each job is represented by a tuple (start_time, end_time, weight).
The function builds a DP table where each entry `dp[i]` represents
the maximum weight achievable
using jobs from index 0 to i.
Return: The maximum achievable weight without overlapping jobs.
>>> find_max_weight([(1, 3, 50), (2, 5, 20), (4, 6, 30)])
80
>>> find_max_weight([(1, 3, 10), (2, 5, 100), (6, 8, 15)])
115
>>> find_max_weight([(1, 3, 20), (3, 5, 30), (6, 19, 60), (2, 100, 200)])
200
"""
# Sort jobs based on their end times
jobs.sort(key=lambda ele: ele[1])

# Initialize dp array to store the maximum weight up to each job
length = len(jobs)
dp = [0] * length
dp[0] = jobs[0][2] # The weight of the first job is the initial value

for i in range(1, length):
# Include the current job
include_weight = jobs[i][2]
latest_job = latest_non_conflict(jobs, i)
if latest_job != -1:
include_weight += dp[latest_job]

# Exclude the current job, and take the maximum of including or
# excluding
dp[i] = max(include_weight, dp[i - 1])

return dp[-1] # The last entry contains the maximum weight


if __name__ == "__main__":
# Example list of jobs (start_time, end_time, weight)
jobs = [(1, 2, 50), (3, 5, 20), (6, 19, 100), (2, 100, 200)]

# Ensure we have jobs to process
if len(jobs) == 0:
print("No jobs available to process")
raise SystemExit(0)

# Calculate the maximum weight for non-overlapping jobs
max_weight = find_max_weight(jobs)

# Print the result
print(f"The maximum weight of non-overlapping jobs is {max_weight}")