diff --git "a/api.py" "b/api.py" --- "a/api.py" +++ "b/api.py" @@ -1,639 +1,1843 @@ -# app.py - -import streamlit as st -import pandas as pd +from fastapi import FastAPI, HTTPException, File, UploadFile, Body +from fastapi.responses import StreamingResponse +from fastapi.middleware.cors import CORSMiddleware +from pydantic import BaseModel +from typing import List, Optional from collections import defaultdict +import traceback from datetime import datetime -from storage import (load_data, save_data, save_schedule, load_schedule, - schedule_exists, clear_schedule, load_history, - add_history_entry, clear_history, - save_original_schedule, load_original_schedule, - original_schedule_exists, clear_original_schedule) +from storage import ( + load_data, save_data, + save_schedule, save_original_schedule, + schedule_exists, load_schedule, + original_schedule_exists, load_original_schedule, + load_history, add_history_entry, save_history, + clear_schedule, clear_history, clear_original_schedule, + save_version, load_versions, restore_version, +) from models import Faculty, Subject, Section, Room, SubjectType from data_loader import Allocation, prepare_scheduling_tasks from solver import TimetableSolver -from slm_inference import get_constraint, check_api_health from partial_optimizer import PartialOptimizer +from slm_inference import get_constraints_batch, smart_parse, get_constraint, check_api_health +from substitution_engine import ( + process_leave_approval, handle_acceptance, handle_decline, check_timeouts +) +from storage import ( + load_leave_requests, save_leave_requests, load_substitution_requests, + save_substitution_requests, load_cancellations, save_cancellations +) +from models import LeaveRequest, LeaveStatus +import uuid import constants as const -st.set_page_config(page_title="VTU Timetable Generator", layout="wide") +app = FastAPI(title="VTU Timetable Generator API", version="2.0.0") + +app.add_middleware( + CORSMiddleware, + allow_origins=["*"], + allow_credentials=True, + allow_methods=["*"], + allow_headers=["*"], +) + + +# ═════════════════════════════════════════════════════════════════════════════ +# INTERNAL HELPERS +# ═════════════════════════════════════════════════════════════════════════════ -# ── Helpers ─────────────────────────────────────────────────────────────── -def get_subject_type_enum(type_str): +def _subject_type(type_str: str) -> SubjectType: return { - "THEORY": SubjectType.THEORY, "LAB": SubjectType.LAB, - "SOFTSKILL": SubjectType.SOFTSKILL, "FORUM": SubjectType.FORUM + "THEORY": SubjectType.THEORY, + "LAB": SubjectType.LAB, + "SOFTSKILL": SubjectType.SOFTSKILL, + "FORUM": SubjectType.FORUM, }.get(type_str.upper(), SubjectType.THEORY) -def convert_json_to_objects(data): - fac_objs = [Faculty(f['id'], f['name'], f['designation'], f['max_hours']) - for f in data['faculties']] - sub_objs = [Subject(s['code'], s['name'], s['credits'], - get_subject_type_enum(s['type']), - s.get('is_core', True), s.get('is_heavy', False)) - for s in data['subjects']] - sec_objs = [Section(s['id'], s['semester'], s['strength']) - for s in data['sections']] - room_objs = [Room(r['id'], r['capacity'], r['is_lab'], r['building']) - for r in data['rooms']] - alloc_objs = [Allocation(a['faculty_id'], a['subject_code'], - a['section_id'], a.get('elective_group')) - for a in data['allocations']] - return fac_objs, sub_objs, sec_objs, room_objs, alloc_objs - -def rebuild_objects(data): - """Rebuild all domain objects from stored data.""" - return convert_json_to_objects(data) - -def render_timetable_html(solution, sections): - """Render the timetable as an HTML table.""" - parent_sections = sorted(list(set( - s.section_id.split('-')[0].upper() for s in sections))) - - # Grid stores list of (subject_code, faculty_name) per slot - merged_grid = defaultdict(lambda: defaultdict(lambda: defaultdict(list))) + +def _build_objects(data: dict): + """Convert raw JSON data dicts into domain model objects.""" + facs = [Faculty(f["id"], f["name"], f["designation"], f["max_hours"]) + for f in data["faculties"]] + subs = [Subject(s["code"], s["name"], s["credits"], + _subject_type(s["type"]), + s.get("is_core", True), s.get("is_heavy", False)) + for s in data["subjects"]] + secs = [Section(s["id"], s["semester"], s["strength"]) + for s in data["sections"]] + rooms = [Room(r["id"], r["capacity"], r["is_lab"], r["building"]) + for r in data["rooms"]] + allocs = [Allocation(a["faculty_id"], a["subject_code"], + a["section_id"], a.get("elective_group")) + for a in data["allocations"]] + return facs, subs, secs, rooms, allocs + + +def _clean(solution: dict) -> dict: + """Remove non-serialisable task_obj from solution.""" + return {k: {kk: vv for kk, vv in v.items() if kk != "task_obj"} + for k, v in solution.items()} + +def diff_schedules(old_sched: dict, new_sched: dict): + changes = [] + affected_sections = set() + all_keys = set(old_sched.keys()) | set(new_sched.keys()) + for k in all_keys: + old_val = old_sched.get(k) + new_val = new_sched.get(k) + if old_val != new_val: + changes.append({ + "task_id": k, + "before": old_val, + "after": new_val + }) + if old_val: affected_sections.add(old_val.get("section_id", "").split("-")[0].upper()) + if new_val: affected_sections.add(new_val.get("section_id", "").split("-")[0].upper()) + return changes, list(affected_sections) + + +def _build_grid(solution: dict, allocations: list = None) -> dict: + """ + Merge sub-sections (6a-E1, 6a-E2) into their parent (6A) — + exactly what render_timetable_html does in app.py. + + Now also enriches grid entries with: + - duration: from the schedule entry + - elective_group: from allocations data + - is_open_elective: true if elective_group contains 'oe' + """ + if allocations is None: + allocations = load_data().get("allocations", []) + + # Build allocation lookup: (section_id, subject_code, faculty_id) -> elective_group + alloc_lookup = {} + if allocations: + for a in allocations: + key = ( + a.get("section_id", "").lower(), + a.get("subject_code", "").lower(), + a.get("faculty_id", "").lower(), + ) + alloc_lookup[key] = a.get("elective_group") + + # Build faculty_id reverse lookup from data + # The schedule stores faculty_name (e.g. "Prof. Anu") but allocations use faculty_id (e.g. "anu") + # We'll also try matching by section_id + subject_code only as fallback + alloc_by_sec_sub = {} + if allocations: + for a in allocations: + key2 = (a.get("section_id", "").lower(), a.get("subject_code", "").lower()) + alloc_by_sec_sub[key2] = a.get("elective_group") + + parent_sections = sorted(set( + info.get("section_id", "").split("-")[0].upper() + for info in solution.values() + )) + + merged = {ps: defaultdict(lambda: defaultdict(list)) for ps in parent_sections} + days_seen = {ps: set() for ps in parent_sections} + for task_id, info in solution.items(): - sec_id = info.get('section_id', '') - parent_sec = sec_id.split('-')[0].upper() - day = info.get('day_index', 0) - period = info.get('period_index', 0) - dur = info.get('duration', 1) - subject = info.get('subject_code', '?').upper() - faculty = info.get('faculty_name', '') - short_fac = (faculty.replace('Prof. ','').replace('Dr. ','') - .replace('Mr. ','').replace('Ms. ','')) + sec_id = info.get("section_id") or "" + ps = sec_id.split("-")[0].upper() if sec_id else "" + day = info.get("day_index", 0) + period = info.get("period_index", 0) + dur = info.get("duration", 1) + + subject_raw = info.get("subject_code") or "?" + subject = subject_raw.upper() + + faculty = info.get("faculty_name") or "" + short_fac = (faculty.replace("Prof. ", "").replace("Dr. ", "") + .replace("Mr. ", "").replace("Ms. ", "")) + + # Look up elective_group from allocations + eg = alloc_by_sec_sub.get((sec_id.lower(), subject_raw.lower())) + is_oe = bool(eg and "oe" in eg.lower()) + + days_seen[ps].add(day) + for i in range(dur): - entry = (subject, short_fac) - if entry not in merged_grid[parent_sec][day][period + i]: - merged_grid[parent_sec][day][period + i].append(entry) - - st.markdown(""" - """, unsafe_allow_html=True) - - for p_sec in parent_sections: - st.markdown(f"### Section: {p_sec}") - html = '' - for h in const.TIMETABLE_HEADERS: - html += f'' - html += '' - - # Detect all days that have classes (including Saturday from extra classes) - days_in_solution = set() - for task_id, info in solution.items(): - sec = info.get('section_id','') - if sec.split('-')[0].upper() == p_sec or sec.upper() == p_sec: - days_in_solution.add(info['day_index']) - - # Always show Mon-Fri; add Saturday only if it has classes - all_day_indices = list(range(len(const.DAYS))) - sat_index = 5 # Saturday index - if sat_index in days_in_solution and sat_index not in all_day_indices: - all_day_indices.append(sat_index) - - # Day name lookup including Saturday - all_day_names = list(const.DAYS) + (['SAT'] if len(const.DAYS) <= 5 else []) - - total_days = len(all_day_indices) - - for row_num, day_idx in enumerate(all_day_indices): - day_name = all_day_names[day_idx] if day_idx < len(all_day_names) else f'Day{day_idx}' - html += f'' - period_counter = 0 - for header_text in const.TIMETABLE_HEADERS: - is_break = (header_text == "10:35-10:50") - is_lunch = (header_text == "12:40-1:40") - if is_break: - if row_num == 0: - html += f'' - elif is_lunch: - if row_num == 0: - html += f'' - else: - entries = merged_grid[p_sec][day_idx].get(period_counter, []) - if entries: - cell = '' - for idx, (subj, fac) in enumerate(entries): - div_cls = 'multi-subj' if idx > 0 else '' - cell += (f'
' - f'
{subj}
' - f'
{fac}
' - f'
') - html += f'' - else: - html += '' - period_counter += 1 - html += '' - html += '
Day \\ Time{h}
{day_name}Tea BreakLunch Break{cell}

' - st.markdown(html, unsafe_allow_html=True) - -# ── Sidebar ─────────────────────────────────────────────────────────────── -st.sidebar.title("🎓 VTU Timetable System") - -# Show schedule status in sidebar -if schedule_exists(): - sched = load_schedule() - gen_at = sched.get('generated_at', '')[:16].replace('T', ' ') - st.sidebar.success(f"📅 Schedule active\nGenerated: {gen_at}") -else: - st.sidebar.warning("No schedule generated yet") - -page = st.sidebar.radio("Navigate", [ - "🗓️ Generate Timetable", - "✏️ Update Timetable", - "📊 Original vs Current", - "📋 Change History", - "👥 Manage Faculties", - "📚 Manage Subjects", - "🏛️ Manage Sections", - "🚪 Manage Rooms", - "🔗 Manage Allocations", -]) - -data = load_data() - -# ═════════════════════════════════════════════════════════════════════════════ -# PAGE: GENERATE TIMETABLE -# ═════════════════════════════════════════════════════════════════════════════ -if page == "🗓️ Generate Timetable": - st.header("🗓️ Generate Semester Timetable") - if schedule_exists(): - st.warning("⚠️ A timetable is already active for this semester.") - col1, col2 = st.columns(2) - with col1: - if st.button("📄 View Current Timetable"): - st.session_state['show_current'] = True - with col2: - if st.button("🔄 Generate New Timetable (replaces current)", type="secondary"): - clear_schedule() - clear_history() - clear_original_schedule() - st.rerun() - - if st.session_state.get('show_current'): - sched = load_schedule() - if sched: - solution = sched['schedule'] - _, _, secs, _, _ = rebuild_objects(data) - render_timetable_html(solution, secs) - else: - st.info("Generate the timetable once — it will be fixed for the semester. " - "Use 'Update Timetable' to make changes later via prompts.") +# ═════════════════════════════════════════════════════════════════════════════ +# PYDANTIC SCHEMAS +# ═════════════════════════════════════════════════════════════════════════════ + +class FacultyIn(BaseModel): + id: str + name: str + designation: str + max_hours: int = 18 + +class SubjectIn(BaseModel): + code: str + name: str + credits: int + type: str = "THEORY" # THEORY | LAB | SOFTSKILL | FORUM + is_core: bool = True + is_heavy: bool = False + +class SectionIn(BaseModel): + id: str + semester: int + strength: int + +class RoomIn(BaseModel): + id: str + capacity: int + is_lab: bool = False + building: str = "Main" + +class AllocationIn(BaseModel): + faculty_id: str + subject_code: str + section_id: str + elective_group: Optional[str] = None + +class GenerateRequest(BaseModel): + time_limit_seconds: int = 30 + version_label: Optional[str] = None + semesters: Optional[List[int]] = None # e.g. [5, 7] for odd sems only + +class UpdateRequest(BaseModel): + prompt: str + preview_only: bool = False + propose_only: bool = False + +class OverwriteRequest(BaseModel): + schedule: dict + +class ProposeRequest(BaseModel): + schedule: dict + proposer: str + proposer_name: str + description: str = "Proposed timetable change" + +class InjectEntry(BaseModel): + section_id: str # e.g. "6A" (parent section) + day_index: int # 0-4 (Mon-Fri) + period_index: int # 0-7 teaching period + subject_code: str # e.g. "ml" + faculty_name: str # e.g. "Dr. Kavitha" + duration: int = 1 # 1 for theory, 2 for lab + room_id: Optional[str] = None + +class InjectRequest(BaseModel): + entries: list[InjectEntry] + +class RemoveRequest(BaseModel): + task_id: str - time_limit = st.slider("Solver time limit (seconds)", 10, 240, 120) - st.markdown("### 📝 Add Constraints Before Generating (Optional)") - st.caption("These rules will be baked into the timetable from the start.") +# ═════════════════════════════════════════════════════════════════════════════ +# HEALTH & CONSTANTS +# ═════════════════════════════════════════════════════════════════════════════ - # Constraint input area - if 'pre_constraints' not in st.session_state: - st.session_state['pre_constraints'] = [] +@app.get("/health") +def health(): + """API liveness check — also returns timetable constants for convenience.""" + return { + "status": "ok", + "schedule_exists": schedule_exists(), + **_timetable_constants(), + } - col1, col2 = st.columns([4, 1]) - with col1: - new_prompt = st.text_input( - "Type a constraint:", - placeholder="e.g. Prof. Anu is not available on Friday", - key="pre_constraint_input" - ) - with col2: - st.markdown("
", unsafe_allow_html=True) - if st.button("➕ Add") and new_prompt.strip(): - st.session_state['pre_constraints'].append(new_prompt.strip()) - st.rerun() - - # Show added constraints - if st.session_state['pre_constraints']: - st.markdown("**Constraints to apply:**") - for i, c in enumerate(st.session_state['pre_constraints']): - col1, col2 = st.columns([5, 1]) - col1.markdown(f"• {c}") - if col2.button("❌", key=f"del_{i}"): - st.session_state['pre_constraints'].pop(i) - st.rerun() - else: - st.info("No constraints added — timetable will be generated with default rules only.") +@app.get("/slm/health") +def slm_health(): + """Check whether the external flan-t5 SLM API is reachable.""" + ok = check_api_health() + return {"slm_online": ok, "message": "SLM API is online" if ok else "SLM API is offline"} + +@app.get("/constants") +def get_constants(): + """Timetable rendering constants (days, time headers, period indices).""" + return _timetable_constants() - st.divider() - if st.button("🚀 Generate Timetable", type="primary"): +# ═════════════════════════════════════════════════════════════════════════════ +# ALL DATA +# ═════════════════════════════════════════════════════════════════════════════ + +@app.get("/data") +def get_all_data(): + """Return all stored academic data in one call.""" + return load_data() + + + +import pandas as pd +import io + +@app.get("/data/template/excel") +def get_excel_template(): + from openpyxl.worksheet.datavalidation import DataValidation + + df_faculties = pd.DataFrame(columns=["id", "name", "designation", "max_hours"]) + df_subjects = pd.DataFrame(columns=["code", "name", "type", "credits", "is_core", "is_heavy"]) + df_sections = pd.DataFrame(columns=["id", "semester", "strength"]) + df_rooms = pd.DataFrame(columns=["id", "capacity", "is_lab", "building"]) + df_allocations = pd.DataFrame(columns=["faculty_id", "subject_code", "section_id", "elective_group"]) + df_scheduling_rules = pd.DataFrame(columns=["rule_type", "faculty_id", "subject_codes", "subject_types", "period", "max_period", "days"]) + + output = io.BytesIO() + with pd.ExcelWriter(output, engine='openpyxl') as writer: + df_faculties.to_excel(writer, sheet_name='Faculties', index=False) + df_subjects.to_excel(writer, sheet_name='Subjects', index=False) + df_sections.to_excel(writer, sheet_name='Sections', index=False) + df_rooms.to_excel(writer, sheet_name='Rooms', index=False) + df_allocations.to_excel(writer, sheet_name='Allocations', index=False) + df_scheduling_rules.to_excel(writer, sheet_name='Scheduling Rules', index=False) + + # Add dropdown validations to Scheduling Rules sheet + ws = writer.sheets['Scheduling Rules'] + + # Rule Type dropdown (column A, rows 2-100) + dv_rule_type = DataValidation(type="list", formula1='"FIXED_PERIOD,BEFORE_TIME,FIXED_DAYS,FACULTY_UNAVAILABLE"', allow_blank=True) + dv_rule_type.prompt = "Select a rule type" + dv_rule_type.promptTitle = "Rule Type" + ws.add_data_validation(dv_rule_type) + dv_rule_type.add(f'A2:A100') + + # Subject Types dropdown (column D, rows 2-100) + dv_subject_types = DataValidation(type="list", formula1='"THEORY,LAB,SOFTSKILL,FORUM"', allow_blank=True) + dv_subject_types.prompt = "Select subject type(s) - comma-separate for multiple" + dv_subject_types.promptTitle = "Subject Types" + ws.add_data_validation(dv_subject_types) + dv_subject_types.add(f'D2:D100') + + # Period dropdown (column E, rows 2-100) + dv_period = DataValidation(type="list", formula1='"Period 1,Period 2,Period 3,Period 4,Period 5,Period 6,Period 7,Period 8"', allow_blank=True) + dv_period.prompt = "Select period (for FIXED_PERIOD rules)" + dv_period.promptTitle = "Period" + ws.add_data_validation(dv_period) + dv_period.add(f'E2:E100') + + # Max Period dropdown (column F, rows 2-100) + dv_max_period = DataValidation(type="list", formula1='"Period 1,Period 2,Period 3,Period 4,Period 5,Period 6,Period 7,Period 8"', allow_blank=True) + dv_max_period.prompt = "Select max period (for BEFORE_TIME rules)" + dv_max_period.promptTitle = "Max Period" + ws.add_data_validation(dv_max_period) + dv_max_period.add(f'F2:F100') + + # Days dropdown (column G, rows 2-100) + dv_days = DataValidation(type="list", formula1='"MON,TUE,WED,THU,FRI,SAT"', allow_blank=True) + dv_days.prompt = "Select day(s) - comma-separate for multiple (for FIXED_DAYS rules)" + dv_days.promptTitle = "Days" + ws.add_data_validation(dv_days) + dv_days.add(f'G2:G100') + + # Designation dropdown for Faculties sheet (column C, rows 2-100) + ws_fac = writer.sheets['Faculties'] + dv_designation = DataValidation(type="list", formula1='"Professor,Assoc. Prof,Asst. Prof,Guest"', allow_blank=True) + dv_designation.prompt = "Select designation" + dv_designation.promptTitle = "Designation" + ws_fac.add_data_validation(dv_designation) + dv_designation.add(f'C2:C100') + + # Subject Type dropdown for Subjects sheet (column C, rows 2-100) + ws_sub = writer.sheets['Subjects'] + dv_sub_type = DataValidation(type="list", formula1='"THEORY,LAB,SOFTSKILL,FORUM"', allow_blank=True) + dv_sub_type.prompt = "Select subject type" + dv_sub_type.promptTitle = "Type" + ws_sub.add_data_validation(dv_sub_type) + dv_sub_type.add(f'C2:C100') + + # Boolean dropdowns for Subjects (is_core col E, is_heavy col F) + dv_bool = DataValidation(type="list", formula1='"TRUE,FALSE"', allow_blank=True) + ws_sub.add_data_validation(dv_bool) + dv_bool.add(f'E2:F100') + + # Boolean dropdown for Rooms is_lab (column C, rows 2-100) + ws_rooms = writer.sheets['Rooms'] + dv_lab = DataValidation(type="list", formula1='"TRUE,FALSE"', allow_blank=True) + ws_rooms.add_data_validation(dv_lab) + dv_lab.add(f'C2:C100') + + output.seek(0) + return StreamingResponse( + output, + media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", + headers={"Content-Disposition": "attachment; filename=timetable_template.xlsx"} + ) + + +@app.get("/data/export/excel") +def export_data_as_excel(): + """Export all current data as an Excel file matching the template format.""" + data = load_data() + + # Build DataFrames from existing data + df_faculties = pd.DataFrame(data.get("faculties", [])) + df_subjects = pd.DataFrame(data.get("subjects", [])) + df_sections = pd.DataFrame(data.get("sections", [])) + df_rooms = pd.DataFrame(data.get("rooms", [])) + df_allocations = pd.DataFrame(data.get("allocations", [])) + + # Build scheduling rules DataFrame with human-readable columns + rules = data.get("scheduling_rules", []) + rules_rows = [] + for rule in rules: + row = { + "rule_type": rule.get("rule_type", ""), + "faculty_id": rule.get("faculty_id", ""), + "subject_codes": ", ".join(rule.get("subject_codes", [])) if isinstance(rule.get("subject_codes"), list) else str(rule.get("subject_codes", "")), + "subject_types": ", ".join(rule.get("subject_types", [])) if isinstance(rule.get("subject_types"), list) else str(rule.get("subject_types", "")), + "period": f"Period {rule['period_index'] + 1}" if rule.get("period_index") is not None else "", + "max_period": f"Period {rule['max_period_index'] + 1}" if rule.get("max_period_index") is not None else "", + "days": ", ".join(rule.get("days", [])) if isinstance(rule.get("days"), list) else str(rule.get("days", "")), + } + rules_rows.append(row) + df_scheduling_rules = pd.DataFrame(rules_rows) if rules_rows else pd.DataFrame(columns=["rule_type", "subject_codes", "subject_types", "period", "max_period", "days"]) + + # Ensure column order matches the template + fac_cols = ["id", "name", "designation", "max_hours"] + sub_cols = ["code", "name", "type", "credits", "is_core", "is_heavy"] + sec_cols = ["id", "semester", "strength"] + room_cols = ["id", "capacity", "is_lab", "building"] + alloc_cols = ["faculty_id", "subject_code", "section_id", "elective_group"] + rule_cols = ["rule_type", "faculty_id", "subject_codes", "subject_types", "period", "max_period", "days"] + + for col in fac_cols: + if col not in df_faculties.columns: + df_faculties[col] = "" + for col in sub_cols: + if col not in df_subjects.columns: + df_subjects[col] = "" + for col in sec_cols: + if col not in df_sections.columns: + df_sections[col] = "" + for col in room_cols: + if col not in df_rooms.columns: + df_rooms[col] = "" + for col in alloc_cols: + if col not in df_allocations.columns: + df_allocations[col] = "" + + output = io.BytesIO() + with pd.ExcelWriter(output, engine='openpyxl') as writer: + df_faculties[fac_cols].to_excel(writer, sheet_name='Faculties', index=False) + df_subjects[sub_cols].to_excel(writer, sheet_name='Subjects', index=False) + df_sections[sec_cols].to_excel(writer, sheet_name='Sections', index=False) + df_rooms[room_cols].to_excel(writer, sheet_name='Rooms', index=False) + df_allocations[alloc_cols].to_excel(writer, sheet_name='Allocations', index=False) + df_scheduling_rules[rule_cols].to_excel(writer, sheet_name='Scheduling Rules', index=False) + + # Add dropdown validations (same as template) + from openpyxl.worksheet.datavalidation import DataValidation + + # Faculties - designation dropdown + ws_fac = writer.sheets['Faculties'] + dv_desig = DataValidation(type="list", formula1='"Professor,Assoc. Prof,Asst. Prof,Guest"', allow_blank=True) + ws_fac.add_data_validation(dv_desig) + dv_desig.add('C2:C1000') + + # Subjects - type dropdown + ws_sub = writer.sheets['Subjects'] + dv_stype = DataValidation(type="list", formula1='"THEORY,LAB,SOFTSKILL,FORUM"', allow_blank=True) + ws_sub.add_data_validation(dv_stype) + dv_stype.add('C2:C1000') + + # Subjects - is_core & is_heavy boolean dropdowns + dv_bool_sub = DataValidation(type="list", formula1='"TRUE,FALSE"', allow_blank=True) + ws_sub.add_data_validation(dv_bool_sub) + dv_bool_sub.add('E2:F1000') + + # Rooms - is_lab boolean dropdown + ws_rooms = writer.sheets['Rooms'] + dv_lab = DataValidation(type="list", formula1='"TRUE,FALSE"', allow_blank=True) + ws_rooms.add_data_validation(dv_lab) + dv_lab.add('C2:C1000') + + # Scheduling Rules - all dropdowns + ws_rules = writer.sheets['Scheduling Rules'] + + dv_rt = DataValidation(type="list", formula1='"FIXED_PERIOD,BEFORE_TIME,FIXED_DAYS"', allow_blank=True) + ws_rules.add_data_validation(dv_rt) + dv_rt.add('A2:A1000') + + dv_st = DataValidation(type="list", formula1='"THEORY,LAB,SOFTSKILL,FORUM"', allow_blank=True) + ws_rules.add_data_validation(dv_st) + dv_st.add('C2:C1000') + + dv_p = DataValidation(type="list", formula1='"Period 1,Period 2,Period 3,Period 4,Period 5,Period 6,Period 7,Period 8"', allow_blank=True) + ws_rules.add_data_validation(dv_p) + dv_p.add('D2:D1000') + + dv_mp = DataValidation(type="list", formula1='"Period 1,Period 2,Period 3,Period 4,Period 5,Period 6,Period 7,Period 8"', allow_blank=True) + ws_rules.add_data_validation(dv_mp) + dv_mp.add('E2:E1000') + + dv_d = DataValidation(type="list", formula1='"MON,TUE,WED,THU,FRI,SAT"', allow_blank=True) + ws_rules.add_data_validation(dv_d) + dv_d.add('F2:F1000') + + output.seek(0) + return StreamingResponse( + output, + media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", + headers={"Content-Disposition": "attachment; filename=timetable_data_export.xlsx"} + ) + +@app.post("/data/import/excel") +async def import_excel(file: UploadFile = File(...)): + contents = await file.read() + try: + xls = pd.ExcelFile(io.BytesIO(contents)) + data = { + "faculties": [], + "subjects": [], + "sections": [], + "rooms": [], + "allocations": [], + "scheduling_rules": [] + } + + import re + def parse_int(val, default): + if pd.isna(val) or val == '': return default try: - facs, subs, secs, rooms, allocs = convert_json_to_objects(data) - tasks = prepare_scheduling_tasks(allocs, facs, subs, secs) - st.write(f"Scheduling {len(tasks)} tasks...") - - # Convert pre-constraints via SLM - pre_slm_constraints = [] - if st.session_state.get('pre_constraints'): - with st.spinner("Converting constraints via SLM API..."): - from slm_inference import get_constraints_batch - pre_slm_constraints = get_constraints_batch( - st.session_state['pre_constraints']) - st.write(f"✅ {len(pre_slm_constraints)} constraint(s) parsed") - - solver = TimetableSolver(tasks, facs, secs, rooms) - with st.spinner("Optimizing schedule..."): - status, solution = solver.solve( - time_limit_seconds=time_limit, - enable_soft_constraints=True, - slm_constraints=pre_slm_constraints) - - if status in ("OPTIMAL", "FEASIBLE"): - save_schedule(solution) - save_original_schedule(solution) # permanent snapshot - st.success(f"✅ Timetable Generated! Status: {status}") - st.balloons() - render_timetable_html(solution, secs) - else: - st.error(f"❌ Solver failed: {status}") - except Exception as e: - st.error(f"Error: {e}") - import traceback; st.code(traceback.format_exc()) - -# ═════════════════════════════════════════════════════════════════════════════ -# PAGE: UPDATE TIMETABLE -# ═════════════════════════════════════════════════════════════════════════════ -elif page == "✏️ Update Timetable": - st.header("✏️ Update Timetable with Natural Language") + return int(val) + except (ValueError, TypeError): + m = re.search(r'\d+', str(val)) + return int(m.group()) if m else default + + if 'Faculties' in xls.sheet_names: + df = pd.read_excel(xls, 'Faculties').fillna('') + for _, row in df.iterrows(): + if row.get('id'): + data["faculties"].append({ + "id": str(row.get('id')), + "name": str(row.get('name', '')), + "designation": str(row.get('designation', 'Asst. Prof')), + "max_hours": parse_int(row.get('max_hours', 18), 18) + }) + + if 'Subjects' in xls.sheet_names: + df = pd.read_excel(xls, 'Subjects').fillna('') + for _, row in df.iterrows(): + if row.get('code'): + data["subjects"].append({ + "code": str(row.get('code')), + "name": str(row.get('name', '')), + "type": str(row.get('type', 'THEORY')), + "credits": parse_int(row.get('credits', 3), 3), + "is_core": bool(row.get('is_core', True)), + "is_heavy": bool(row.get('is_heavy', False)) + }) + + if 'Sections' in xls.sheet_names: + df = pd.read_excel(xls, 'Sections').fillna('') + for _, row in df.iterrows(): + if row.get('id'): + data["sections"].append({ + "id": str(row.get('id')), + "semester": parse_int(row.get('semester', 1), 1), + "strength": parse_int(row.get('strength', 60), 60) + }) + + if 'Rooms' in xls.sheet_names: + df = pd.read_excel(xls, 'Rooms').fillna('') + for _, row in df.iterrows(): + if row.get('id'): + data["rooms"].append({ + "id": str(row.get('id')), + "capacity": parse_int(row.get('capacity', 60), 60), + "is_lab": bool(row.get('is_lab', False)), + "building": str(row.get('building', 'Main')) + }) + + if 'Allocations' in xls.sheet_names: + df = pd.read_excel(xls, 'Allocations').fillna('') + for _, row in df.iterrows(): + if row.get('faculty_id') and row.get('subject_code') and row.get('section_id'): + eg = row.get('elective_group') + data["allocations"].append({ + "faculty_id": str(row.get('faculty_id')), + "subject_code": str(row.get('subject_code')), + "section_id": str(row.get('section_id')), + "elective_group": str(eg) if eg else None + }) + + # Parse Scheduling Rules sheet if present, otherwise preserve existing rules + if 'Scheduling Rules' in xls.sheet_names: + import uuid as _uuid + df = pd.read_excel(xls, 'Scheduling Rules').fillna('') + for _, row in df.iterrows(): + rule_type = str(row.get('rule_type', '')).strip() + if not rule_type: + continue + rule = {"id": str(_uuid.uuid4()), "rule_type": rule_type} + + # Parse faculty_id + fid = str(row.get('faculty_id', '')).strip() + if fid: + rule["faculty_id"] = fid + + # Parse subject_codes (comma-separated) + sc = str(row.get('subject_codes', '')).strip() + rule["subject_codes"] = [s.strip() for s in sc.split(',') if s.strip()] if sc else [] + + # Parse subject_types (comma-separated) + st = str(row.get('subject_types', '')).strip() + rule["subject_types"] = [s.strip() for s in st.split(',') if s.strip()] if st else [] + + # Parse period (e.g. "Period 3" -> 2) + period_str = str(row.get('period', '')).strip() + if period_str and 'Period' in period_str: + try: + rule["period_index"] = int(period_str.replace('Period ', '')) - 1 + except ValueError: + pass + + # Parse max_period + max_period_str = str(row.get('max_period', '')).strip() + if max_period_str and 'Period' in max_period_str: + try: + rule["max_period_index"] = int(max_period_str.replace('Period ', '')) - 1 + except ValueError: + pass + + # Parse days (comma-separated) + days_str = str(row.get('days', '')).strip() + if days_str: + rule["days"] = [d.strip() for d in days_str.split(',') if d.strip()] + + data["scheduling_rules"].append(rule) + else: + # Preserve existing scheduling_rules if no sheet in upload + existing_data = load_data() + data["scheduling_rules"] = existing_data.get("scheduling_rules", []) + + save_data(data) + return {"message": "Data imported successfully", "imported": {k: len(v) for k, v in data.items()}} + + except Exception as e: + raise HTTPException(400, f"Failed to parse Excel file: {str(e)}") + +# ═════════════════════════════════════════════════════════════════════════════ +# MANAGE FACULTIES +# ═════════════════════════════════════════════════════════════════════════════ + +@app.get("/data/faculties") +def list_faculties(): + return {"faculties": load_data()["faculties"]} + +@app.post("/data/faculties", status_code=201) +def add_faculty(faculty: FacultyIn): + data = load_data() + if any(f["id"] == faculty.id for f in data["faculties"]): + raise HTTPException(400, f"Faculty ID '{faculty.id}' already exists.") + data["faculties"].append(faculty.model_dump()) + save_data(data) + return {"message": "Faculty added.", "faculty": faculty.model_dump()} + + +@app.put("/data/faculties/{faculty_id}") +def update_faculty(faculty_id: str, faculty: FacultyIn): + data = load_data() + for i, f in enumerate(data["faculties"]): + if f["id"] == faculty_id: + updated = faculty.model_dump() + data["faculties"][i] = updated + # Cascade to allocations and scheduling_rules if ID changed + if faculty_id != updated["id"]: + for a in data.get("allocations", []): + if a.get("faculty_id") == faculty_id: + a["faculty_id"] = updated["id"] + for r in data.get("scheduling_rules", []): + if r.get("faculty_id") == faculty_id: + r["faculty_id"] = updated["id"] + save_data(data) + return {"message": "Faculty updated.", "faculty": updated} + raise HTTPException(404, f"Faculty '{faculty_id}' not found.") + +@app.delete("/data/faculties") +def clear_faculties(): + data = load_data() + data["faculties"] = [] + save_data(data) + return {"message": "All faculties cleared."} + +@app.delete("/data/faculties/{faculty_id}") +def delete_faculty(faculty_id: str): + data = load_data() + before = len(data["faculties"]) + data["faculties"] = [f for f in data["faculties"] if f["id"] != faculty_id] + if len(data["faculties"]) == before: + raise HTTPException(404, f"Faculty '{faculty_id}' not found.") + save_data(data) + return {"message": f"Faculty '{faculty_id}' deleted."} + + +# ═════════════════════════════════════════════════════════════════════════════ +# MANAGE SUBJECTS +# ═════════════════════════════════════════════════════════════════════════════ + +@app.get("/data/subjects") +def list_subjects(): + return {"subjects": load_data()["subjects"]} + +@app.post("/data/subjects", status_code=201) +def add_subject(subject: SubjectIn): + data = load_data() + if any(s["code"] == subject.code for s in data["subjects"]): + raise HTTPException(400, f"Subject code '{subject.code}' already exists.") + data["subjects"].append(subject.model_dump()) + save_data(data) + return {"message": "Subject added.", "subject": subject.model_dump()} + + +@app.put("/data/subjects/{subject_code}") +def update_subject(subject_code: str, subject: SubjectIn): + data = load_data() + for i, s in enumerate(data["subjects"]): + if s["code"] == subject_code: + updated = subject.model_dump() + data["subjects"][i] = updated + # Cascade to allocations and scheduling_rules if code changed + if subject_code != updated["code"]: + for a in data.get("allocations", []): + if a.get("subject_code") == subject_code: + a["subject_code"] = updated["code"] + for r in data.get("scheduling_rules", []): + if subject_code in r.get("subject_codes", []): + r["subject_codes"] = [updated["code"] if c == subject_code else c for c in r["subject_codes"]] + save_data(data) + return {"message": "Subject updated.", "subject": updated} + raise HTTPException(404, f"Subject '{subject_code}' not found.") + +@app.delete("/data/subjects") +def clear_subjects(): + data = load_data() + data["subjects"] = [] + save_data(data) + return {"message": "All subjects cleared."} + +@app.delete("/data/subjects/{subject_code}") +def delete_subject(subject_code: str): + data = load_data() + before = len(data["subjects"]) + data["subjects"] = [s for s in data["subjects"] if s["code"] != subject_code] + if len(data["subjects"]) == before: + raise HTTPException(404, f"Subject '{subject_code}' not found.") + save_data(data) + return {"message": f"Subject '{subject_code}' deleted."} + + +# ═════════════════════════════════════════════════════════════════════════════ +# MANAGE SECTIONS +# ═════════════════════════════════════════════════════════════════════════════ + +@app.get("/data/sections") +def list_sections(): + return {"sections": load_data()["sections"]} + +@app.post("/data/sections", status_code=201) +def add_section(section: SectionIn): + data = load_data() + if any(s["id"] == section.id for s in data["sections"]): + raise HTTPException(400, f"Section ID '{section.id}' already exists.") + data["sections"].append(section.model_dump()) + save_data(data) + return {"message": "Section added.", "section": section.model_dump()} + + +@app.put("/data/sections/{section_id}") +def update_section(section_id: str, section: SectionIn): + data = load_data() + for i, s in enumerate(data["sections"]): + if s["id"] == section_id: + updated = section.model_dump() + data["sections"][i] = updated + # Cascade to allocations if ID changed + if section_id != updated["id"]: + for a in data.get("allocations", []): + if a.get("section_id") == section_id: + a["section_id"] = updated["id"] + save_data(data) + return {"message": "Section updated.", "section": updated} + raise HTTPException(404, f"Section '{section_id}' not found.") + +@app.delete("/data/sections") +def clear_sections(): + data = load_data() + data["sections"] = [] + save_data(data) + return {"message": "All sections cleared."} + +@app.get("/data/semesters") +def list_semesters(): + """Return all unique semester numbers from sections data.""" + data = load_data() + semesters = sorted(set(s["semester"] for s in data.get("sections", []))) + return {"semesters": semesters} + + +@app.delete("/data/sections/{section_id}") +def delete_section(section_id: str): + data = load_data() + before = len(data["sections"]) + data["sections"] = [s for s in data["sections"] if s["id"] != section_id] + if len(data["sections"]) == before: + raise HTTPException(404, f"Section '{section_id}' not found.") + save_data(data) + return {"message": f"Section '{section_id}' deleted."} + + +# ═════════════════════════════════════════════════════════════════════════════ +# MANAGE ROOMS +# ═════════════════════════════════════════════════════════════════════════════ + +@app.get("/data/rooms") +def list_rooms(): + return {"rooms": load_data()["rooms"]} + +@app.post("/data/rooms", status_code=201) +def add_room(room: RoomIn): + data = load_data() + if any(r["id"] == room.id for r in data["rooms"]): + raise HTTPException(400, f"Room ID '{room.id}' already exists.") + data["rooms"].append(room.model_dump()) + save_data(data) + return {"message": "Room added.", "room": room.model_dump()} + + +@app.put("/data/rooms/{room_id}") +def update_room(room_id: str, room: RoomIn): + data = load_data() + for i, r in enumerate(data["rooms"]): + if r["id"] == room_id: + updated = room.model_dump() + data["rooms"][i] = updated + save_data(data) + return {"message": "Room updated.", "room": updated} + raise HTTPException(404, f"Room '{room_id}' not found.") + +@app.delete("/data/rooms") +def clear_rooms(): + data = load_data() + data["rooms"] = [] + save_data(data) + return {"message": "All rooms cleared."} + +@app.delete("/data/rooms/{room_id}") +def delete_room(room_id: str): + data = load_data() + before = len(data["rooms"]) + data["rooms"] = [r for r in data["rooms"] if r["id"] != room_id] + if len(data["rooms"]) == before: + raise HTTPException(404, f"Room '{room_id}' not found.") + save_data(data) + return {"message": f"Room '{room_id}' deleted."} + + +# ═════════════════════════════════════════════════════════════════════════════ +# MANAGE ALLOCATIONS +# ═════════════════════════════════════════════════════════════════════════════ + +@app.get("/data/allocations") +def list_allocations(): + return {"allocations": load_data()["allocations"]} + +@app.post("/data/allocations", status_code=201) +def add_allocation(alloc: AllocationIn): + data = load_data() + data["allocations"].append(alloc.model_dump()) + save_data(data) + return {"message": "Allocation added.", "allocation": alloc.model_dump()} + + +@app.put("/data/allocations/{idx}") +def update_allocation(idx: int, alloc: AllocationIn): + data = load_data() + if idx < 0 or idx >= len(data["allocations"]): + raise HTTPException(404, f"Allocation index {idx} out of range.") + data["allocations"][idx] = alloc.model_dump() + save_data(data) + return {"message": "Allocation updated.", "allocation": alloc.model_dump()} + +@app.delete("/data/allocations") +def clear_allocations(): + data = load_data() + data["allocations"] = [] + save_data(data) + return {"message": "All allocations cleared."} + +@app.delete("/data/allocations/{idx}") +def delete_allocation(idx: int): + """Delete allocation by its 0-based index in the list.""" + data = load_data() + if idx < 0 or idx >= len(data["allocations"]): + raise HTTPException(404, f"Allocation index {idx} out of range.") + removed = data["allocations"].pop(idx) + save_data(data) + return {"message": "Allocation deleted.", "removed": removed} + + +# ═════════════════════════════════════════════════════════════════════════════ +# SCHEDULING RULES +# ═════════════════════════════════════════════════════════════════════════════ + +@app.get("/scheduling-rules") +def get_scheduling_rules(): + data = load_data() + return {"rules": data.get("scheduling_rules", [])} + +@app.post("/scheduling-rules") +def add_scheduling_rule(rule: dict): + import uuid as _uuid + data = load_data() + rules = data.get("scheduling_rules", []) + rule["id"] = str(_uuid.uuid4()) + rules.append(rule) + data["scheduling_rules"] = rules + save_data(data) + return {"message": "Rule added.", "rule": rule} + + +@app.put("/scheduling-rules/{rule_id}") +def update_scheduling_rule(rule_id: str, rule: dict): + data = load_data() + rules = data.get("scheduling_rules", []) + for i, r in enumerate(rules): + if r.get("id") == rule_id: + rule["id"] = rule_id + rules[i] = rule + data["scheduling_rules"] = rules + save_data(data) + return {"message": "Rule updated.", "rule": rule} + raise HTTPException(404, "Rule not found.") + +@app.delete("/scheduling-rules/{rule_id}") +def delete_scheduling_rule(rule_id: str): + data = load_data() + rules = data.get("scheduling_rules", []) + data["scheduling_rules"] = [r for r in rules if r.get("id") != rule_id] + save_data(data) + return {"message": "Rule deleted."} + + +# ═════════════════════════════════════════════════════════════════════════════ +# GENERATE TIMETABLE +# ═════════════════════════════════════════════════════════════════════════════ + +@app.post("/generate") +def generate(req: GenerateRequest): + try: + data = load_data() + + missing = [k for k in ("faculties", "subjects", "sections", "rooms", "allocations") + if not data.get(k)] + if missing: + raise HTTPException(400, f"Missing data for: {', '.join(missing)}.") + + # ── Semester filtering ──────────────────────────────────────────── + if req.semesters: + selected_sems = set(req.semesters) + # 1. Filter sections to only selected semesters + data["sections"] = [ + s for s in data["sections"] if s["semester"] in selected_sems + ] + if not data["sections"]: + raise HTTPException(400, f"No sections found for semesters: {req.semesters}") + + # 2. Filter allocations to only reference surviving sections + valid_section_ids = {s["id"] for s in data["sections"]} + data["allocations"] = [ + a for a in data["allocations"] + if a["section_id"] in valid_section_ids + ] + + # 3. Filter subjects to only those referenced by surviving allocations + used_subject_codes = {a["subject_code"] for a in data["allocations"]} + data["subjects"] = [ + s for s in data["subjects"] if s["code"] in used_subject_codes + ] + + # 4. Filter faculties to only those referenced by surviving allocations + used_faculty_ids = {a["faculty_id"] for a in data["allocations"]} + data["faculties"] = [ + f for f in data["faculties"] if f["id"] in used_faculty_ids + ] + + print(f"[generate] Filtered to semesters {req.semesters}: " + f"{len(data['sections'])} sections, {len(data['allocations'])} allocations, " + f"{len(data['subjects'])} subjects, {len(data['faculties'])} faculties") + + facs, subs, secs, rooms, allocs = _build_objects(data) + tasks = prepare_scheduling_tasks(allocs, facs, subs, secs) + + if not tasks: + raise HTTPException(400, "No schedulable tasks found. Check your allocations.") + + solver = TimetableSolver(tasks, facs, secs, rooms) + try: + status, solution = solver.solve( + time_limit_seconds=req.time_limit_seconds, + enable_soft_constraints=True, + scheduling_rules=data.get("scheduling_rules", []), + ) + except ValueError as e: + raise HTTPException(status_code=400, detail=str(e)) + + if status not in ("OPTIMAL", "FEASIBLE"): + diagnosis = solver.diagnose_infeasibility() + raise HTTPException(400, f"Timetable generation failed. {diagnosis}") + # Auto-save current schedule as a version before overwriting + if schedule_exists(): + save_version(label=req.version_label) + + save_schedule(solution) + save_original_schedule(solution) + clean = _clean(solution) + + return { + "status": status, + "task_count": len(tasks), + "semesters_generated": req.semesters or "all", + "schedule": clean, + "grid": _build_grid(clean), + **_timetable_constants(), + } + + except HTTPException: + raise + except Exception: + raise HTTPException(500, traceback.format_exc()) + + +# ═════════════════════════════════════════════════════════════════════════════ +# GET / DELETE TIMETABLE +# ═════════════════════════════════════════════════════════════════════════════ + +@app.get("/schedule") +def get_schedule(): if not schedule_exists(): - st.error("❌ No timetable generated yet. Go to 'Generate Timetable' first.") - st.stop() - - # API Health Check - with st.spinner("Checking SLM API..."): - api_ok = check_api_health() - if api_ok: - st.success("✅ SLM API is online") - else: - st.error("❌ SLM API is offline. Check HuggingFace Space.") - st.stop() + return {"exists": False, "schedule": None, "grid": None} + data = load_schedule() + clean = _clean(data["schedule"]) + return { + "exists": True, + "generated_at": data.get("generated_at"), + "schedule": clean, + "grid": _build_grid(clean), + **_timetable_constants(), + } + +@app.get("/schedule/original") +def get_original_schedule(): + if not original_schedule_exists(): + return {"exists": False, "grid": None} + data = load_original_schedule() + clean = _clean(data["schedule"]) + return { + "exists": True, + "generated_at": data.get("generated_at"), + "grid": _build_grid(clean), + **_timetable_constants(), + } - st.markdown(""" - **How it works:** Type a natural language instruction. Only the affected - slots will be rescheduled — the rest of the timetable stays unchanged. +@app.delete("/schedule") +def delete_schedule(version_label: Optional[str] = None): + # Auto-save as version before clearing + if schedule_exists(): + save_version(label=version_label) + clear_schedule() + clear_original_schedule() + clear_history() + return {"message": "Schedule saved as version and cleared."} + +@app.post("/schedule/revert") +def revert_schedule(): + if not original_schedule_exists(): + raise HTTPException(400, "No original schedule found to revert to.") + data = load_original_schedule() + save_schedule(data["schedule"]) + clear_history() + clean = _clean(data["schedule"]) + return { + "status": "SUCCESS", + "message": "Reverted to original schedule and cleared history.", + "schedule": clean, + "grid": _build_grid(clean), + **_timetable_constants(), + } + +# ═════════════════════════════════════════════════════════════════════════════ +# SCHEDULE VERSIONS +# ═════════════════════════════════════════════════════════════════════════════ + +@app.get("/schedule/versions") +def get_versions(): + versions = load_versions() + # Return summary only (not full schedule data) for the list view + return { + "versions": [ + { + "id": v["id"], + "label": v.get("label", f"Version {i+1}"), + "timestamp": v.get("timestamp", ""), + "generated_at": v.get("generated_at", ""), + "history_count": len(v.get("history", [])), + } + for i, v in enumerate(versions) + ] + } + +@app.get("/schedule/versions/{version_id}") +def get_version_details(version_id: str): + versions = load_versions() + target = next((v for v in versions if v["id"] == version_id), None) + if not target: + raise HTTPException(404, "Version not found.") + clean = _clean(target.get("schedule", {})) + return { + "status": "SUCCESS", + "version_id": version_id, + "label": target.get("label"), + "schedule": clean, + "grid": _build_grid(clean), + **_timetable_constants(), + } + +@app.post("/schedule/versions/restore/{version_id}") +def restore_version_endpoint(version_id: str): + # Auto-save current as a version before restoring + if schedule_exists(): + save_version() + result = restore_version(version_id) + if not result: + raise HTTPException(404, "Version not found.") + clean = _clean(result["schedule"]) + return { + "status": "SUCCESS", + "message": f"Restored version: {result.get('label', version_id)}", + "schedule": clean, + "grid": _build_grid(clean), + **_timetable_constants(), + } + +@app.post("/schedule/overwrite") +def overwrite_schedule(req: OverwriteRequest): + if not req.schedule: + raise HTTPException(400, "Schedule payload cannot be empty.") + + old_raw = load_schedule() + old_schedule = _clean(old_raw.get("schedule", {})) if old_raw else {} + clean_new = _clean(req.schedule) + + changes, affected = diff_schedules(old_schedule, clean_new) + + save_schedule(clean_new) + if changes: + add_history_entry( + operation_type="MANUAL_OVERWRITE", + description=f"Manual drag & drop modifications ({len(changes)} cells affected)", + affected_sections=affected, + changes=changes, + status="SUCCESS", + constraints=[] + ) + return { + "status": "SUCCESS", + "message": "Schedule overwritten manually.", + "schedule": clean_new, + "grid": _build_grid(clean_new), + **_timetable_constants(), + } + + +# ═════════════════════════════════════════════════════════════════════════════ +# SCHEDULE PROPOSALS (SUPER TEACHER WORKFLOW) +# ═════════════════════════════════════════════════════════════════════════════ + +@app.get("/schedule/proposals") +def get_proposals(): + from storage import load_proposals + return {"proposals": load_proposals()} + +@app.post("/schedule/propose") +def propose_schedule(req: ProposeRequest): + from storage import save_proposal + import uuid as _uuid + + old_raw = load_schedule() + old_schedule = _clean(old_raw.get("schedule", {})) if old_raw else {} + clean_new = _clean(req.schedule) + + changes, affected = diff_schedules(old_schedule, clean_new) + + proposal = { + "id": str(_uuid.uuid4()), + "proposer": req.proposer, + "proposer_name": req.proposer_name, + "description": req.description, + "timestamp": datetime.now().isoformat(), + "changes_count": len(changes), + "changes": changes, + "schedule": clean_new, + "status": "PENDING" + } + save_proposal(proposal) + return {"status": "SUCCESS", "message": "Proposal submitted.", "proposal": proposal} + +@app.post("/schedule/proposals/{proposal_id}/approve") +def approve_proposal(proposal_id: str): + from storage import load_proposals, delete_proposal + proposals = load_proposals() + target = next((p for p in proposals if p.get("id") == proposal_id), None) + if not target: + raise HTTPException(404, "Proposal not found.") - **Examples:** - - `Prof. Anu is not available on Friday` - - `NLP Lab must be in consecutive slots` - - `ML should be scheduled in the morning` - - `Slot 5 is the lunch break` - - `Limit Sanjay to 3 hours per day` - - `Prof. Kavitha should have Wednesday free` - """) - - st.divider() - - prompt = st.text_input("💬 Enter your instruction:", - placeholder="e.g. Prof. Anu is not available on Friday") - - col1, col2 = st.columns([1, 3]) - with col1: - apply = st.button("✅ Apply Change", type="primary", - disabled=not prompt.strip()) - with col2: - preview = st.button("👁️ Preview Constraint Only", - disabled=not prompt.strip()) - - # Preview mode — just show the constraint without applying - if preview and prompt.strip(): - from slm_inference import smart_parse - local = smart_parse(prompt, data['faculties'], - data.get('subjects',[]), data.get('sections',[])) - if local: - st.subheader("Constraint that would be applied (local parser):") - st.json(local) + clean_new = target["schedule"] + save_schedule(clean_new) + if target.get("changes"): + add_history_entry( + operation_type="PROPOSAL_APPROVED", + description=f"Approved changes by {target.get('proposer_name')} ({len(target['changes'])} cells affected)", + affected_sections=[], + changes=target["changes"], + status="SUCCESS", + constraints=[] + ) + + delete_proposal(proposal_id) + return {"status": "SUCCESS", "message": "Proposal approved and applied."} + +@app.delete("/schedule/proposals/{proposal_id}") +def reject_proposal(proposal_id: str): + from storage import delete_proposal + delete_proposal(proposal_id) + return {"status": "SUCCESS", "message": "Proposal rejected."} + + +# ═════════════════════════════════════════════════════════════════════════════ +# INJECT SUBJECT (ADD TO TIMETABLE) +# ═════════════════════════════════════════════════════════════════════════════ + +@app.post("/schedule/inject") +def inject_subject(req: InjectRequest): + """Insert one or more new class entries into the current schedule.""" + if not schedule_exists(): + raise HTTPException(400, "No timetable generated yet. Call POST /generate first.") + + if not req.entries: + raise HTTPException(400, "No entries provided.") + + data = load_data() + sched_data = load_schedule() + current = sched_data["schedule"] + old_schedule = _clean(dict(current)) # snapshot before mutation + + # Build a set of existing task IDs to avoid collisions + existing_ids = set(current.keys()) + + # Build a lookup of faculty_name -> faculty for room assignment + rooms = data.get("rooms", []) + subjects_lookup = {s["code"].lower(): s for s in data.get("subjects", [])} + + injected_entries = [] + + for entry in req.entries: + # Check for section collisions first + for t_id, info in current.items(): + if info.get("section_id") == entry.section_id and info.get("day_index") == entry.day_index: + c_start = info.get("period_index", 0) + c_dur = info.get("duration", 1) + c_end = c_start + c_dur + + e_start = entry.period_index + e_dur = entry.duration + e_end = e_start + e_dur + + if max(c_start, e_start) < min(c_end, e_end): + raise HTTPException(400, f"Section {entry.section_id} already has a class scheduled at {const.DAYS[entry.day_index]} Period {e_start + 1}.") + + # Determine room: use provided room_id, or auto-pick first available + room_id = entry.room_id or "" + room_name = "" + if room_id: + for r in rooms: + if r["id"] == room_id: + room_name = f"{r['id']} ({'Lab' if r.get('is_lab') else r.get('building', 'Main')})" + break + elif rooms: + # Auto-assign: pick a room not occupied at this slot + occupied_rooms = set() + for info in current.values(): + if info.get("day_index") == entry.day_index and info.get("period_index") == entry.period_index: + occupied_rooms.add(info.get("room_id", "")) + # For multi-period blocks, also check period_index + 1 + dur = info.get("duration", 1) + if dur > 1: + for di in range(dur): + if info.get("day_index") == entry.day_index and info.get("period_index") + di == entry.period_index: + occupied_rooms.add(info.get("room_id", "")) + + # Prefer labs for lab subjects, regular rooms for theory + sub_info = subjects_lookup.get(entry.subject_code.lower(), {}) + is_lab_subject = sub_info.get("type", "THEORY").upper() == "LAB" + + for r in rooms: + if r["id"] not in occupied_rooms: + if is_lab_subject and r.get("is_lab"): + room_id = r["id"] + room_name = f"{r['id']} (Lab)" + break + elif not is_lab_subject and not r.get("is_lab"): + room_id = r["id"] + room_name = f"{r['id']} ({r.get('building', 'Main')})" + break + # Fallback: just pick the first available + if not room_id: + for r in rooms: + if r["id"] not in occupied_rooms: + room_id = r["id"] + room_name = f"{r['id']} ({r.get('building', 'Main')})" + break + + # Generate a unique task ID + base_id = f"{entry.subject_code.lower()}-{entry.section_id.lower()}-INJECT" + task_id = base_id + counter = 0 + while task_id in existing_ids: + counter += 1 + task_id = f"{base_id}-{counter}" + existing_ids.add(task_id) + + # Compute start_slot for compatibility + start_slot = entry.day_index * const.NUM_TEACHING_SLOTS_PER_DAY + entry.period_index + + schedule_entry = { + "start_slot": start_slot, + "day_index": entry.day_index, + "day_name": const.DAYS[entry.day_index] if entry.day_index < len(const.DAYS) else f"Day{entry.day_index}", + "period_index": entry.period_index, + "room_id": room_id, + "room_name": room_name, + "faculty_name": entry.faculty_name, + "subject_code": entry.subject_code.lower(), + "section_id": entry.section_id, + "duration": entry.duration, + } + + current[task_id] = schedule_entry + injected_entries.append({"task_id": task_id, **schedule_entry}) + + # Save updated schedule + save_schedule(current) + # Log history + clean = _clean(current) + changes, affected = diff_schedules(old_schedule, clean) + if changes: + add_history_entry( + operation_type="INJECT_SUBJECT", + description=f"Injected {len(injected_entries)} class(es) (e.g. {injected_entries[0].get('subject_code') if injected_entries else 'subject'})", + affected_sections=affected, + changes=changes, + status="SUCCESS", + constraints=[] + ) + + return { + "status": "SUCCESS", + "message": f"Injected {len(injected_entries)} class(es) into the timetable.", + "injected": injected_entries, + "schedule": clean, + "grid": _build_grid(clean), + **_timetable_constants(), + } + +@app.post("/schedule/remove") +def remove_class(req: RemoveRequest): + """Remove a specific class from the timetable manually.""" + if not schedule_exists(): + raise HTTPException(400, "No timetable generated yet.") + + sched_data = load_schedule() + current = sched_data["schedule"] + old_schedule = _clean(dict(current)) # snapshot before mutation + + if req.task_id not in current: + raise HTTPException(404, f"Class with ID {req.task_id} not found.") + + removed_entry = current.pop(req.task_id) + + # Save updated schedule + save_schedule(current) + clean = _clean(current) + changes, affected = diff_schedules(old_schedule, clean) + if changes: + add_history_entry( + operation_type="REMOVE_SUBJECT", + description=f"Removed class: {removed_entry.get('subject_code')} from {removed_entry.get('section_id')}", + affected_sections=affected, + changes=changes, + status="SUCCESS", + constraints=[] + ) + + clean = _clean(current) + return { + "status": "SUCCESS", + "message": "Class removed successfully.", + "removed": req.task_id, + "schedule": clean, + "grid": _build_grid(clean), + **_timetable_constants(), + } + + +@app.get("/schedule/free-teachers") +def get_free_teachers(day_index: int, period_index: int): + """Return teachers who have NO class at the given (day, period) slot.""" + if not schedule_exists(): + raise HTTPException(400, "No timetable generated yet.") + + data = load_data() + sched_data = load_schedule() + current = sched_data["schedule"] + faculties = data.get("faculties", []) + + # Find all faculty names busy at this slot + busy_names = set() + for info in current.values(): + d = info.get("day_index") + p = info.get("period_index") + dur = info.get("duration", 1) + if d == day_index: + for i in range(dur): + if p + i == period_index: + busy_names.add(info.get("faculty_name", "").strip().lower()) + + # Return faculties NOT busy + free = [] + busy = [] + for fac in faculties: + fac_name = fac.get("name", "").strip() + if fac_name.lower() in busy_names: + busy.append({ + "id": fac["id"], + "name": fac_name, + "designation": fac.get("designation", ""), + "max_hours": fac.get("max_hours", 18), + "status": "busy", + }) else: - with st.spinner("Calling SLM API..."): - result = get_constraint(prompt) - if result.get('success'): - st.subheader("Constraint that would be applied (SLM API):") - for c in result['constraints']: - st.json(c) - else: - st.error(f"Failed to parse: {result.get('error')}") - st.code(result.get('raw', '')) - - # Apply mode — actually update the timetable - if apply and prompt.strip(): - from slm_inference import smart_parse - - # ── Step 1: Try smart_parse FIRST for high-confidence direct operations ── - # These patterns are reliably detected locally without needing the SLM - priority_keywords = [ - # Faculty operations - 'replace','substitute','take over','will take','on leave','cover', - 'permanently','change faculty','hand over','assign all', - # Cancel / holiday - 'cancel','no class','holiday','off day', - 'no toc','no nlp','no ml','no cn','no sepm','no nosql', - 'no rmipr','no dvlab','no cnlab','no nlplab','no mllab', - 'no iks','no evs','no genai','no devops','no hcai', - # Move / reschedule - 'move','shift','reschedule','transfer','relocate', - # Room - 'change room','to lab','to room','assign room', - # Extra class - 'extra class','makeup','compensatory','schedule extra', - 'add extra','add makeup','additional session','extra session', - 'extra ml','extra nlp','extra toc','extra cn','extra sepm', - 'schedule.*class','add.*class', - # Swap / freeze - 'swap','exchange','freeze','lock slot', - # NO_FREE_PERIOD — must be here so smart_parse runs first - 'should not be free','must not be free','cannot be free', - 'must have a class','should have a class','always occupied', - 'no free period','no free slot','must be filled', - 'first hour','first period','last period','last hour', - ] - use_smart_parse_first = any(kw in prompt.lower() for kw in priority_keywords) + free.append({ + "id": fac["id"], + "name": fac_name, + "designation": fac.get("designation", ""), + "max_hours": fac.get("max_hours", 18), + "status": "free", + }) + + return { + "day_index": day_index, + "period_index": period_index, + "free_count": len(free), + "busy_count": len(busy), + "free_teachers": free, + "busy_teachers": busy, + } + + +# ═════════════════════════════════════════════════════════════════════════════ +# UPDATE TIMETABLE +# ═════════════════════════════════════════════════════════════════════════════ + +_PRIORITY_KEYWORDS = [ + "replace", "substitute", "take over", "will take", "on leave", "cover", + "permanently", "change faculty", "hand over", "assign all", + "cancel", "no class", "holiday", "off day", + "no toc", "no nlp", "no ml", "no cn", "no sepm", "no nosql", + "move", "shift", "reschedule", "transfer", "relocate", + "change room", "to lab", "to room", "assign room", + "extra class", "makeup", "compensatory", "schedule extra", + "swap", "exchange", "freeze", "lock slot", + "should not be free", "must not be free", "cannot be free", + "must have a class", "should have a class", "no free period", + "first hour", "first period", "last period", "last hour", +] + +@app.post("/update") +def update(req: UpdateRequest): + if not schedule_exists(): + raise HTTPException(400, "No timetable generated yet. Call POST /generate first.") + + prompt = req.prompt.strip() + if not prompt: + raise HTTPException(400, "Prompt cannot be empty.") + + try: + data = load_data() constraints = [] - if use_smart_parse_first: - local = smart_parse(prompt, data['faculties'], - data.get('subjects',[]), data.get('sections',[])) + use_local = any(kw in prompt.lower() for kw in _PRIORITY_KEYWORDS) + if use_local: + local = smart_parse(prompt, data["faculties"], + data.get("subjects", []), data.get("sections", [])) if local: constraints = [local] - st.info(f"🔄 Operation detected: `{local['type']}`") - # ── Step 2: Fall back to SLM API if smart_parse couldn't handle it ── if not constraints: - with st.spinner("🤖 Converting instruction to constraint..."): - result = get_constraint(prompt) - if not result.get('success'): - st.error(f"❌ Could not parse instruction: {result.get('error')}") - st.code(result.get('raw', '')) - st.stop() - constraints = result.get('constraints', []) - if constraints: - st.info(f"Parsed constraint: `{constraints[0].get('type','?')}`") + result = get_constraint(prompt) + if not result.get("success"): + raise HTTPException(400, result.get("error", "Constraint parse failed.")) + constraints = result.get("constraints", []) if not constraints: - st.error("No constraints parsed."); st.stop() + raise HTTPException(422, "Could not parse any constraint from the instruction.") - st.info(f"Parsed constraint: `{constraints[0]['type'] if constraints else 'none'}`") + if req.preview_only: + return { + "preview": True, + "parsed_constraints": constraints, + "constraint_type": constraints[0].get("type", "?"), + } - # Load current schedule and rebuild objects sched = load_schedule() - current_solution = sched['schedule'] - facs, subs, secs, rooms, allocs = rebuild_objects(data) - tasks = prepare_scheduling_tasks(allocs, facs, subs, secs) + current_solution = sched["schedule"] + previous_schedule_clean = _clean(current_solution) - # Rebuild task objects into solution (attach task_obj) + facs, subs, secs, rooms, allocs = _build_objects(data) + tasks = prepare_scheduling_tasks(allocs, facs, subs, secs) tasks_by_id = {t.task_id: t for t in tasks} + for tid, info in current_solution.items(): if tid in tasks_by_id: - info['task_obj'] = tasks_by_id[tid] + info["task_obj"] = tasks_by_id[tid] - # Run partial optimizer for each constraint - all_changes = [] final_solution = current_solution + all_changes = [] for constraint in constraints: - with st.spinner(f"Rescheduling tasks for: {constraint['type']}..."): - optimizer = PartialOptimizer( - tasks, facs, secs, rooms, final_solution) - status, new_solution, affected, summary = \ - optimizer.apply_constraint_and_reoptimize(constraint) + optimizer = PartialOptimizer(tasks, facs, secs, rooms, final_solution) + op_status, new_solution, _, summary = \ + optimizer.apply_constraint_and_reoptimize(constraint) - if status in ('OPTIMAL', 'FEASIBLE', 'NO_CHANGE'): + if op_status in ("OPTIMAL", "FEASIBLE", "NO_CHANGE"): final_solution = new_solution all_changes.append(summary) else: - st.warning(f"⚠️ {summary}") + all_changes.append(f"⚠️ {summary}") + + if req.propose_only: + # Compute but don't save — return schedule for the frontend to propose + clean = _clean(final_solution) + return { + "status": "PROPOSED", + "parsed_constraints": constraints, + "changes": all_changes, + "schedule": clean, + } - # Save updated schedule save_schedule(final_solution) + clean = _clean(final_solution) + changes, affected = diff_schedules(previous_schedule_clean, clean) + add_history_entry( - operation_type="LLM_UPDATE", + operation_type="SLM_UPDATE", description=f"AI Update: {prompt}", - affected_sections=[], - changes=[], - status='SUCCESS', + affected_sections=affected, + changes=changes, + status="SUCCESS", constraints=constraints ) - # Show results - st.success("✅ Timetable updated!") - for change in all_changes: - st.markdown(change) - - st.divider() - - # ── Side-by-side: Original vs Updated ──────────────────────────── - col_orig, col_new = st.columns(2) - with col_orig: - st.subheader("📋 Original Timetable") - render_timetable_html(current_solution, secs) - with col_new: - st.subheader("📅 Updated Timetable") - render_timetable_html(final_solution, secs) - - # ── Show current timetable below ────────────────────────────────────── - st.divider() - with st.expander("📄 View Current Full Timetable"): - sched = load_schedule() - if sched: - _, _, secs, _, _ = rebuild_objects(data) - render_timetable_html(sched['schedule'], secs) + clean = _clean(final_solution) + return { + "status": "SUCCESS", + "parsed_constraints": constraints, + "constraint_type": constraints[0].get("type", "?"), + "changes": all_changes, + "previous_schedule": previous_schedule_clean, + "updated_schedule": clean, + "grid": _build_grid(clean), + **_timetable_constants(), + } -# ═════════════════════════════════════════════════════════════════════════════ -# PAGE: ORIGINAL VS CURRENT -# ═════════════════════════════════════════════════════════════════════════════ -elif page == "📊 Original vs Current": - st.header("📊 Original vs Current Timetable") + except HTTPException: + raise + except Exception: + raise HTTPException(500, traceback.format_exc()) - if not schedule_exists(): - st.error("❌ No timetable generated yet.") - st.stop() - _, _, secs, _, _ = rebuild_objects(data) +# ═════════════════════════════════════════════════════════════════════════════ +# LEAVES & SUBSTITUTIONS +# ═════════════════════════════════════════════════════════════════════════════ - has_orig = original_schedule_exists() - orig = load_original_schedule() if has_orig else None - current = load_schedule() +class LeaveRequestIn(BaseModel): + faculty_id: str + days: list[str] + reason: str + +@app.get("/leave") +def get_leaves(): + return {"leaves": load_leave_requests()} + +@app.post("/leave", status_code=201) +def create_leave(req: LeaveRequestIn): + leaves = load_leave_requests() + leave_id = str(uuid.uuid4()) + new_leave = { + "leave_id": leave_id, + "faculty_id": req.faculty_id, + "days": req.days, + "reason": req.reason, + "status": "PENDING" + } + leaves.append(new_leave) + save_leave_requests(leaves) + return {"message": "Leave request created", "leave": new_leave} + +@app.post("/leave/approve/{leave_id}") +def approve_leave(leave_id: str): + leaves = load_leave_requests() + target_leave: dict | None = next((l for l in leaves if l.get("leave_id") == leave_id), None) + + if not target_leave: + raise HTTPException(404, "Leave request not found") + + if target_leave.get("status") != "PENDING": + raise HTTPException(400, f"Leave is already {target_leave.get('status')}") + + target_leave["status"] = "APPROVED" + save_leave_requests(leaves) + + # Convert dict to namedtuple or dataclass instance expected by engine + from models import LeaveRequest as LRModel + lr_obj = LRModel(**target_leave) + + # Trigger substitution finder + process_leave_approval(lr_obj) + + return {"message": "Leave approved and substitution process started."} + +@app.post("/leave/reject/{leave_id}") +def reject_leave(leave_id: str): + leaves = load_leave_requests() + target = next((l for l in leaves if l["leave_id"] == leave_id), None) + if not target: raise HTTPException(404, "Leave request not found") + target["status"] = "REJECTED" + save_leave_requests(leaves) + return {"message": "Leave request rejected."} + +@app.get("/substitution") +@app.get("/substitution/pending") +def get_pending_substitutions(faculty_id: str | None = None): + check_timeouts() + reqs = load_substitution_requests() + + if faculty_id: + reqs = [r for r in reqs if r["candidate_faculty_id"] == faculty_id and r["status"] == "PENDING"] + else: + reqs = [r for r in reqs if r["status"] == "PENDING"] + + return {"substitutions": reqs} + +@app.post("/substitution/{request_id}/accept") +def accept_substitution(request_id: str): + success, msg = handle_acceptance(request_id) + if not success: + raise HTTPException(400, msg) + return {"message": msg} + +@app.post("/substitution/{request_id}/decline") +def decline_substitution(request_id: str): + success, msg = handle_decline(request_id) + if not success: + raise HTTPException(400, msg) + return {"message": msg} + +@app.get("/substitution/unresolved") +def get_unresolved_substitutions(): + check_timeouts() + # A slot is unresolved if all requests for it are DECLINED/TIMEOUT, and no ACCEPTED exists + # Or if no requests were generated at all (handled separately or indicated by lack of requests) + reqs = load_substitution_requests() + leaves = load_leave_requests() + + unresolved_slots = [] + # simplified logic: find slots where all reqs are not PENDING/ACCEPTED + # A true implementation would group by leave_id + slot + + return {"unresolved": unresolved_slots, "message": "Not fully implemented for MVP"} - if orig: - orig_at = orig.get('generated_at', '')[:16].replace('T', ' ') - curr_at = current.get('generated_at', '')[:16].replace('T', ' ') if current else '' +# ═════════════════════════════════════════════════════════════════════════════ +# CANCELLATION REQUESTS +# ═════════════════════════════════════════════════════════════════════════════ - # Summary badge - history = load_history() - n_changes = len(history) - col1, col2, col3 = st.columns(3) - col1.metric("Original Generated", orig_at if orig else "N/A") - col2.metric("Last Updated", curr_at) - col3.metric("Total Changes Applied", n_changes) - - st.divider() - - if orig: - tab1, tab2 = st.tabs(["🗂️ Original Timetable", "📅 Current Timetable"]) - with tab1: - st.caption(f"Generated on {orig_at} — never modified") - render_timetable_html(orig['schedule'], secs) - with tab2: - st.caption(f"Last updated: {curr_at} — {n_changes} change(s) applied") - render_timetable_html(current['schedule'], secs) - else: - st.info("Original snapshot not available. " - "Regenerate the timetable to create one.") - st.subheader("Current Timetable") - render_timetable_html(current['schedule'], secs) +class CancellationRequestIn(BaseModel): + section_id: str + day: str + period: int + subject: str + reason: str + faculty_id: str + +@app.post("/cancellations/request", status_code=201) +def create_cancellation_request(req: CancellationRequestIn): + cancellations = load_cancellations() + cancel_id = str(uuid.uuid4()) + new_cancel = { + "id": cancel_id, + "section_id": req.section_id, + "day": req.day, + "period": req.period, + "subject": req.subject, + "reason": req.reason, + "faculty_id": req.faculty_id, + "status": "PENDING", + "created_at": datetime.now().isoformat() + } + cancellations.append(new_cancel) + save_cancellations(cancellations) + return {"message": "Cancellation request submitted", "cancellation": new_cancel} + +@app.get("/cancellations") +def get_cancellations(): + return {"cancellations": load_cancellations()} + +@app.post("/cancellations/{cancel_id}/status") +def update_cancellation_status(cancel_id: str, payload: dict = Body(...)): + status = payload.get("status") + if not status: + raise HTTPException(400, "Missing status") + + cancellations = load_cancellations() + target = next((c for c in cancellations if c["id"] == cancel_id), None) + if not target: + raise HTTPException(404, "Cancellation not found") + + target["status"] = status + save_cancellations(cancellations) + return {"message": f"Cancellation marked as {status}", "cancellation": target} # ═════════════════════════════════════════════════════════════════════════════ -# PAGE: CHANGE HISTORY +# CHANGE HISTORY # ═════════════════════════════════════════════════════════════════════════════ -elif page == "📋 Change History": - st.header("📋 Change History") +@app.get("/history") +def get_history(): history = load_history() - if not history: - st.info("No changes have been made yet.") - else: - st.write(f"**{len(history)} change(s) recorded**") - for i, entry in enumerate(reversed(history)): - ts = entry['timestamp'][:16].replace('T', ' ') - desc = entry.get('description', entry.get('prompt', '')) - op_type = entry.get('operation_type', 'UPDATE') - with st.expander(f"#{len(history)-i} — {ts} — [{op_type}] {desc}"): - st.markdown(f"**Description:** {desc}") - st.markdown(f"**Operation:** {op_type}") - st.markdown(f"**Status:** {entry.get('status', 'UNKNOWN')}") - sections = entry.get('affected_sections', []) - if sections: - st.markdown(f"**Affected Sections:** {', '.join(sections)}") - changes = entry.get('changes', []) - if changes: - st.subheader(f"Detailed Changes ({len(changes)} cells)") - for c in changes[:20]: - st.json(c) - if entry.get('constraints'): - st.subheader("Constraint applied:") - st.json(entry['constraints']) - - if st.button("🗑️ Clear History"): - clear_history() - st.rerun() - -# ═════════════════════════════════════════════════════════════════════════════ -# MANAGEMENT PAGES (unchanged from original) -# ═════════════════════════════════════════════════════════════════════════════ -elif page == "👥 Manage Faculties": - st.header("Manage Faculties") - with st.form("add_faculty"): - col1, col2 = st.columns(2) - f_id = col1.text_input("ID (e.g., 'F001')") - f_name = col2.text_input("Name") - f_desig = col1.selectbox("Designation", - ["Professor","Assoc. Prof","Asst. Prof","Guest"]) - f_max = col2.number_input("Max Hours/Week", min_value=1, value=18) - if st.form_submit_button("Add Faculty"): - if f_id and f_name: - data['faculties'].append({"id": f_id, "name": f_name, - "designation": f_desig, "max_hours": f_max}) - save_data(data); st.success("Added!"); st.rerun() - else: - st.error("ID and Name are required.") - if data['faculties']: - st.dataframe(pd.DataFrame(data['faculties'])) - if st.button("Clear All Faculties"): - data['faculties'] = []; save_data(data); st.rerun() - -elif page == "📚 Manage Subjects": - st.header("Manage Subjects") - with st.form("add_subject"): - col1, col2 = st.columns(2) - s_code = col1.text_input("Code") - s_name = col2.text_input("Name") - s_type = col1.selectbox("Type", ["THEORY","LAB","SOFTSKILL","FORUM"]) - s_cred = col2.number_input("Credits", min_value=0, value=3) - s_core = col1.checkbox("Is Core?", value=True) - s_heavy = col2.checkbox("Is Heavy?", value=False) - if st.form_submit_button("Add Subject"): - if s_code: - data['subjects'].append({"code": s_code, "name": s_name, - "type": s_type, "credits": s_cred, - "is_core": s_core, "is_heavy": s_heavy}) - save_data(data); st.success("Added!"); st.rerun() - if data['subjects']: - st.dataframe(pd.DataFrame(data['subjects'])) - if st.button("Clear All Subjects"): - data['subjects'] = []; save_data(data); st.rerun() - -elif page == "🏛️ Manage Sections": - st.header("Manage Sections") - with st.form("add_section"): - col1, col2 = st.columns(2) - sec_id = col1.text_input("Section ID (e.g., '6A')") - sem = col2.number_input("Semester", min_value=1, value=6) - strength = col1.number_input("Student Strength", min_value=1, value=60) - if st.form_submit_button("Add Section"): - if sec_id: - data['sections'].append({"id": sec_id, "semester": sem, - "strength": strength}) - save_data(data); st.success("Added!"); st.rerun() - if data['sections']: - st.dataframe(pd.DataFrame(data['sections'])) - if st.button("Clear All Sections"): - data['sections'] = []; save_data(data); st.rerun() - -elif page == "🚪 Manage Rooms": - st.header("Manage Rooms") - with st.form("add_room"): - col1, col2 = st.columns(2) - r_id = col1.text_input("Room ID") - cap = col2.number_input("Capacity", min_value=1, value=80) - is_lab = col1.checkbox("Is Lab?", value=False) - bld = col2.text_input("Building", value="Main") - if st.form_submit_button("Add Room"): - if r_id: - data['rooms'].append({"id": r_id, "capacity": cap, - "is_lab": is_lab, "building": bld}) - save_data(data); st.success("Added!"); st.rerun() - if data['rooms']: - st.dataframe(pd.DataFrame(data['rooms'])) - if st.button("Clear All Rooms"): - data['rooms'] = []; save_data(data); st.rerun() - -elif page == "🔗 Manage Allocations": - st.header("Manage Allocations") - if not (data['faculties'] and data['subjects'] and data['sections']): - st.warning("Add Faculties, Subjects, and Sections first.") - else: - fac_opts = {f['name']: f['id'] for f in data['faculties']} - sub_opts = {s['name']: s['code'] for s in data['subjects']} - sec_opts = [s['id'] for s in data['sections']] - with st.form("add_alloc"): - col1, col2 = st.columns(2) - f = col1.selectbox("Faculty", list(fac_opts.keys())) - s = col2.selectbox("Subject", list(sub_opts.keys())) - sec = col1.selectbox("Section", sec_opts) - grp = col2.text_input("Elective Group ID (Optional)") - if st.form_submit_button("Add Allocation"): - data['allocations'].append({ - "faculty_id": fac_opts[f], - "subject_code": sub_opts[s], - "section_id": sec, - "elective_group": grp if grp else None - }) - save_data(data); st.success("Allocation Added!"); st.rerun() - if data['allocations']: - st.dataframe(pd.DataFrame(data['allocations'])) - if st.button("Clear Allocations"): - data['allocations'] = []; save_data(data); st.rerun() \ No newline at end of file + return { + "count": len(history), + "history": list(reversed(history)), + } + +@app.post("/history/revert/{history_id}") +def revert_history(history_id: str, force: bool = False): + history = load_history() + entry = next((e for e in history if e.get("id") == history_id), None) + if not entry: + raise HTTPException(404, "History entry not found.") + + sched_raw = load_schedule() or {} + current_sched = sched_raw.get("schedule", {}) if isinstance(sched_raw, dict) else {} + changes = entry.get("changes", []) + + if not force: + # Check for conflicts + conflicts = [] + for c in changes: + tid = c["task_id"] + curr_val = _clean({tid: current_sched[tid]}).get(tid) if tid in current_sched else None + after_val = c["after"] + if curr_val != after_val: + conflicts.append(tid) + + if conflicts: + raise HTTPException(409, { + "message": "Conflict detected: The timetable has been modified since this change was made. Reverting will overwrite those newer modifications. Do you want to force revert?", + "conflicts": conflicts + }) + + # Apply revert + new_sched = dict(current_sched) + for c in changes: + tid = c["task_id"] + before_val = c["before"] + if before_val is None: + if tid in new_sched: + del new_sched[tid] + else: + new_sched[tid] = dict(before_val) + + clean_new = _clean(new_sched) + save_schedule(clean_new) + + # Mark the original entry as REVERTED (keep it visible) + entry["status"] = "REVERTED" + save_history(history) + + # Log the revert itself as a new entry + rev_changes, rev_affected = diff_schedules(_clean(current_sched), clean_new) + if rev_changes: + add_history_entry( + operation_type="REVERT", + description=f"Reverted: {entry.get('description', history_id)}", + affected_sections=rev_affected, + changes=rev_changes, + status="SUCCESS" + ) + + return { + "status": "SUCCESS", + "message": "Revert successful.", + "schedule": clean_new, + "grid": _build_grid(clean_new), + **_timetable_constants(), + } + +@app.delete("/history") +def delete_history(): + clear_history() + return {"message": "History cleared."}