diff --git a/DIRECTORY.md b/DIRECTORY.md index cdbbac684fd2..dfb87d324a28 100644 --- a/DIRECTORY.md +++ b/DIRECTORY.md @@ -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) diff --git a/scheduling/weighted_interval_scheduling.py b/scheduling/weighted_interval_scheduling.py new file mode 100644 index 000000000000..c757d6705ce8 --- /dev/null +++ b/scheduling/weighted_interval_scheduling.py @@ -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}")