Update partial_optimizer.py

partial_optimizer.py

@@ -71,61 +71,22 @@ class PartialOptimizer:
             self.tasks_by_subject[t.subject.subject_code].append(t)
 
     # ═════════════════════════════════════════════════════════════════
-    # TYPE CLASSIFICATION HELPERS
+    # PUBLIC ENTRY POINT
     # ═════════════════════════════════════════════════════════════════
 
-    # Direct mutation handlers (no CP-SAT needed)
-    DIRECT_TYPES = {
-        'FACULTY_SUBSTITUTION', 'MOVE_CLASS', 'SWAP_CLASSES',
-        'CANCEL_CLASS', 'CHANGE_ROOM', 'ADD_EXTRA_CLASS',
-        'MARK_HOLIDAY', 'RESCHEDULE_LAB', 'CHANGE_FACULTY',
-        'FREEZE_SLOT',
-    }
-
-    # Re-optimization handlers (CP-SAT partial solve)
-    REOPT_TYPES = {
-        'FACULTY_UNAVAILABLE', 'FACULTY_FREE_DAY',
-        'FACULTY_MAX_DAILY_HOURS', 'FACULTY_NO_CONSECUTIVE',
-        'SECTION_FREE_SLOT', 'WORKING_DAYS',
-        'SUBJECT_PREFERRED_TIME', 'HEAVY_SUBJECT_MORNING',
-        'LAB_MUST_CONSECUTIVE', 'NO_BACK_TO_BACK_SUBJECTS',
-        'DISTRIBUTE_SUBJECTS_EVENLY', 'SUBJECT_SPACING',
-        'NO_FREE_PERIOD',
-    }
-
-    # Type aliases — map SLM variants to canonical types
-    ALIASES = {
-        'SUBJECT_FREE_DAY':         'CANCEL_CLASS',
-        'FACULTY_LEAVE':            'FACULTY_SUBSTITUTION',
-        'BLOCK_SLOT':               'SECTION_FREE_SLOT',
-        'REMOVE_CLASS':             'CANCEL_CLASS',
-        'DELETE_CLASS':             'CANCEL_CLASS',
-        'CLASS_CANCELLED':          'CANCEL_CLASS',
-        'SHIFT_CLASS':              'MOVE_CLASS',
-        'RELOCATE_CLASS':           'MOVE_CLASS',
-        'TEACHER_SUBSTITUTION':     'FACULTY_SUBSTITUTION',
-        'REPLACE_FACULTY':          'FACULTY_SUBSTITUTION',
-        'SWAP_FACULTY':             'FACULTY_SUBSTITUTION',
-        'SUBJECT_UNAVAILABLE':      'CANCEL_CLASS',
-        'NO_CLASS':                 'CANCEL_CLASS',
-        'HOLIDAY':                  'MARK_HOLIDAY',
-        'MOVE_LAB':                 'RESCHEDULE_LAB',
-        'SHIFT_LAB':                'RESCHEDULE_LAB',
-        'LOCK_SLOT':                'FREEZE_SLOT',
-        'PIN_SLOT':                 'FREEZE_SLOT',
-    }
-
-    def _resolve_alias(self, constraint: Dict[str, Any]) -> Dict[str, Any]:
-        """Resolve type aliases to canonical constraint types."""
-        ctype = constraint.get('type', '').upper()
-        if ctype in self.ALIASES:
-            constraint = dict(constraint)
-            constraint['type'] = self.ALIASES[ctype]
-        return constraint
-
-    def _get_direct_handler(self, ctype: str):
-        """Return the direct-mutation handler function for a constraint type."""
-        handlers = {
+    def apply_constraint_and_reoptimize(
+        self,
+        slm_constraint: Dict[str, Any],
+        time_limit: int = 30
+    ) -> Tuple[str, Dict[str, Any], List[str], str]:
+        """
+        Route to the correct handler based on constraint type.
+        Returns: (status, new_solution, affected_task_ids, summary)
+        """
+        ctype = slm_constraint.get('type', '').upper()
+
+        # Direct mutation handlers (no CP-SAT needed)
+        direct_handlers = {
             'FACULTY_SUBSTITUTION':  self._op_faculty_substitution,
             'MOVE_CLASS':            self._op_move_class,
             'SWAP_CLASSES':          self._op_swap_classes,
@@ -137,254 +98,58 @@ class PartialOptimizer:
             'CHANGE_FACULTY':        self._op_change_faculty,
             'FREEZE_SLOT':           self._op_freeze_slot,
         }
-        return handlers.get(ctype)
-
-    # ═════════════════════════════════════════════════════════════════
-    # PUBLIC ENTRY POINT (single constraint — legacy)
-    # ═════════════════════════════════════════════════════════════════
 
-    def apply_constraint_and_reoptimize(
-        self,
-        slm_constraint: Dict[str, Any],
-        time_limit: int = 30
-    ) -> Tuple[str, Dict[str, Any], List[str], str]:
-        """
-        Route to the correct handler based on constraint type.
-        Returns: (status, new_solution, affected_task_ids, summary)
-        """
-        slm_constraint = self._resolve_alias(slm_constraint)
-        ctype = slm_constraint.get('type', '').upper()
+        # Re-optimization handlers (CP-SAT partial solve)
+        reopt_handlers = {
+            'FACULTY_UNAVAILABLE':        True,
+            'FACULTY_FREE_DAY':           True,
+            'FACULTY_MAX_DAILY_HOURS':    True,
+            'FACULTY_NO_CONSECUTIVE':     True,
+            'SECTION_FREE_SLOT':          True,
+            'WORKING_DAYS':               True,
+            'SUBJECT_PREFERRED_TIME':     True,
+            'HEAVY_SUBJECT_MORNING':      True,
+            'LAB_MUST_CONSECUTIVE':       True,
+            'NO_BACK_TO_BACK_SUBJECTS':   True,
+            'DISTRIBUTE_SUBJECTS_EVENLY': True,
+            'SUBJECT_SPACING':            True,
+            'NO_FREE_PERIOD':             True,
+        }
 
-        handler = self._get_direct_handler(ctype)
-        if handler:
-            return handler(slm_constraint)
-        elif ctype in self.REOPT_TYPES:
+        # ── Type aliases — map SLM variants to canonical types ──────────
+        aliases = {
+            'SUBJECT_FREE_DAY':         'CANCEL_CLASS',
+            'FACULTY_LEAVE':            'FACULTY_SUBSTITUTION',
+            'BLOCK_SLOT':               'SECTION_FREE_SLOT',
+            'REMOVE_CLASS':             'CANCEL_CLASS',
+            'DELETE_CLASS':             'CANCEL_CLASS',
+            'CLASS_CANCELLED':          'CANCEL_CLASS',
+            'SHIFT_CLASS':              'MOVE_CLASS',
+            'RELOCATE_CLASS':           'MOVE_CLASS',
+            'TEACHER_SUBSTITUTION':     'FACULTY_SUBSTITUTION',
+            'REPLACE_FACULTY':          'FACULTY_SUBSTITUTION',
+            'SWAP_FACULTY':             'FACULTY_SUBSTITUTION',
+            'SUBJECT_UNAVAILABLE':      'CANCEL_CLASS',
+            'NO_CLASS':                 'CANCEL_CLASS',
+            'HOLIDAY':                  'MARK_HOLIDAY',
+            'MOVE_LAB':                 'RESCHEDULE_LAB',
+            'SHIFT_LAB':                'RESCHEDULE_LAB',
+            'LOCK_SLOT':                'FREEZE_SLOT',
+            'PIN_SLOT':                 'FREEZE_SLOT',
+        }
+        if ctype in aliases:
+            slm_constraint = dict(slm_constraint)
+            slm_constraint['type'] = aliases[ctype]
+            ctype = aliases[ctype]
+
+        if ctype in direct_handlers:
+            return direct_handlers[ctype](slm_constraint)
+        elif ctype in reopt_handlers:
             return self._reoptimize(slm_constraint, time_limit)
         else:
             return ('NO_CHANGE', self.current_solution, [],
                     f'⚠️ Unknown constraint type: {ctype}')
 
-    # ═════════════════════════════════════════════════════════════════
-    # PUBLIC ENTRY POINT (N constraints — batch)
-    # ═════════════════════════════════════════════════════════════════
-
-    def apply_constraints_batch(
-        self,
-        constraints: List[Dict[str, Any]],
-        time_limit: int = 30,
-    ) -> Tuple[str, Dict[str, Any], List[str], List[str]]:
-        """
-        Apply N constraints together. Direct-mutation constraints are
-        applied immediately in sequence; all re-optimization constraints
-        are batched into a SINGLE CP-SAT solve so the solver sees them
-        simultaneously (avoiding sequential conflicts).
-
-        Returns: (status, new_solution, affected_task_ids, change_summaries)
-        """
-        if not constraints:
-            return ('NO_CHANGE', self.current_solution, [], ['No constraints provided.'])
-
-        # ── 1. Resolve aliases ──────────────────────────────────────────
-        resolved = [self._resolve_alias(c) for c in constraints]
-
-        # ── 2. Classify into direct vs. reopt ───────────────────────────
-        direct_constraints = []
-        reopt_constraints = []
-        unknown_types = []
-        for c in resolved:
-            ctype = c.get('type', '').upper()
-            if ctype in self.DIRECT_TYPES:
-                direct_constraints.append(c)
-            elif ctype in self.REOPT_TYPES:
-                reopt_constraints.append(c)
-            else:
-                unknown_types.append(ctype)
-
-        all_changes: List[str] = []
-        final_solution = self.current_solution
-        overall_status = 'NO_CHANGE'
-
-        # ── 3. Apply direct-mutation constraints sequentially ───────────
-        #    (These don't need CP-SAT; they mutate the schedule dict
-        #     directly. Each sees the result of the previous one.)
-        for c in direct_constraints:
-            ctype = c.get('type', '').upper()
-            handler = self._get_direct_handler(ctype)
-            if handler:
-                # Update self.current_solution so the handler sees latest state
-                self.current_solution = final_solution
-                status, new_sol, _, summary = handler(c)
-                if status in ('OPTIMAL', 'FEASIBLE'):
-                    final_solution = new_sol
-                    overall_status = 'FEASIBLE'
-                all_changes.append(summary)
-
-        # ── 4. Batch all re-optimization constraints into ONE solve ─────
-        if reopt_constraints:
-            # Update solution reference for the reopt phase
-            self.current_solution = final_solution
-
-            # Detect conflicts before solving
-            conflicts = self.detect_conflicts(reopt_constraints)
-            if conflicts:
-                all_changes.append(
-                    f"⚠️ Potential conflicts detected:\n"
-                    + "\n".join(f"  • {cf}" for cf in conflicts)
-                )
-
-            # Collect affected tasks across ALL reopt constraints
-            all_affected = set()
-            for c in reopt_constraints:
-                affected = self._find_affected_tasks(c)
-                all_affected.update(affected)
-
-            if all_affected:
-                # Scale time limit based on constraint count and task count
-                scaled_limit = time_limit + (len(reopt_constraints) * 10)
-                scaled_limit = min(scaled_limit, 300)  # cap at 5 minutes
-
-                status, partial = self._solve_partial(
-                    list(all_affected), reopt_constraints, scaled_limit
-                )
-
-                if status in ('OPTIMAL', 'FEASIBLE'):
-                    final_solution = dict(final_solution)
-                    final_solution.update(partial)
-                    overall_status = status
-
-                    # Build change summaries
-                    changes = []
-                    for tid in all_affected:
-                        old = self.current_solution.get(tid, {})
-                        new = final_solution.get(tid, {})
-                        if old and new:
-                            old_slot = f"{const.DAYS[old['day_index']]} P{old['period_index']+1}"
-                            new_slot = f"{const.DAYS[new['day_index']]} P{new['period_index']+1}"
-                            if old_slot != new_slot:
-                                changes.append(
-                                    f"• {new.get('subject_code','?').upper()} "
-                                    f"({new.get('section_id','?')}): "
-                                    f"{old_slot} → {new_slot}"
-                                )
-
-                    summary = (
-                        f"✅ Batch: {len(reopt_constraints)} constraint(s), "
-                        f"{len(changes)} slot(s) moved:\n" + "\n".join(changes)
-                        if changes else
-                        f"✅ {len(reopt_constraints)} constraint(s) applied — no slot changes needed."
-                    )
-                    all_changes.append(summary)
-                else:
-                    all_changes.append(
-                        f"❌ Could not satisfy {len(reopt_constraints)} batched "
-                        f"constraint(s) affecting {len(all_affected)} task(s). "
-                        f"Constraints may be too restrictive or mutually conflicting."
-                    )
-                    overall_status = status
-            else:
-                all_changes.append(
-                    f"ℹ️ {len(reopt_constraints)} re-optimization constraint(s) "
-                    f"matched no affected tasks."
-                )
-
-        # ── 5. Report unknown types ─────────────────────────────────────
-        for utype in unknown_types:
-            all_changes.append(f"⚠️ Unknown constraint type: {utype}")
-
-        return (overall_status, final_solution, [], all_changes)
-
-    # ═════════════════════════════════════════════════════════════════
-    # CONFLICT DETECTION
-    # ═════════════════════════════════════════════════════════════════
-
-    def detect_conflicts(
-        self, constraints: List[Dict[str, Any]]
-    ) -> List[str]:
-        """
-        Detect obvious conflicts between constraints BEFORE solving.
-        Returns a list of human-readable conflict descriptions.
-        Does not block solving — just warns the user.
-        """
-        conflicts = []
-
-        # Index constraints by faculty and section for cross-checking
-        faculty_constraints = defaultdict(list)
-        section_constraints = defaultdict(list)
-        day_constraints = defaultdict(list)
-
-        for i, c in enumerate(constraints):
-            ctype = c.get('type', '').upper()
-            fid = c.get('faculty_id')
-            sid = c.get('section_id')
-            days = c.get('days') or []
-
-            if fid:
-                faculty_constraints[fid].append((i, c))
-            if sid:
-                section_constraints[sid].append((i, c))
-            for d in days:
-                day_constraints[d].append((i, c))
-
-        # Check: Faculty made unavailable on a day where another
-        # constraint tries to move their class TO that day
-        for fid, fac_cs in faculty_constraints.items():
-            unavailable_days = set()
-            move_to_days = set()
-            for _, c in fac_cs:
-                ctype = c.get('type', '').upper()
-                if ctype == 'FACULTY_UNAVAILABLE':
-                    for d in (c.get('days') or []):
-                        unavailable_days.add(d)
-                if ctype == 'SUBJECT_PREFERRED_TIME':
-                    # Check if we're trying to move this faculty's
-                    # subject when they're also being made unavailable
-                    move_to_days.add(c.get('subject_code', '?'))
-
-            if unavailable_days and move_to_days:
-                fac_name = self.fac_by_id.get(fid)
-                fac_label = fac_name.name if fac_name else fid
-                conflicts.append(
-                    f"Faculty '{fac_label}' is made unavailable on "
-                    f"{', '.join(unavailable_days)} but other constraints "
-                    f"affect their subjects ({', '.join(move_to_days)})"
-                )
-
-        # Check: Section has conflicting free-slot and no-free-period
-        for sid, sec_cs in section_constraints.items():
-            free_slots = set()
-            no_free_periods = set()
-            for _, c in sec_cs:
-                ctype = c.get('type', '').upper()
-                if ctype == 'SECTION_FREE_SLOT':
-                    slot = c.get('slot')
-                    if slot is not None:
-                        free_slots.add(slot - 1)  # 0-indexed
-                if ctype == 'NO_FREE_PERIOD':
-                    for p in (c.get('periods') or []):
-                        no_free_periods.add(p)
-
-            overlap = free_slots & no_free_periods
-            if overlap:
-                conflicts.append(
-                    f"Section '{sid}': period(s) {[p+1 for p in overlap]} "
-                    f"are set as BOTH free-slot AND no-free-period"
-                )
-
-        # Check: Working-day restrictions that eliminate too many days
-        for fid, fac_cs in faculty_constraints.items():
-            for _, c in fac_cs:
-                if c.get('type', '').upper() == 'WORKING_DAYS':
-                    allowed = c.get('days') or []
-                    if len(allowed) < 2:
-                        fac_name = self.fac_by_id.get(fid)
-                        fac_label = fac_name.name if fac_name else fid
-                        conflicts.append(
-                            f"Working-days constraint for '{fac_label}' allows "
-                            f"only {len(allowed)} day(s) — likely too restrictive"
-                        )
-
-        return conflicts
-
     # ═════════════════════════════════════════════════════════════════
     # DIRECT MUTATION OPERATIONS
     # ═════════════════════════════════════════════════════════════════