Skip to content

Instantly share code, notes, and snippets.

@luisdelatorre012
Last active June 18, 2024 11:55
Show Gist options
  • Save luisdelatorre012/4a83e54a090094c6795d93cefb5472df to your computer and use it in GitHub Desktop.
Save luisdelatorre012/4a83e54a090094c6795d93cefb5472df to your computer and use it in GitHub Desktop.
Job scheduler with no overlap
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
from ortools.sat.python import cp_model
# Initialize the random generator
rng = np.random.default_rng(seed=42) # For reproducibility
# Generate random data for 10 jobs
num_jobs = 10
data = {
'job_id': range(1, num_jobs + 1),
'duration': rng.integers(1, 200, size=num_jobs), # Random durations between 1 and 200 minutes
'earliest_start': rng.integers(0, 1200, size=num_jobs) # Random earliest start times between 0 and 1200 minutes
}
df = pd.DataFrame(data)
# Display the generated dataframe
print("Generated Jobs:")
print(df)
# Define the model
model = cp_model.CpModel()
# Define variables
start_times = []
end_times = []
intervals = []
for i in range(len(df)):
start_var = model.NewIntVar(df.loc[i, 'earliest_start'], 1439 - df.loc[i, 'duration'], f'start_{i}')
end_var = model.NewIntVar(df.loc[i, 'earliest_start'] + df.loc[i, 'duration'], 1439, f'end_{i}')
interval_var = model.NewIntervalVar(start_var, df.loc[i, 'duration'], end_var, f'interval_{i}')
start_times.append(start_var)
end_times.append(end_var)
intervals.append(interval_var)
# Add cumulative constraint to allow at most 2 overlapping jobs
model.AddCumulative(intervals, [1] * len(intervals), 2)
# Define the objective: minimize total difference between assigned start time and earliest start time
objective_terms = []
for i in range(len(df)):
diff_pos = model.NewIntVar(0, 1439, f'diff_pos_{i}')
diff_neg = model.NewIntVar(0, 1439, f'diff_neg_{i}')
model.Add(diff_pos >= start_times[i] - df.loc[i, 'earliest_start'])
model.Add(diff_neg >= df.loc[i, 'earliest_start'] - start_times[i])
objective_terms.append(diff_pos)
objective_terms.append(diff_neg)
model.Minimize(sum(objective_terms))
# Solve the model
solver = cp_model.CpSolver()
status = solver.Solve(model)
# Check the solver status
if status == cp_model.OPTIMAL or status == cp_model.FEASIBLE:
print('Solution found!')
assigned_start_times = [solver.Value(start_times[i]) for i in range(len(df))]
df['assigned_start'] = assigned_start_times
print(df)
# Generate Gantt chart
fig, ax = plt.subplots(figsize=(10, 6))
for i in range(len(df)):
job_id = df.loc[i, 'job_id']
start = df.loc[i, 'assigned_start']
duration = df.loc[i, 'duration']
ax.broken_barh([(start, duration)], (i - 0.4, 0.8), facecolors=('tab:blue'))
# Set the x-axis to represent hours of the day
ax.set_xlim(0, 1440)
ax.set_xticks(np.arange(0, 1441, 60))
ax.set_xticklabels([f'{int(x/60):02d}:00' for x in np.arange(0, 1441, 60)])
# Set y-axis
ax.set_yticks(range(len(df)))
ax.set_yticklabels(df['job_id'])
ax.set_xlabel('Time of Day')
ax.set_ylabel('Job ID')
ax.set_title('Job Schedule')
plt.show()
else:
print('No solution found. Solver status:', status)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment