Skip to content
Closed
Show file tree
Hide file tree
Changes from 3 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]], n: 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(n - 1, -1, -1):
if jobs[j][1] <= jobs[n][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 x: x[1])

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

for i in range(1, n):
# 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}")