# 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)