Spaces:
Sleeping
Sleeping
| # objective_engine.py | |
| """ | |
| This module implements the ObjectiveEngine for the VTU Automated Timetable Generator. | |
| It handles ONLY SOFT CONSTRAINTS by adding weighted penalties to the solver's objective function. | |
| Responsibilities: | |
| - Define penalties for undesirable schedules (e.g., gaps, late core classes). | |
| - Create auxiliary variables to calculate complex metrics (like daily span). | |
| - Sum all weighted penalties and set the Minimization objective. | |
| This module is optional and pluggable. It does not enforce hard rules. | |
| """ | |
| from typing import List, Dict | |
| from ortools.sat.python import cp_model | |
| from collections import defaultdict | |
| # Import project-specific modules | |
| from models import Task, Faculty, Section, Room, SubjectType | |
| import constants as const | |
| # Type hint for the ConstraintEngine | |
| from constraint_engine import ConstraintEngine | |
| class ObjectiveEngine: | |
| """ | |
| Manages soft constraints and the optimization objective. | |
| """ | |
| def __init__( | |
| self, | |
| model: cp_model.CpModel, | |
| constraint_engine: ConstraintEngine, | |
| tasks: List[Task], | |
| faculties: List[Faculty], | |
| sections: List[Section], | |
| rooms: List[Room], | |
| weights: Dict[str, int] = None | |
| ): | |
| """ | |
| Initializes the ObjectiveEngine. | |
| """ | |
| self.model = model | |
| self.ce = constraint_engine | |
| self.tasks = tasks | |
| self.faculties = faculties | |
| self.sections = sections | |
| self.rooms = rooms | |
| # Default weights if none provided | |
| self.weights = weights or { | |
| "subject_repetition": 10, | |
| "morning_core": 5, | |
| "late_heavy": 5, | |
| "faculty_gaps": 2, | |
| "campus_movement": 3, | |
| "faculty_load_balance": 1, | |
| "no_first_hour_free": 20 | |
| } | |
| self.penalties: List[cp_model.IntVar] = [] | |
| 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_campus_movement() | |
| self._penalize_first_hour_free() | |
| # 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): | |
| """ | |
| Penalizes scheduling the same theory subject multiple times on the same day | |
| for a specific section. | |
| """ | |
| weight = self.weights.get("subject_repetition", 0) | |
| if weight == 0: return | |
| # Group tasks by (section, subject) | |
| 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 | |
| # Compare every pair | |
| for i in range(len(subject_tasks)): | |
| for j in range(i + 1, len(subject_tasks)): | |
| t1 = subject_tasks[i] | |
| t2 = subject_tasks[j] | |
| start_var_1 = self.ce.task_vars[t1.task_id][0] | |
| start_var_2 = self.ce.task_vars[t2.task_id][0] | |
| # Create variables representing the day index (0-4) | |
| day_1 = self.model.NewIntVar(0, const.NUM_WORKING_DAYS - 1, f"day_{t1.task_id}") | |
| day_2 = self.model.NewIntVar(0, const.NUM_WORKING_DAYS - 1, f"day_{t2.task_id}") | |
| # Helper: day = start_slot // slots_per_day | |
| self.model.AddDivisionEquality(day_1, start_var_1, const.NUM_TEACHING_SLOTS_PER_DAY) | |
| self.model.AddDivisionEquality(day_2, start_var_2, const.NUM_TEACHING_SLOTS_PER_DAY) | |
| # Reify: are they on the same day? | |
| 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()) | |
| # Add penalty | |
| self.penalties.append(same_day * weight) | |
| def _prioritize_morning_core_subjects(self): | |
| """ | |
| Penalizes Core subjects if they are scheduled after the lunch break. | |
| """ | |
| weight = self.weights.get("morning_core", 0) | |
| if weight == 0: return | |
| # Assume slots 0-3 are morning, 4-7 are afternoon | |
| afternoon_start_index = 4 | |
| for task in self.tasks: | |
| if task.subject.is_core and task.subject.subject_type == SubjectType.THEORY: | |
| start_var = self.ce.task_vars[task.task_id][0] | |
| daily_slot = self.model.NewIntVar(0, const.NUM_TEACHING_SLOTS_PER_DAY - 1, f"daily_slot_{task.task_id}") | |
| self.model.AddModuloEquality(daily_slot, start_var, const.NUM_TEACHING_SLOTS_PER_DAY) | |
| # Penalty if daily_slot >= afternoon_start_index | |
| 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): | |
| """ | |
| Penalizes Heavy subjects if they are scheduled in the very last slot of the day. | |
| """ | |
| 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: | |
| start_var = self.ce.task_vars[task.task_id][0] | |
| daily_slot = self.model.NewIntVar(0, const.NUM_TEACHING_SLOTS_PER_DAY - 1, f"daily_slot_heavy_{task.task_id}") | |
| self.model.AddModuloEquality(daily_slot, start_var, const.NUM_TEACHING_SLOTS_PER_DAY) | |
| 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): | |
| """ | |
| Penalizes 'idle spans' for faculty. | |
| We approximate this by minimizing (Daily End Time - Daily Start Time - Total Teaching Duration). | |
| """ | |
| weight = self.weights.get("faculty_gaps", 0) | |
| if weight == 0: return | |
| # Group tasks by faculty | |
| tasks_by_faculty = {f.id: [] for f in self.faculties} | |
| for task in self.tasks: | |
| tasks_by_faculty[task.faculty.id].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): | |
| day_offset_start = day * const.NUM_TEACHING_SLOTS_PER_DAY | |
| day_offset_end = (day + 1) * const.NUM_TEACHING_SLOTS_PER_DAY | |
| # Variables to track if faculty is active on this day, and their start/end | |
| day_active = self.model.NewBoolVar(f"active_{faculty_id}_{day}") | |
| day_start = self.model.NewIntVar(day_offset_start, day_offset_end, f"start_{faculty_id}_{day}") | |
| day_end = self.model.NewIntVar(day_offset_start, day_offset_end, f"end_{faculty_id}_{day}") | |
| task_on_day_lits = [] | |
| total_duration_on_day = self.model.NewIntVar(0, const.NUM_TEACHING_SLOTS_PER_DAY, f"dur_{faculty_id}_{day}") | |
| durations_sum = [] | |
| for task in f_tasks: | |
| t_start = self.ce.task_vars[task.task_id][0] | |
| t_end = self.ce.task_vars[task.task_id][1] | |
| is_on_day = self.model.NewBoolVar(f"{task.task_id}_on_day_{day}") | |
| t_day = self.model.NewIntVar(0, const.NUM_WORKING_DAYS - 1, f"t_day_{task.task_id}_{day}") | |
| self.model.AddDivisionEquality(t_day, t_start, const.NUM_TEACHING_SLOTS_PER_DAY) | |
| self.model.Add(t_day == day).OnlyEnforceIf(is_on_day) | |
| self.model.Add(t_day != day).OnlyEnforceIf(is_on_day.Not()) | |
| task_on_day_lits.append(is_on_day) | |
| # Update min start and max end for the day ONLY if task is on this day | |
| self.model.Add(day_start <= t_start).OnlyEnforceIf(is_on_day) | |
| self.model.Add(day_end >= t_end).OnlyEnforceIf(is_on_day) | |
| # Accumulate duration | |
| dur_term = self.model.NewIntVar(0, task.duration, f"dur_term_{task.task_id}_{day}") | |
| self.model.Add(dur_term == task.duration).OnlyEnforceIf(is_on_day) | |
| self.model.Add(dur_term == 0).OnlyEnforceIf(is_on_day.Not()) | |
| durations_sum.append(dur_term) | |
| # If no tasks on this day, force active to false | |
| self.model.Add(sum(task_on_day_lits) > 0).OnlyEnforceIf(day_active) | |
| self.model.Add(sum(task_on_day_lits) == 0).OnlyEnforceIf(day_active.Not()) | |
| # --- FIX: Use Python sum() inside Add() instead of self.model.Sum() --- | |
| self.model.Add(total_duration_on_day == sum(durations_sum)) | |
| span = self.model.NewIntVar(0, const.NUM_TEACHING_SLOTS_PER_DAY, f"span_{faculty_id}_{day}") | |
| self.model.Add(span == day_end - day_start).OnlyEnforceIf(day_active) | |
| self.model.Add(span == 0).OnlyEnforceIf(day_active.Not()) | |
| idle_time = self.model.NewIntVar(0, const.NUM_TEACHING_SLOTS_PER_DAY, f"idle_{faculty_id}_{day}") | |
| self.model.Add(idle_time == span - total_duration_on_day).OnlyEnforceIf(day_active) | |
| self.model.Add(idle_time == 0).OnlyEnforceIf(day_active.Not()) | |
| self.penalties.append(idle_time * weight) | |
| def _minimize_campus_movement(self): | |
| """ | |
| Penalizes consecutive tasks for a section that are in different buildings. | |
| """ | |
| weight = self.weights.get("campus_movement", 0) | |
| if weight == 0: return | |
| unique_buildings = sorted(list(set(r.building for r in self.rooms))) | |
| building_to_id = {b: i for i, b in enumerate(unique_buildings)} | |
| room_idx_to_building_id = [building_to_id[r.building] for r in self.rooms] | |
| tasks_by_section = defaultdict(list) | |
| for task in self.tasks: | |
| tasks_by_section[task.section.section_id].append(task) | |
| for sec_id, sec_tasks in tasks_by_section.items(): | |
| if len(sec_tasks) < 2: continue | |
| for i in range(len(sec_tasks)): | |
| for j in range(len(sec_tasks)): | |
| if i == j: continue | |
| t1 = sec_tasks[i] | |
| t2 = sec_tasks[j] | |
| t1_end = self.ce.task_vars[t1.task_id][1] | |
| t2_start = self.ce.task_vars[t2.task_id][0] | |
| is_consecutive = self.model.NewBoolVar(f"consec_{t1.task_id}_{t2.task_id}") | |
| self.model.Add(t1_end == t2_start).OnlyEnforceIf(is_consecutive) | |
| self.model.Add(t1_end != t2_start).OnlyEnforceIf(is_consecutive.Not()) | |
| b1_var = self.model.NewIntVar(0, len(unique_buildings), f"bld_{t1.task_id}") | |
| b2_var = self.model.NewIntVar(0, len(unique_buildings), f"bld_{t2.task_id}") | |
| room_var_1 = self.ce.task_vars[t1.task_id][3] | |
| room_var_2 = self.ce.task_vars[t2.task_id][3] | |
| self.model.AddElement(room_var_1, room_idx_to_building_id, b1_var) | |
| self.model.AddElement(room_var_2, room_idx_to_building_id, b2_var) | |
| diff_building = self.model.NewBoolVar(f"diff_bld_{t1.task_id}_{t2.task_id}") | |
| self.model.Add(b1_var != b2_var).OnlyEnforceIf(diff_building) | |
| self.model.Add(b1_var == b2_var).OnlyEnforceIf(diff_building.Not()) | |
| penalty_active = self.model.NewBoolVar(f"move_pen_{t1.task_id}_{t2.task_id}") | |
| self.model.AddBoolAnd([is_consecutive, diff_building]).OnlyEnforceIf(penalty_active) | |
| self.penalties.append(penalty_active * weight) | |
| def _penalize_first_hour_free(self): | |
| """ | |
| Strongly penalizes having the first hour (period index 0) free for any | |
| section on any day. The solver will avoid this unless there is genuinely | |
| no other feasible assignment. | |
| Since tasks never cross day boundaries, a task covers period 0 of a day | |
| if and only if its start_var equals that day's first absolute slot index. | |
| """ | |
| weight = self.weights.get("no_first_hour_free", 20) | |
| if weight == 0: | |
| return | |
| # Group tasks by section | |
| tasks_by_section = defaultdict(list) | |
| for task in self.tasks: | |
| tasks_by_section[task.section.section_id].append(task) | |
| for sec_id, sec_tasks in tasks_by_section.items(): | |
| for day in range(const.NUM_WORKING_DAYS): | |
| # The absolute slot index for period 0 of this day | |
| first_slot = day * const.NUM_TEACHING_SLOTS_PER_DAY | |
| # For each task, create a bool: does it start at exactly first_slot? | |
| starts_at_first = [] | |
| for task in sec_tasks: | |
| start_var = self.ce.task_vars[task.task_id][0] | |
| at_first = self.model.NewBoolVar(f"at1st_{task.task_id}_d{day}") | |
| self.model.Add(start_var == first_slot).OnlyEnforceIf(at_first) | |
| self.model.Add(start_var != first_slot).OnlyEnforceIf(at_first.Not()) | |
| starts_at_first.append(at_first) | |
| # any_at_first = True if at least one task starts at period 0 | |
| any_at_first = self.model.NewBoolVar(f"any_at1st_{sec_id}_d{day}") | |
| self.model.AddBoolOr(starts_at_first).OnlyEnforceIf(any_at_first) | |
| for lit in starts_at_first: | |
| self.model.AddImplication(any_at_first.Not(), lit.Not()) | |
| # Penalty when the first hour IS free (no task at period 0) | |
| first_free = self.model.NewBoolVar(f"first_free_{sec_id}_d{day}") | |
| self.model.Add(first_free == 1).OnlyEnforceIf(any_at_first.Not()) | |
| self.model.Add(first_free == 0).OnlyEnforceIf(any_at_first) | |
| self.penalties.append(first_free * weight) |