Update partial_optimizer.py

partial_optimizer.py

@@ -71,22 +71,61 @@ class PartialOptimizer:
             self.tasks_by_subject[t.subject.subject_code].append(t)
 
     # ═════════════════════════════════════════════════════════════════
-    # PUBLIC ENTRY POINT
+    # TYPE CLASSIFICATION HELPERS
     # ═════════════════════════════════════════════════════════════════
 
-    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 = {
+    # 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 = {
             'FACULTY_SUBSTITUTION':  self._op_faculty_substitution,
             'MOVE_CLASS':            self._op_move_class,
             'SWAP_CLASSES':          self._op_swap_classes,
@@ -98,58 +137,254 @@ class PartialOptimizer:
             'CHANGE_FACULTY':        self._op_change_faculty,
             'FREEZE_SLOT':           self._op_freeze_slot,
         }
+        return handlers.get(ctype)
 
-        # 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,
-        }
+    # ═════════════════════════════════════════════════════════════════
+    # PUBLIC ENTRY POINT (single constraint — legacy)
+    # ═════════════════════════════════════════════════════════════════
 
-        # ── 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:
+    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()
+
+        handler = self._get_direct_handler(ctype)
+        if handler:
+            return handler(slm_constraint)
+        elif ctype in self.REOPT_TYPES:
             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
     # ═════════════════════════════════════════════════════════════════