Spaces:
Sleeping
Sleeping
| # objective_engine.py | |
| from collections import defaultdict | |
| from typing import List, Dict, Any | |
| from ortools.sat.python import cp_model | |
| from models import Task, Faculty, Section, SubjectType | |
| from constraint_engine import ConstraintEngine | |
| import constants as const | |
| class ObjectiveEngine: | |
| def __init__( | |
| self, | |
| model: cp_model.CpModel, | |
| ce: ConstraintEngine, | |
| tasks: List[Task], | |
| faculties: List[Faculty], | |
| sections: List[Section], | |
| weights: Dict[str, int] = None | |
| ): | |
| self.model = model | |
| self.ce = ce | |
| self.tasks = tasks | |
| self.faculties = faculties | |
| self.sections = sections | |
| self.weights = weights or { | |
| "subject_repetition": 100, | |
| "morning_core": 50, | |
| "late_heavy": 50, | |
| "faculty_gaps": 5000, | |
| "student_gaps": 10000, | |
| "isolated_afternoon": 1000, | |
| "campus_movement": 30, | |
| "faculty_load_balance": 10, | |
| "no_first_hour_free": 5000, | |
| "pack_morning": 200, | |
| "avoid_late_afternoon": 200 | |
| } | |
| self.penalties: List[cp_model.IntVar] = [] | |
| # Precompute common variables for extreme efficiency | |
| self.t_day = {} | |
| self.t_dstart = {} | |
| self.t_dend = {} | |
| for task in self.tasks: | |
| start_var = self.ce.task_vars[task.task_id][0] | |
| day = self.model.NewIntVar(0, const.NUM_WORKING_DAYS - 1, f"t_{task.task_id}_day") | |
| self.model.AddDivisionEquality(day, start_var, const.NUM_TEACHING_SLOTS_PER_DAY) | |
| self.t_day[task.task_id] = day | |
| d_start = self.model.NewIntVar(0, const.NUM_TEACHING_SLOTS_PER_DAY - 1, f"t_{task.task_id}_dstart") | |
| self.model.AddModuloEquality(d_start, start_var, const.NUM_TEACHING_SLOTS_PER_DAY) | |
| self.t_dstart[task.task_id] = d_start | |
| # Note: End slot could reach NUM_TEACHING_SLOTS_PER_DAY (which means it ends at the very end of the day) | |
| d_end = self.model.NewIntVar(0, const.NUM_TEACHING_SLOTS_PER_DAY, f"t_{task.task_id}_dend") | |
| self.model.Add(d_end == d_start + task.duration) | |
| self.t_dend[task.task_id] = d_end | |
| def build_objective(self): | |
| """ | |
| Applies all configured soft constraints and sets the minimization objective. | |
| """ | |
| print("Building Objective Function...") | |
| self._minimize_subject_repetition() | |
| self._prioritize_morning_core_subjects() | |
| self._avoid_late_heavy_subjects() | |
| self._minimize_faculty_gaps() | |
| self._minimize_student_gaps() | |
| self._penalize_isolated_afternoon_classes() | |
| self._minimize_campus_movement() | |
| self._penalize_first_hour_free() | |
| self._penalize_empty_morning_slots() | |
| self._penalize_late_afternoon_slots() | |
| # Summation of all penalties | |
| if self.penalties: | |
| total_cost = sum(self.penalties) | |
| self.model.Minimize(total_cost) | |
| else: | |
| self.model.Minimize(0) | |
| def _minimize_subject_repetition(self): | |
| weight = self.weights.get("subject_repetition", 0) | |
| if weight == 0: return | |
| tasks_by_sec_sub = {} | |
| for task in self.tasks: | |
| if task.subject.subject_type == SubjectType.THEORY: | |
| key = (task.section.section_id, task.subject.subject_code) | |
| if key not in tasks_by_sec_sub: | |
| tasks_by_sec_sub[key] = [] | |
| tasks_by_sec_sub[key].append(task) | |
| for (sec_id, sub_code), subject_tasks in tasks_by_sec_sub.items(): | |
| if len(subject_tasks) < 2: continue | |
| for i in range(len(subject_tasks)): | |
| for j in range(i + 1, len(subject_tasks)): | |
| t1 = subject_tasks[i] | |
| t2 = subject_tasks[j] | |
| day_1 = self.t_day[t1.task_id] | |
| day_2 = self.t_day[t2.task_id] | |
| same_day = self.model.NewBoolVar(f"same_day_{t1.task_id}_{t2.task_id}") | |
| self.model.Add(day_1 == day_2).OnlyEnforceIf(same_day) | |
| self.model.Add(day_1 != day_2).OnlyEnforceIf(same_day.Not()) | |
| self.penalties.append(same_day * weight) | |
| def _prioritize_morning_core_subjects(self): | |
| weight = self.weights.get("morning_core", 0) | |
| if weight == 0: return | |
| afternoon_start_index = 4 | |
| for task in self.tasks: | |
| if task.subject.is_core and task.subject.subject_type == SubjectType.THEORY: | |
| daily_slot = self.t_dstart[task.task_id] | |
| is_afternoon = self.model.NewBoolVar(f"is_afternoon_{task.task_id}") | |
| self.model.Add(daily_slot >= afternoon_start_index).OnlyEnforceIf(is_afternoon) | |
| self.model.Add(daily_slot < afternoon_start_index).OnlyEnforceIf(is_afternoon.Not()) | |
| self.penalties.append(is_afternoon * weight) | |
| def _avoid_late_heavy_subjects(self): | |
| weight = self.weights.get("late_heavy", 0) | |
| if weight == 0: return | |
| last_slot_index = const.NUM_TEACHING_SLOTS_PER_DAY - 1 | |
| for task in self.tasks: | |
| if task.subject.is_heavy: | |
| daily_slot = self.t_dstart[task.task_id] | |
| is_last_slot = self.model.NewBoolVar(f"is_last_slot_{task.task_id}") | |
| self.model.Add(daily_slot == last_slot_index).OnlyEnforceIf(is_last_slot) | |
| self.model.Add(daily_slot != last_slot_index).OnlyEnforceIf(is_last_slot.Not()) | |
| self.penalties.append(is_last_slot * weight) | |
| def _minimize_faculty_gaps(self): | |
| weight = self.weights.get("faculty_gaps", 30) | |
| if weight == 0: return | |
| tasks_by_faculty = {f.id: [] for f in self.faculties} | |
| faculty_ids_set = {f.id for f in self.faculties} | |
| for task in self.tasks: | |
| parts = task.faculty.id.split('_') | |
| fids = parts if len(parts) > 1 and all(p in faculty_ids_set for p in parts) else [task.faculty.id] | |
| for fid in fids: | |
| if fid in tasks_by_faculty: | |
| tasks_by_faculty[fid].append(task) | |
| for faculty_id, f_tasks in tasks_by_faculty.items(): | |
| if not f_tasks: continue | |
| for day in range(const.NUM_WORKING_DAYS): | |
| min_start_on_day = self.model.NewIntVar(0, const.NUM_TEACHING_SLOTS_PER_DAY, f"fac_{faculty_id}_min_s_d{day}") | |
| max_end_on_day = self.model.NewIntVar(0, const.NUM_TEACHING_SLOTS_PER_DAY, f"fac_{faculty_id}_max_e_d{day}") | |
| start_vars = [const.NUM_TEACHING_SLOTS_PER_DAY] | |
| end_vars = [0] | |
| total_duration = 0 | |
| for task in f_tasks: | |
| is_on_day = self.model.NewBoolVar(f"fac_{faculty_id}_on_d{day}_t{task.task_id}") | |
| t_day = self.t_day[task.task_id] | |
| self.model.Add(t_day == day).OnlyEnforceIf(is_on_day) | |
| self.model.Add(t_day != day).OnlyEnforceIf(is_on_day.Not()) | |
| start_on_day = self.model.NewIntVar(0, const.NUM_TEACHING_SLOTS_PER_DAY, "") | |
| self.model.Add(start_on_day == self.t_dstart[task.task_id]).OnlyEnforceIf(is_on_day) | |
| self.model.Add(start_on_day == const.NUM_TEACHING_SLOTS_PER_DAY).OnlyEnforceIf(is_on_day.Not()) | |
| start_vars.append(start_on_day) | |
| end_on_day = self.model.NewIntVar(0, const.NUM_TEACHING_SLOTS_PER_DAY, "") | |
| self.model.Add(end_on_day == self.t_dend[task.task_id]).OnlyEnforceIf(is_on_day) | |
| self.model.Add(end_on_day == 0).OnlyEnforceIf(is_on_day.Not()) | |
| end_vars.append(end_on_day) | |
| total_duration += task.duration * is_on_day | |
| self.model.AddMinEquality(min_start_on_day, start_vars) | |
| self.model.AddMaxEquality(max_end_on_day, end_vars) | |
| gap = self.model.NewIntVar(0, const.NUM_TEACHING_SLOTS_PER_DAY, f"fac_{faculty_id}_gap_d{day}") | |
| self.model.AddMaxEquality(gap, [0, max_end_on_day - min_start_on_day - total_duration]) | |
| self.penalties.append(gap * weight) | |
| def _minimize_student_gaps(self): | |
| weight = self.weights.get("student_gaps", 100) | |
| if weight == 0: return | |
| tasks_by_section = defaultdict(list) | |
| for task in self.tasks: | |
| tasks_by_section[task.section.section_id].append(task) | |
| for sec_id, s_tasks in tasks_by_section.items(): | |
| if not s_tasks: continue | |
| for day in range(const.NUM_WORKING_DAYS): | |
| min_start_on_day = self.model.NewIntVar(0, const.NUM_TEACHING_SLOTS_PER_DAY, f"sec_{sec_id}_min_s_d{day}") | |
| max_end_on_day = self.model.NewIntVar(0, const.NUM_TEACHING_SLOTS_PER_DAY, f"sec_{sec_id}_max_e_d{day}") | |
| start_vars = [const.NUM_TEACHING_SLOTS_PER_DAY] | |
| end_vars = [0] | |
| total_duration = 0 | |
| for task in s_tasks: | |
| is_on_day = self.model.NewBoolVar(f"sec_{sec_id}_on_d{day}_t{task.task_id}") | |
| t_day = self.t_day[task.task_id] | |
| self.model.Add(t_day == day).OnlyEnforceIf(is_on_day) | |
| self.model.Add(t_day != day).OnlyEnforceIf(is_on_day.Not()) | |
| start_on_day = self.model.NewIntVar(0, const.NUM_TEACHING_SLOTS_PER_DAY, "") | |
| self.model.Add(start_on_day == self.t_dstart[task.task_id]).OnlyEnforceIf(is_on_day) | |
| self.model.Add(start_on_day == const.NUM_TEACHING_SLOTS_PER_DAY).OnlyEnforceIf(is_on_day.Not()) | |
| start_vars.append(start_on_day) | |
| end_on_day = self.model.NewIntVar(0, const.NUM_TEACHING_SLOTS_PER_DAY, "") | |
| self.model.Add(end_on_day == self.t_dend[task.task_id]).OnlyEnforceIf(is_on_day) | |
| self.model.Add(end_on_day == 0).OnlyEnforceIf(is_on_day.Not()) | |
| end_vars.append(end_on_day) | |
| total_duration += task.duration * is_on_day | |
| self.model.AddMinEquality(min_start_on_day, start_vars) | |
| self.model.AddMaxEquality(max_end_on_day, end_vars) | |
| late_start_weight = self.weights.get("no_first_hour_free", 5000) | |
| if weight > 0: | |
| gap = self.model.NewIntVar(0, const.NUM_TEACHING_SLOTS_PER_DAY, f"sec_{sec_id}_gap_d{day}") | |
| self.model.AddMaxEquality(gap, [0, max_end_on_day - min_start_on_day - total_duration]) | |
| self.penalties.append(gap * weight) | |
| if late_start_weight > 0: | |
| is_free_day = self.model.NewBoolVar(f"sec_{sec_id}_free_d{day}") | |
| self.model.Add(total_duration == 0).OnlyEnforceIf(is_free_day) | |
| self.model.Add(total_duration > 0).OnlyEnforceIf(is_free_day.Not()) | |
| late_start_penalty = self.model.NewIntVar(0, const.NUM_TEACHING_SLOTS_PER_DAY, f"sec_{sec_id}_late_d{day}") | |
| self.model.Add(late_start_penalty == 0).OnlyEnforceIf(is_free_day) | |
| self.model.Add(late_start_penalty == min_start_on_day).OnlyEnforceIf(is_free_day.Not()) | |
| self.penalties.append(late_start_penalty * late_start_weight) | |
| def _penalize_isolated_afternoon_classes(self): | |
| weight = self.weights.get("isolated_afternoon", 10) | |
| if weight == 0: return | |
| afternoon_start_index = 4 | |
| tasks_by_section = defaultdict(list) | |
| for task in self.tasks: | |
| tasks_by_section[task.section.section_id].append(task) | |
| for sec_id, s_tasks in tasks_by_section.items(): | |
| for day in range(const.NUM_WORKING_DAYS): | |
| afternoon_duration_sum = [] | |
| for task in s_tasks: | |
| daily_slot = self.t_dstart[task.task_id] | |
| t_day = self.t_day[task.task_id] | |
| is_on_day = self.model.NewBoolVar(f"is_on_day_{task.task_id}_{day}_aft") | |
| self.model.Add(t_day == day).OnlyEnforceIf(is_on_day) | |
| self.model.Add(t_day != day).OnlyEnforceIf(is_on_day.Not()) | |
| is_afternoon = self.model.NewBoolVar(f"is_afternoon_{task.task_id}_{day}_aft") | |
| self.model.Add(daily_slot >= afternoon_start_index).OnlyEnforceIf(is_afternoon) | |
| self.model.Add(daily_slot < afternoon_start_index).OnlyEnforceIf(is_afternoon.Not()) | |
| is_on_day_and_afternoon = self.model.NewBoolVar(f"is_on_day_and_afternoon_{task.task_id}_{day}") | |
| self.model.AddBoolAnd([is_on_day, is_afternoon]).OnlyEnforceIf(is_on_day_and_afternoon) | |
| dur_term = self.model.NewIntVar(0, task.duration, f"aft_dur_{task.task_id}_{day}") | |
| self.model.Add(dur_term == task.duration).OnlyEnforceIf(is_on_day_and_afternoon) | |
| self.model.Add(dur_term == 0).OnlyEnforceIf(is_on_day_and_afternoon.Not()) | |
| afternoon_duration_sum.append(dur_term) | |
| total_aft_duration = self.model.NewIntVar(0, const.NUM_TEACHING_SLOTS_PER_DAY, f"total_aft_dur_sec{sec_id}_d{day}") | |
| self.model.Add(total_aft_duration == sum(afternoon_duration_sum)) | |
| is_isolated = self.model.NewBoolVar(f"is_isolated_aft_sec{sec_id}_d{day}") | |
| is_gt_0 = self.model.NewBoolVar(f"aft_gt_0_sec{sec_id}_d{day}") | |
| self.model.Add(total_aft_duration > 0).OnlyEnforceIf(is_gt_0) | |
| self.model.Add(total_aft_duration == 0).OnlyEnforceIf(is_gt_0.Not()) | |
| is_le_2 = self.model.NewBoolVar(f"aft_le_2_sec{sec_id}_d{day}") | |
| self.model.Add(total_aft_duration <= 2).OnlyEnforceIf(is_le_2) | |
| self.model.Add(total_aft_duration > 2).OnlyEnforceIf(is_le_2.Not()) | |
| self.model.AddBoolAnd([is_gt_0, is_le_2]).OnlyEnforceIf(is_isolated) | |
| self.penalties.append(is_isolated * weight) | |
| def _minimize_campus_movement(self): | |
| weight = self.weights.get("campus_movement", 3) | |
| if weight == 0: return | |
| tasks_by_section = defaultdict(list) | |
| for task in self.tasks: | |
| tasks_by_section[task.section.section_id].append(task) | |
| for sec_id, s_tasks in tasks_by_section.items(): | |
| if len(s_tasks) < 2: continue | |
| for i in range(len(s_tasks)): | |
| for j in range(i + 1, len(s_tasks)): | |
| t1 = s_tasks[i] | |
| t2 = s_tasks[j] | |
| start1 = self.ce.task_vars[t1.task_id][0] | |
| end1 = self.ce.task_vars[t1.task_id][1] | |
| room1 = self.ce.task_vars[t1.task_id][3] | |
| start2 = self.ce.task_vars[t2.task_id][0] | |
| end2 = self.ce.task_vars[t2.task_id][1] | |
| room2 = self.ce.task_vars[t2.task_id][3] | |
| is_consecutive_1_2 = self.model.NewBoolVar(f"cons_{t1.task_id}_{t2.task_id}") | |
| self.model.Add(end1 == start2).OnlyEnforceIf(is_consecutive_1_2) | |
| self.model.Add(end1 != start2).OnlyEnforceIf(is_consecutive_1_2.Not()) | |
| is_consecutive_2_1 = self.model.NewBoolVar(f"cons_{t2.task_id}_{t1.task_id}") | |
| self.model.Add(end2 == start1).OnlyEnforceIf(is_consecutive_2_1) | |
| self.model.Add(end2 != start1).OnlyEnforceIf(is_consecutive_2_1.Not()) | |
| are_consecutive = self.model.NewBoolVar(f"are_cons_{t1.task_id}_{t2.task_id}") | |
| self.model.AddBoolOr([is_consecutive_1_2, is_consecutive_2_1]).OnlyEnforceIf(are_consecutive) | |
| self.model.AddBoolAnd([is_consecutive_1_2.Not(), is_consecutive_2_1.Not()]).OnlyEnforceIf(are_consecutive.Not()) | |
| same_room = self.model.NewBoolVar(f"same_room_{t1.task_id}_{t2.task_id}") | |
| self.model.Add(room1 == room2).OnlyEnforceIf(same_room) | |
| self.model.Add(room1 != room2).OnlyEnforceIf(same_room.Not()) | |
| move_penalty = self.model.NewBoolVar(f"move_{t1.task_id}_{t2.task_id}") | |
| self.model.AddBoolAnd([are_consecutive, same_room.Not()]).OnlyEnforceIf(move_penalty) | |
| self.penalties.append(move_penalty * weight) | |
| def _penalize_first_hour_free(self): | |
| pass | |
| def _penalize_empty_morning_slots(self): | |
| weight = self.weights.get("pack_morning", 50) | |
| if weight == 0: return | |
| tasks_by_section = defaultdict(list) | |
| for task in self.tasks: | |
| tasks_by_section[task.section.section_id].append(task) | |
| for sec_id, s_tasks in tasks_by_section.items(): | |
| if not s_tasks: continue | |
| morning_overlaps = [] | |
| for task in s_tasks: | |
| d_start = self.t_dstart[task.task_id] | |
| d_end = self.t_dend[task.task_id] | |
| capped_start = self.model.NewIntVar(0, 4, "") | |
| self.model.AddMinEquality(capped_start, [d_start, 4]) | |
| capped_end = self.model.NewIntVar(0, 4, "") | |
| self.model.AddMinEquality(capped_end, [d_end, 4]) | |
| overlap = self.model.NewIntVar(0, 4, "") | |
| self.model.Add(overlap == capped_end - capped_start) | |
| morning_overlaps.append(overlap) | |
| # Total morning slots are 20 (5 days * 4 slots) | |
| total_slots = const.NUM_WORKING_DAYS * 4 | |
| empty_slots = self.model.NewIntVar(0, total_slots, f"sec_{sec_id}_empty_morning") | |
| self.model.Add(empty_slots == total_slots - sum(morning_overlaps)) | |
| self.penalties.append(empty_slots * weight) | |
| def _penalize_late_afternoon_slots(self): | |
| weight = self.weights.get("avoid_late_afternoon", 40) | |
| if weight == 0: return | |
| for task in self.tasks: | |
| d_start = self.t_dstart[task.task_id] | |
| d_end = self.t_dend[task.task_id] | |
| capped_start = self.model.NewIntVar(6, const.NUM_TEACHING_SLOTS_PER_DAY, "") | |
| self.model.AddMaxEquality(capped_start, [d_start, 6]) | |
| capped_end = self.model.NewIntVar(6, const.NUM_TEACHING_SLOTS_PER_DAY, "") | |
| self.model.AddMaxEquality(capped_end, [d_end, 6]) | |
| overlap = self.model.NewIntVar(0, const.NUM_TEACHING_SLOTS_PER_DAY - 6, "") | |
| self.model.Add(overlap == capped_end - capped_start) | |
| self.penalties.append(overlap * weight) | |