succesfull testing

This commit is contained in:
Talhadeveloperr
2026-09-09 18:55:53 +05:00
parent 868bcbe175
commit 9079ba2739
268 changed files with 2284 additions and 1495 deletions

View File

54
controller/solver/grid.py Normal file
View File

@@ -0,0 +1,54 @@
"""
Time-grid constants: days, slot definitions, lab double-slot pairs, and
the special long-lecture slot. Pure data — no scheduling logic lives here.
"""
DAYS = ["Monday", "Tuesday", "Wednesday", "Thursday", "Friday"]
# Normal theory grid: 5 x 90-minute slots, 09:00-16:30, no lunch break.
# Every placeable session (normal lecture, lab, or the special long
# lecture) is expressed as one or more of these slot indices (0-4) for
# clash-tracking purposes — see LONG_LECTURE_SLOT_INDEX below for why the
# long lecture also uses index 0 despite rendering at a different clock time.
TIME_SLOTS = [
{"label": "S1", "start": "09:00", "end": "10:30", "is_lunch": False},
{"label": "S2", "start": "10:30", "end": "12:00", "is_lunch": False},
{"label": "S3", "start": "12:00", "end": "13:30", "is_lunch": False},
{"label": "S4", "start": "13:30", "end": "15:00", "is_lunch": False},
{"label": "S5", "start": "15:00", "end": "16:30", "is_lunch": False},
]
# Indices into TIME_SLOTS that are real teaching slots (no lunch to skip
# anymore, but kept for readability at call sites)
TEACHING_SLOT_INDICES = [i for i, s in enumerate(TIME_SLOTS) if not s["is_lunch"]]
# Consecutive pairs of normal slots usable as a 180-minute lab double-slot:
# 10:30-13:30 (S2+S3) and 13:30-16:30 (S4+S5). Deliberately starts at slot
# 1, not slot 0 — slot 0 (09:00-10:30) is left free for normal lectures and
# the special long-lecture slot (see LONG_LECTURE_SLOT_INDEX below), so
# labs never crowd out a batch's only long-lecture opportunity for the day.
LAB_SLOT_PAIRS = [(1, 2), (3, 4)]
# The special fixed slot for courses whose lecture is "2 hours once a
# week": 08:30-10:30. This overlaps 09:00-10:30 (normal slot index 0) in
# wall-clock time, so a long-lecture session is tracked as occupying slot
# index 0 too (same key _is_free/_mark_busy use) — that's what makes it
# correctly clash with anything else placed in normal slot 0 that day for
# the same room/instructor/batch. duration_minutes on the session dict is
# what distinguishes it from a normal 90-minute slot-0 session at render time.
LONG_LECTURE_SLOT_INDEX = 0
LONG_LECTURE_START = "08:30"
LONG_LECTURE_END = "10:30"
# Rows to render on every generated timetable grid: the special long-lecture
# row first (it starts earliest, 08:30), then the 5 normal 90-minute rows.
# Templates/xlsx_grid_helper key session lookups on (day, slot_index,
# duration_minutes) using these rows' slot_index/duration_minutes pairs.
DISPLAY_ROWS = [
{"label": "Long", "start": LONG_LECTURE_START, "end": LONG_LECTURE_END,
"slot_index": LONG_LECTURE_SLOT_INDEX, "duration_minutes": 120},
] + [
{"label": s["label"], "start": s["start"], "end": s["end"],
"slot_index": i, "duration_minutes": 90}
for i, s in enumerate(TIME_SLOTS)
]

View File

@@ -0,0 +1,59 @@
"""
Phase 1: general courses (shared across every batch of a session).
Simple strategy — small offering count, Tue/Thu preferred first. After
placing everything, self-repair: any batch left with two general-course
sessions in the same slot has the losing session unmarked and re-placed
(scanning every day) until this phase's own sessions are clash-free.
"""
from . import grid
def run(board, data, preferred_days=("Tuesday", "Thursday"), max_repair_attempts=20):
"""Places every general-course offering's weekly lectures. Returns this phase's sessions."""
general = [o for o in data["offerings"] if o["course"]["is_general"]]
preferred_order = list(preferred_days) + [d for d in grid.DAYS if d not in preferred_days]
phase_start = len(board.sessions)
for offering in general:
course = offering["course"]
room_candidates = board.room_candidates(course["lecture_room_id"], board.classrooms)
for _ in range(course["lectures_per_week"]):
session = board.place_lecture(offering, room_candidates, preferred_order)
if session is None:
print(f"WARNING: could not place a lecture for general course {offering['course_id']} section {offering['section_id']}")
phase_sessions = board.sessions[phase_start:]
_repair_general_conflicts(board, data, phase_sessions, max_repair_attempts)
return phase_sessions
def _repair_general_conflicts(board, data, phase_sessions, max_attempts):
"""Moves general-course sessions that clash for a batch to a free slot on any day."""
from conflict_checker.student_conflict_checker import check_student_conflicts
any_day_order = list(grid.DAYS)
for _ in range(max_attempts):
conflicts = check_student_conflicts(phase_sessions)
if not conflicts:
return
# Re-place the last session involved in the first conflict; the
# other session(s) in that cell stay put.
losing_session = conflicts[0]["sessions"][-1]
offering = next(
o for o in data["offerings"]
if o["course_id"] == losing_session["course_id"] and o["section_id"] == losing_session["section_id"]
)
board.remove_session(losing_session)
phase_sessions.remove(losing_session)
room_candidates = board.room_candidates(offering["course"]["lecture_room_id"], board.classrooms)
new_session = board.place_lecture(offering, room_candidates, any_day_order)
if new_session is None:
print(f"WARNING: could not repair general-course conflict for {offering['course_id']} section {offering['section_id']}")
return
phase_sessions.append(new_session)
print("WARNING: general-course conflicts remained unresolved after max repair attempts")

View File

@@ -0,0 +1,116 @@
"""
Phase 2: labs.
Slightly more advanced than Phase 1 — places every lab offering's weekly
labs greedily first, against Phase 1's already-fixed placements. When a
lab can't be placed, this backtracks *recursively*: it bumps an
already-placed lab from earlier in this phase that shares the contested
resource (same instructor or same lab room), and tries to re-place the
stuck lab; if the bumped lab then can't find a new slot either, that
bumped lab is itself backtracked the same way (bump one of *its*
resource-sharing predecessors), and so on, until either everything
settles or the attempt budget is exhausted. Phase 1's sessions are never
touched — only Phase 2's own placements are ever moved.
"""
from . import grid
MAX_BACKTRACK_ATTEMPTS = 2000
def run(board, data):
"""Places every lab offering's weekly labs. Returns this phase's sessions."""
phase_start = len(board.sessions)
offerings = [o for o in data["offerings"] if not o["course"]["is_general"] and o["course"]["labs_per_week"] > 0]
placed_so_far = [] # [(offering, session), ...] placed so far in this phase
attempts = [0]
for offering in offerings:
course = offering["course"]
room_candidates = board.room_candidates(course["lab_room_id"], board.labs)
for _ in range(course["labs_per_week"]):
session = _place_lab_with_backtracking(board, offering, room_candidates, placed_so_far, attempts)
if session is None:
print(f"WARNING: could not place a lab for course {offering['course_id']} section {offering['section_id']} ({offering['batch_codes']}) even after backtracking")
else:
placed_so_far.append((offering, session))
return board.sessions[phase_start:]
def _place_lab_with_backtracking(board, offering, room_candidates, placed_so_far, attempts):
"""
Tries the plain first-fit placement first; if that fails, tries
bumping each resource-sharing predecessor (most recent first) out of
the way and recursively re-seating it, backing out cleanly if a given
bump doesn't lead anywhere. Returns the newly placed session dict, or
None if no arrangement works within the attempt budget.
"""
session = board.place_lab(offering, room_candidates, grid.DAYS)
if session is not None:
return session
# Snapshot the current candidates once — recursive calls will extend
# placed_so_far with their own successful re-seatings, but we only
# want to consider *this call's* view of what existed when we started.
candidate_room_ids = {r["room_id"] for r in room_candidates}
candidates = [
(other_offering, other_session)
for other_offering, other_session in placed_so_far
if _shares_contested_resource(offering, other_session, candidate_room_ids)
]
for other_offering, other_session in reversed(candidates):
if attempts[0] >= MAX_BACKTRACK_ATTEMPTS:
return None
attempts[0] += 1
if (other_offering, other_session) not in placed_so_far:
# Already moved by an earlier (failed) branch of this search.
continue
placed_so_far.remove((other_offering, other_session))
board.remove_session(other_session)
session = board.place_lab(offering, room_candidates, grid.DAYS)
if session is not None:
other_room_candidates = board.room_candidates(other_offering["course"]["lab_room_id"], board.labs)
new_other_session = _place_lab_with_backtracking(board, other_offering, other_room_candidates, placed_so_far, attempts)
if new_other_session is not None:
placed_so_far.append((other_offering, new_other_session))
return session
# Could not reseat the bumped lab even with recursion — undo
# our placement and put the bump back exactly as it was.
board.remove_session(session)
placed_so_far.append((other_offering, _restore_lab(board, other_offering, other_session)))
continue
# Bumping didn't free a usable slot for `offering` — restore and
# try the next candidate.
placed_so_far.append((other_offering, _restore_lab(board, other_offering, other_session)))
return None
def _restore_lab(board, offering, session):
"""Re-marks and re-adds a lab session dict exactly as it was."""
room = board.data["rooms_by_id"][session["room_id"]]
for slot_index in session["slot_indices"]:
board.mark_busy(session["day"], slot_index, session["room_id"], session["instructor_code"], session["batch_codes"])
return board.add_session(offering, session["day"], session["slot_indices"], room)
def _shares_contested_resource(offering, other_session, candidate_room_ids):
"""
True if bumping other_session could plausibly free a slot offering
needs: shares the instructor, a usable lab room, or one of offering's
own batches (a batch can only be in one place at a time, so another
session competing for that batch's slot is a real contested resource
too, not just instructor/room).
"""
if other_session["instructor_code"] == offering["instructor_code"]:
return True
if other_session["room_id"] in candidate_room_ids:
return True
return bool(set(other_session["batch_codes"]) & set(offering["batch_codes"]))

View File

@@ -0,0 +1,193 @@
"""
Phase 3: remaining (non-general) theory courses — batch by batch, with
recursive multi-bump backtracking.
Processes batches in sorted batch_code order. Each batch must be fully
placed with zero conflicts before moving to the next batch. If a batch's
courses can't all be placed, this backtracks into *earlier, already-
completed Phase 3 batches*: it bumps sessions that share a contested
resource (same instructor, same room, or same batch) with the stuck
offering out of the way — trying combinations of more than one bumped
session where a single bump isn't enough — then reseats every bumped
session afterward (recursively backtracking again if a reseat itself gets
stuck), until either everything settles or the attempt budget is
exhausted.
Phase 1 and Phase 2 sessions are never touched — only sessions this phase
itself placed for earlier batches are ever reopened.
"""
from . import grid
MAX_BACKTRACK_ATTEMPTS = 5000
MAX_SIMULTANEOUS_BUMPS = 3
def run(board, data):
"""
Places every remaining (non-general) theory offering's weekly
lectures, batch by batch. Returns this phase's sessions.
"""
phase_start = len(board.sessions)
offerings = [o for o in data["offerings"] if not o["course"]["is_general"]]
# placed_so_far: flat list of (offering, session) for every session
# this phase has placed so far (across all completed batches) — the
# pool backtracking is allowed to bump from.
placed_so_far = []
attempts = [0]
for batch in sorted(data["batches"], key=lambda b: b["batch_code"]):
batch_code = batch["batch_code"]
batch_offerings = [o for o in offerings if batch_code in o["batch_codes"]]
# Place the most slot-constrained offerings first (long-lecture
# courses can only ever use slot 0, so they have the fewest
# options) — this way flexible 90-min courses fill in around them
# instead of greedily grabbing slot 0 first and starving a
# long-lecture course that has nowhere else to go.
batch_offerings.sort(key=lambda o: 0 if o["course"]["lecture_duration_minutes"] == 120 else 1)
for offering in batch_offerings:
course = offering["course"]
room_candidates = board.room_candidates(course["lecture_room_id"], board.classrooms)
for _ in range(course["lectures_per_week"]):
session = _place_lecture_with_backtracking(board, offering, room_candidates, placed_so_far, attempts)
if session is None:
print(
f"WARNING: could not place a lecture for course {offering['course_id']} "
f"section {offering['section_id']} (batch {batch_code}) even after backtracking"
)
else:
placed_so_far.append((offering, session))
return board.sessions[phase_start:]
def _place_lecture_with_backtracking(board, offering, room_candidates, placed_so_far, attempts):
"""
Tries the plain first-fit placement first; if that fails, tries
bumping *combinations* of resource-sharing predecessors out of the way
(starting with one at a time, then pairs, up to MAX_SIMULTANEOUS_BUMPS
together) until `offering` fits, then reseats every bumped session
(recursively backtracking again if a reseat itself gets stuck).
Returns the newly placed session dict, or None if no arrangement works
within the attempt budget.
"""
session = board.place_lecture(offering, room_candidates, grid.DAYS)
if session is not None:
return session
candidate_room_ids = {r["room_id"] for r in room_candidates}
usable_slot_indices = {grid.LONG_LECTURE_SLOT_INDEX} if offering["course"]["lecture_duration_minutes"] == 120 else set(grid.TEACHING_SLOT_INDICES)
candidates = [
(other_offering, other_session)
for other_offering, other_session in placed_so_far
if _shares_contested_resource(offering, other_session, candidate_room_ids, usable_slot_indices)
]
# Most recently placed first — later placements are more likely to be
# "loose" (easier to reseat) than earlier, already-settled ones.
candidates.reverse()
for bump_count in range(1, min(MAX_SIMULTANEOUS_BUMPS, len(candidates)) + 1):
result = _try_bump_combinations(board, offering, room_candidates, placed_so_far, attempts, candidates, bump_count)
if result is not None:
return result
if attempts[0] >= MAX_BACKTRACK_ATTEMPTS:
return None
return None
def _try_bump_combinations(board, offering, room_candidates, placed_so_far, attempts, candidates, bump_count):
"""
Tries every way of picking `bump_count` candidates (in order) to bump
simultaneously, retrying `offering`'s placement after each bump.
Returns the newly placed session on success (with all bumped sessions
reseated), or None if no combination of this size works.
"""
from itertools import combinations
for combo in combinations(range(len(candidates)), bump_count):
if attempts[0] >= MAX_BACKTRACK_ATTEMPTS:
return None
chosen = [candidates[i] for i in combo]
if any((o, s) not in placed_so_far for o, s in chosen):
# One of these was already moved by an earlier failed branch.
continue
bumped = []
for other_offering, other_session in chosen:
attempts[0] += 1
placed_so_far.remove((other_offering, other_session))
board.remove_session(other_session)
bumped.append((other_offering, other_session))
session = board.place_lecture(offering, room_candidates, grid.DAYS)
if session is not None:
if _reseat_all(board, placed_so_far, attempts, bumped):
return session
# Reseating some bumped session failed even with recursion —
# undo our placement and restore everything bumped this round.
board.remove_session(session)
for other_offering, other_session in bumped:
placed_so_far.append((other_offering, _restore_session(board, other_offering, other_session)))
continue
# This combination didn't free a usable slot — restore everything
# bumped this round and try the next combination.
for other_offering, other_session in bumped:
placed_so_far.append((other_offering, _restore_session(board, other_offering, other_session)))
return None
def _reseat_all(board, placed_so_far, attempts, bumped):
"""
Reseats every (offering, session) in `bumped` (each currently removed
from the board), backtracking recursively if needed. All-or-nothing:
on any failure, every session reseated so far in this attempt is
removed again and `bumped` remains fully un-seated (the caller is
responsible for restoring them to their original slots).
"""
reseated = []
for other_offering, other_session in bumped:
other_room_candidates = board.room_candidates(other_offering["course"]["lecture_room_id"], board.classrooms)
new_session = _place_lecture_with_backtracking(board, other_offering, other_room_candidates, placed_so_far, attempts)
if new_session is None:
for reseated_offering, reseated_session in reseated:
placed_so_far.remove((reseated_offering, reseated_session))
board.remove_session(reseated_session)
return False
reseated.append((other_offering, new_session))
placed_so_far.append((other_offering, new_session))
return True
def _restore_session(board, offering, session):
"""Re-marks and re-adds a session dict exactly as it was, at the same day/slots/room."""
room = board.data["rooms_by_id"][session["room_id"]]
for slot_index in session["slot_indices"]:
board.mark_busy(session["day"], slot_index, session["room_id"], session["instructor_code"], session["batch_codes"])
return board.add_session(
offering, session["day"], session["slot_indices"], room,
is_long_lecture=(session["duration_minutes"] == 120),
)
def _shares_contested_resource(offering, other_session, candidate_room_ids, usable_slot_indices):
"""
True if bumping other_session could plausibly free a slot `offering`
can actually use: other_session must occupy at least one slot index
`offering` is eligible for, AND share the instructor, a usable room,
or (crucially) one of offering's own batches — a batch can only be in
one place at a time, so another course competing for the *same batch's*
slot-0 time is just as real a contested resource as instructor/room.
"""
if not any(slot_index in usable_slot_indices for slot_index in other_session["slot_indices"]):
return False
if other_session["instructor_code"] == offering["instructor_code"]:
return True
if other_session["room_id"] in candidate_room_ids:
return True
return bool(set(other_session["batch_codes"]) & set(offering["batch_codes"]))

View File

@@ -0,0 +1,171 @@
"""
PlacementBoard: shared clash-tracking state and low-level placement
primitives. Owns the room/instructor/batch busy-maps and the accumulated
session list. Has zero scheduling policy of its own — every phase module
(phase1_general, phase2_labs, phase3_remaining) reads and writes the same
board, which is what makes each later phase automatically see earlier
phases' placements as fixed/locked.
"""
from . import grid
class PlacementBoard:
def __init__(self, data):
self.data = data
self.classrooms = [r for r in data["rooms"] if r["room_type"] == "classroom"]
self.labs = [r for r in data["rooms"] if r["room_type"] == "lab"]
# busy[(day, slot_index)] -> set of keys already occupied
self.room_busy = {}
self.instructor_busy = {}
self.batch_busy = {}
self.sessions = []
# -- serialization -----------------------------------------------------
def to_state(self):
"""Serializes accumulated sessions + busy-maps to a JSON-able dict."""
def _dump_busy(busy_map):
return {f"{day}|{slot_index}": sorted(values) for (day, slot_index), values in busy_map.items()}
return {
"sessions": self.sessions,
"room_busy": _dump_busy(self.room_busy),
"instructor_busy": _dump_busy(self.instructor_busy),
"batch_busy": _dump_busy(self.batch_busy),
}
@classmethod
def from_state(cls, data, state):
"""Rebuilds a PlacementBoard from a dict previously produced by to_state()."""
def _load_busy(dumped):
busy = {}
for key, values in dumped.items():
day, slot_index = key.rsplit("|", 1)
busy[(day, int(slot_index))] = set(values)
return busy
board = cls(data)
board.sessions = list(state["sessions"])
board.room_busy = _load_busy(state["room_busy"])
board.instructor_busy = _load_busy(state["instructor_busy"])
board.batch_busy = _load_busy(state["batch_busy"])
return board
# -- clash-tracking primitives ------------------------------------------
def is_free(self, day, slot_index, room_id, instructor_code, batch_codes):
key = (day, slot_index)
if room_id in self.room_busy.get(key, set()):
return False
if instructor_code in self.instructor_busy.get(key, set()):
return False
if any(b in self.batch_busy.get(key, set()) for b in batch_codes):
return False
return True
def mark_busy(self, day, slot_index, room_id, instructor_code, batch_codes):
key = (day, slot_index)
self.room_busy.setdefault(key, set()).add(room_id)
self.instructor_busy.setdefault(key, set()).add(instructor_code)
self.batch_busy.setdefault(key, set()).update(batch_codes)
def unmark_busy(self, day, slot_index, room_id, instructor_code, batch_codes):
key = (day, slot_index)
self.room_busy.get(key, set()).discard(room_id)
self.instructor_busy.get(key, set()).discard(instructor_code)
for b in batch_codes:
self.batch_busy.get(key, set()).discard(b)
def room_candidates(self, preferred_room_id, room_pool):
"""Preferred room first, then the rest of the same-type pool."""
preferred = self.data["rooms_by_id"].get(preferred_room_id)
rest = [r for r in room_pool if r["room_id"] != preferred_room_id]
return ([preferred] if preferred else []) + rest
# -- session creation ----------------------------------------------------
def add_session(self, offering, day, slot_indices, room, is_long_lecture=False):
course = offering["course"]
if is_long_lecture:
start, end = grid.LONG_LECTURE_START, grid.LONG_LECTURE_END
duration_minutes = 120
session_type = "theory"
else:
start = grid.TIME_SLOTS[slot_indices[0]]["start"]
end = grid.TIME_SLOTS[slot_indices[-1]]["end"]
duration_minutes = 90 * len(slot_indices)
session_type = "lab" if len(slot_indices) > 1 else "theory"
session = {
"day": day,
"slot_indices": slot_indices,
"start": start,
"end": end,
"duration_minutes": duration_minutes,
"course_id": offering["course_id"],
"course_name": course["course_name"],
"course_type": session_type,
"is_general": course["is_general"],
"section_id": offering["section_id"],
"instructor_code": offering["instructor_code"],
"instructor_name": offering["instructor_name"],
"room_id": room["room_id"],
"room_name": room["room_name"],
"batch_codes": offering["batch_codes"],
}
self.sessions.append(session)
return session
def remove_session(self, session):
"""Unmarks a previously placed session's slots and removes it from self.sessions."""
for slot_index in session["slot_indices"]:
self.unmark_busy(
session["day"], slot_index,
session["room_id"], session["instructor_code"], session["batch_codes"],
)
self.sessions.remove(session)
# -- placement helpers shared by phase modules --------------------------
def place_theory_session(self, offering, room_candidates, day_order):
"""Finds and locks a free normal 90-minute single-slot session; returns the session dict or None."""
for day in day_order:
for slot_index in grid.TEACHING_SLOT_INDICES:
for room in room_candidates:
if self.is_free(day, slot_index, room["room_id"], offering["instructor_code"], offering["batch_codes"]):
self.mark_busy(day, slot_index, room["room_id"], offering["instructor_code"], offering["batch_codes"])
return self.add_session(offering, day, [slot_index], room)
return None
def place_long_lecture(self, offering, room_candidates, day_order):
"""
Finds and locks a free 08:30-10:30 long-lecture session. Occupies
the same clash-tracking slot index as normal slot 0 since they
overlap in wall-clock time. Returns the session dict or None.
"""
for day in day_order:
for room in room_candidates:
if self.is_free(day, grid.LONG_LECTURE_SLOT_INDEX, room["room_id"], offering["instructor_code"], offering["batch_codes"]):
self.mark_busy(day, grid.LONG_LECTURE_SLOT_INDEX, room["room_id"], offering["instructor_code"], offering["batch_codes"])
return self.add_session(offering, day, [grid.LONG_LECTURE_SLOT_INDEX], room, is_long_lecture=True)
return None
def place_lecture(self, offering, room_candidates, day_order):
"""Dispatches to the long (120-min) or normal (90-min) lecture placement based on the course's lecture_duration_minutes."""
if offering["course"]["lecture_duration_minutes"] == 120:
return self.place_long_lecture(offering, room_candidates, day_order)
return self.place_theory_session(offering, room_candidates, day_order)
def place_lab(self, offering, room_candidates, day_order):
"""Finds and locks a free 180-minute lab double-slot session; returns the session dict or None."""
for day in day_order:
for slot_a, slot_b in grid.LAB_SLOT_PAIRS:
for room in room_candidates:
if self.is_free(day, slot_a, room["room_id"], offering["instructor_code"], offering["batch_codes"]) and \
self.is_free(day, slot_b, room["room_id"], offering["instructor_code"], offering["batch_codes"]):
self.mark_busy(day, slot_a, room["room_id"], offering["instructor_code"], offering["batch_codes"])
self.mark_busy(day, slot_b, room["room_id"], offering["instructor_code"], offering["batch_codes"])
return self.add_session(offering, day, [slot_a, slot_b], room)
return None

View File

@@ -1,24 +1,23 @@
"""
Loads dummy CSV data and builds a clash-free schedule of sessions that all
five templates/*-wise template modules render from.
Loads dummy CSV data and exposes a thin Scheduler facade over the solver
engine in controller/solver/ (grid constants, PlacementBoard, and one
module per phase — phase1_general, phase2_labs, phase3_remaining).
Scheduling is done in three sequential, independently-verified phases (see
app/template_generator.py for the orchestration + per-phase conflict
checks). Each phase's placements are locked before the next phase starts:
once a (day, slot, room/instructor/batch) combination is marked busy by an
earlier phase, later phases can never reuse it.
Scheduling is done in three sequential, independently-verified phases, each
with its own tailored strategy (see controller/solver/phase*_*.py for the
per-phase logic). Each phase's placements are locked before the next phase
starts: once a (day, slot, room/instructor/batch) combination is marked
busy by an earlier phase, later phases can never reuse it — except Phase
3, which is allowed to backtrack and reopen *its own* earlier batches
(never Phase 1 or Phase 2) when a later batch can't otherwise be placed.
1. General courses (shared across every batch of a session), preferring
Tuesday/Thursday first. Each offering gets `lectures_per_week`
sessions. If two general courses still land in the same slot for a
batch, the losing session is moved to another day/slot until this
phase is clash-free (see `phase_1_general_courses`).
2. Labs: each offering gets `labs_per_week` 180-minute / double-slot
sessions, in lab rooms only, scheduled against Phase 1's fixed
placements plus labs placed so far this phase.
Tuesday/Thursday first, self-repaired until clash-free.
2. Labs, scheduled against Phase 1's fixed placements, then self-repaired
for lab-vs-lab clashes within Phase 2 only.
3. Remaining (non-general) theory sessions, scheduled batch by batch in
sorted batch_code order: for each batch, every one of its remaining
offerings is placed and locked before the next batch is considered.
sorted batch_code order, with backtracking into earlier Phase 3
batches when a later batch can't otherwise fit.
Time grid: 5 normal 90-minute slots, 09:00-16:30, no lunch break. Labs use
180-minute double-slots (09:00-12:00 and 12:00-15:00). A course whose
@@ -34,55 +33,15 @@ preferred one is busy for that day/slot.
import csv
import os
from solver.grid import ( # noqa: F401 (re-exported for templates/backward compatibility)
DAYS, TIME_SLOTS, TEACHING_SLOT_INDICES, LAB_SLOT_PAIRS,
LONG_LECTURE_SLOT_INDEX, LONG_LECTURE_START, LONG_LECTURE_END, DISPLAY_ROWS,
)
from solver.placement_board import PlacementBoard
from solver import phase1_general, phase2_labs, phase3_remaining
DATA_DIR = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "data", "dummy")
DAYS = ["Monday", "Tuesday", "Wednesday", "Thursday", "Friday"]
# Normal theory grid: 5 x 90-minute slots, 09:00-16:30, no lunch break.
# Every placeable session (normal lecture, lab, or the special long
# lecture) is expressed as one or more of these slot indices (0-4) for
# clash-tracking purposes — see LONG_LECTURE_SLOT below for why the long
# lecture also uses index 0 despite rendering at a different clock time.
TIME_SLOTS = [
{"label": "S1", "start": "09:00", "end": "10:30", "is_lunch": False},
{"label": "S2", "start": "10:30", "end": "12:00", "is_lunch": False},
{"label": "S3", "start": "12:00", "end": "13:30", "is_lunch": False},
{"label": "S4", "start": "13:30", "end": "15:00", "is_lunch": False},
{"label": "S5", "start": "15:00", "end": "16:30", "is_lunch": False},
]
# Indices into TIME_SLOTS that are real teaching slots (no lunch to skip
# anymore, but kept for readability at call sites)
TEACHING_SLOT_INDICES = [i for i, s in enumerate(TIME_SLOTS) if not s["is_lunch"]]
# Consecutive pairs of normal slots usable as a 180-minute lab double-slot:
# 09:00-12:00 (S1+S2) and 12:00-15:00 (S3+S4).
LAB_SLOT_PAIRS = [(0, 1), (2, 3)]
# The special fixed slot for courses whose lecture is "2 hours once a
# week": 08:30-10:30. This overlaps 09:00-10:30 (normal slot index 0) in
# wall-clock time, so a long-lecture session is tracked as occupying slot
# index 0 too (same key _is_free/_mark_busy use) — that's what makes it
# correctly clash with anything else placed in normal slot 0 that day for
# the same room/instructor/batch. duration_minutes on the session dict is
# what distinguishes it from a normal 90-minute slot-0 session at render time.
LONG_LECTURE_SLOT_INDEX = 0
LONG_LECTURE_START = "08:30"
LONG_LECTURE_END = "10:30"
# Rows to render on every generated timetable grid: the special long-lecture
# row first (it starts earliest, 08:30), then the 5 normal 90-minute rows.
# Templates/xlsx_grid_helper key session lookups on (day, slot_index,
# duration_minutes) using these rows' slot_index/duration_minutes pairs.
DISPLAY_ROWS = [
{"label": "Long", "start": LONG_LECTURE_START, "end": LONG_LECTURE_END,
"slot_index": LONG_LECTURE_SLOT_INDEX, "duration_minutes": 120},
] + [
{"label": s["label"], "start": s["start"], "end": s["end"],
"slot_index": i, "duration_minutes": 90}
for i, s in enumerate(TIME_SLOTS)
]
def _read_csv(relative_path):
path = os.path.join(DATA_DIR, relative_path)
@@ -139,263 +98,44 @@ def load_data():
class Scheduler:
"""Greedy clash-free scheduler over rooms / instructors / batches."""
"""
Thin facade over the solver engine: owns a PlacementBoard and
delegates each phase to its dedicated solver/phase*_*.py module.
Keeps the same public surface the rest of the app already relies on
(app/template_generator.py, controller/conflict_checker/*).
"""
def __init__(self, data):
self.data = data
self.classrooms = [r for r in data["rooms"] if r["room_type"] == "classroom"]
self.labs = [r for r in data["rooms"] if r["room_type"] == "lab"]
self.board = PlacementBoard(data)
# busy[(day, slot_index)] -> set of keys already occupied
self.room_busy = {}
self.instructor_busy = {}
self.batch_busy = {}
self.sessions = []
@property
def sessions(self):
return self.board.sessions
def to_state(self):
"""
Serializes this scheduler's accumulated sessions and busy-maps to a
JSON-able dict, so a later process can resume exactly where this
one left off (used to run each phase as a separate CLI command).
"""
def _dump_busy(busy_map):
return {f"{day}|{slot_index}": sorted(values) for (day, slot_index), values in busy_map.items()}
return {
"sessions": self.sessions,
"room_busy": _dump_busy(self.room_busy),
"instructor_busy": _dump_busy(self.instructor_busy),
"batch_busy": _dump_busy(self.batch_busy),
}
return self.board.to_state()
@classmethod
def from_state(cls, data, state):
"""Rebuilds a Scheduler from a dict previously produced by to_state()."""
def _load_busy(dumped):
busy = {}
for key, values in dumped.items():
day, slot_index = key.rsplit("|", 1)
busy[(day, int(slot_index))] = set(values)
return busy
scheduler = cls(data)
scheduler.sessions = list(state["sessions"])
scheduler.room_busy = _load_busy(state["room_busy"])
scheduler.instructor_busy = _load_busy(state["instructor_busy"])
scheduler.batch_busy = _load_busy(state["batch_busy"])
scheduler.board = PlacementBoard.from_state(data, state)
return scheduler
def _is_free(self, day, slot_index, room_id, instructor_code, batch_codes):
key = (day, slot_index)
if room_id in self.room_busy.get(key, set()):
return False
if instructor_code in self.instructor_busy.get(key, set()):
return False
if any(b in self.batch_busy.get(key, set()) for b in batch_codes):
return False
return True
def _mark_busy(self, day, slot_index, room_id, instructor_code, batch_codes):
key = (day, slot_index)
self.room_busy.setdefault(key, set()).add(room_id)
self.instructor_busy.setdefault(key, set()).add(instructor_code)
self.batch_busy.setdefault(key, set()).update(batch_codes)
def _unmark_busy(self, day, slot_index, room_id, instructor_code, batch_codes):
key = (day, slot_index)
self.room_busy.get(key, set()).discard(room_id)
self.instructor_busy.get(key, set()).discard(instructor_code)
for b in batch_codes:
self.batch_busy.get(key, set()).discard(b)
def _room_candidates(self, preferred_room_id, room_pool):
"""Preferred room first, then the rest of the same-type pool."""
preferred = self.data["rooms_by_id"].get(preferred_room_id)
rest = [r for r in room_pool if r["room_id"] != preferred_room_id]
return ([preferred] if preferred else []) + rest
def _add_session(self, offering, day, slot_indices, room, is_long_lecture=False):
course = offering["course"]
if is_long_lecture:
start, end = LONG_LECTURE_START, LONG_LECTURE_END
duration_minutes = 120
session_type = "theory"
else:
start = TIME_SLOTS[slot_indices[0]]["start"]
end = TIME_SLOTS[slot_indices[-1]]["end"]
duration_minutes = 90 * len(slot_indices)
session_type = "lab" if len(slot_indices) > 1 else "theory"
session = {
"day": day,
"slot_indices": slot_indices,
"start": start,
"end": end,
"duration_minutes": duration_minutes,
"course_id": offering["course_id"],
"course_name": course["course_name"],
"course_type": session_type,
"is_general": course["is_general"],
"section_id": offering["section_id"],
"instructor_code": offering["instructor_code"],
"instructor_name": offering["instructor_name"],
"room_id": room["room_id"],
"room_name": room["room_name"],
"batch_codes": offering["batch_codes"],
}
self.sessions.append(session)
return session
def _place_theory_session(self, offering, room_candidates, day_order):
"""Finds and locks a free normal 90-minute single-slot session; returns the session dict or None."""
for day in day_order:
for slot_index in TEACHING_SLOT_INDICES:
for room in room_candidates:
if self._is_free(day, slot_index, room["room_id"], offering["instructor_code"], offering["batch_codes"]):
self._mark_busy(day, slot_index, room["room_id"], offering["instructor_code"], offering["batch_codes"])
return self._add_session(offering, day, [slot_index], room)
return None
def _place_long_lecture(self, offering, room_candidates, day_order):
"""
Finds and locks a free 08:30-10:30 long-lecture session (for
courses with lecture_duration_minutes=120). Occupies the same
clash-tracking slot index as normal slot 0 since they overlap in
wall-clock time (09:00-10:30). Returns the session dict or None.
"""
for day in day_order:
for room in room_candidates:
if self._is_free(day, LONG_LECTURE_SLOT_INDEX, room["room_id"], offering["instructor_code"], offering["batch_codes"]):
self._mark_busy(day, LONG_LECTURE_SLOT_INDEX, room["room_id"], offering["instructor_code"], offering["batch_codes"])
return self._add_session(offering, day, [LONG_LECTURE_SLOT_INDEX], room, is_long_lecture=True)
return None
def _place_lecture(self, offering, room_candidates, day_order):
"""Dispatches to the long (120-min) or normal (90-min) lecture placement based on the course's lecture_duration_minutes."""
if offering["course"]["lecture_duration_minutes"] == 120:
return self._place_long_lecture(offering, room_candidates, day_order)
return self._place_theory_session(offering, room_candidates, day_order)
def phase_1_general_courses(self, preferred_days=("Tuesday", "Thursday"), max_repair_attempts=20):
"""
Places every general-course offering's weekly lectures, preferring
`preferred_days` first. Afterwards, self-repairs: any batch left
with two general-course sessions in the same slot has the losing
session unmarked and re-placed (scanning every day) until this
phase's own sessions are clash-free. Returns this phase's sessions.
"""
general = [o for o in self.data["offerings"] if o["course"]["is_general"]]
preferred_order = list(preferred_days) + [d for d in DAYS if d not in preferred_days]
phase_start = len(self.sessions)
for offering in general:
course = offering["course"]
room_candidates = self._room_candidates(course["lecture_room_id"], self.classrooms)
for _ in range(course["lectures_per_week"]):
session = self._place_lecture(offering, room_candidates, preferred_order)
if session is None:
print(f"WARNING: could not place a lecture for general course {offering['course_id']} section {offering['section_id']}")
phase_sessions = self.sessions[phase_start:]
self._repair_general_conflicts(phase_sessions, preferred_order, max_repair_attempts)
return phase_sessions
def _repair_general_conflicts(self, phase_sessions, day_order, max_attempts):
"""Moves general-course sessions that clash for a batch to a free slot on any day."""
from conflict_checker.student_conflict_checker import check_student_conflicts
any_day_order = list(DAYS)
for _ in range(max_attempts):
conflicts = check_student_conflicts(phase_sessions)
if not conflicts:
return
# Re-place the last session involved in the first conflict; the
# other session(s) in that cell stay put.
losing_session = conflicts[0]["sessions"][-1]
offering = next(
o for o in self.data["offerings"]
if o["course_id"] == losing_session["course_id"] and o["section_id"] == losing_session["section_id"]
)
self._unmark_busy(
losing_session["day"], losing_session["slot_indices"][0],
losing_session["room_id"], losing_session["instructor_code"], losing_session["batch_codes"],
)
self.sessions.remove(losing_session)
phase_sessions.remove(losing_session)
room_candidates = self._room_candidates(offering["course"]["lecture_room_id"], self.classrooms)
new_session = self._place_lecture(offering, room_candidates, any_day_order)
if new_session is None:
print(f"WARNING: could not repair general-course conflict for {offering['course_id']} section {offering['section_id']}")
return
phase_sessions.append(new_session)
print("WARNING: general-course conflicts remained unresolved after max repair attempts")
def phase_1_general_courses(self):
return phase1_general.run(self.board, self.data)
def phase_2_labs(self):
"""
Places every lab offering's weekly labs into lab rooms, treating
Phase 1's placements (already marked busy) as fixed. Returns this
phase's sessions.
"""
phase_start = len(self.sessions)
offerings = [o for o in self.data["offerings"] if not o["course"]["is_general"] and o["course"]["labs_per_week"] > 0]
for offering in offerings:
course = offering["course"]
room_candidates = self._room_candidates(course["lab_room_id"], self.labs)
for _ in range(course["labs_per_week"]):
placed = False
for day in DAYS:
for slot_a, slot_b in LAB_SLOT_PAIRS:
for room in room_candidates:
if self._is_free(day, slot_a, room["room_id"], offering["instructor_code"], offering["batch_codes"]) and \
self._is_free(day, slot_b, room["room_id"], offering["instructor_code"], offering["batch_codes"]):
self._mark_busy(day, slot_a, room["room_id"], offering["instructor_code"], offering["batch_codes"])
self._mark_busy(day, slot_b, room["room_id"], offering["instructor_code"], offering["batch_codes"])
self._add_session(offering, day, [slot_a, slot_b], room)
placed = True
break
if placed:
break
if placed:
break
if not placed:
print(f"WARNING: could not place a lab for course {offering['course_id']} section {offering['section_id']} ({offering['batch_codes']})")
return self.sessions[phase_start:]
return phase2_labs.run(self.board, self.data)
def phase_3_remaining_courses(self):
"""
Places every remaining (non-general) theory offering's weekly
lectures, batch by batch in sorted batch_code order: all of one
batch's offerings are placed and locked before the next batch is
considered. Treats Phases 1+2 (already marked busy) as fixed.
Returns this phase's sessions.
"""
phase_start = len(self.sessions)
offerings = [o for o in self.data["offerings"] if not o["course"]["is_general"]]
for batch in sorted(self.data["batches"], key=lambda b: b["batch_code"]):
batch_code = batch["batch_code"]
batch_offerings = [o for o in offerings if batch_code in o["batch_codes"]]
for offering in batch_offerings:
course = offering["course"]
room_candidates = self._room_candidates(course["lecture_room_id"], self.classrooms)
for _ in range(course["lectures_per_week"]):
session = self._place_lecture(offering, room_candidates, DAYS)
if session is None:
print(f"WARNING: could not place a lecture for course {offering['course_id']} section {offering['section_id']} ({offering['batch_codes']})")
return self.sessions[phase_start:]
return phase3_remaining.run(self.board, self.data)
def run(self):
self.phase_1_general_courses()
self.phase_2_labs()
self.phase_3_remaining_courses()
return self.sessions
return self.board.sessions
def build_schedule():