app.py

# app.py
import csv
import io
import os
import random
from typing import List, Dict, Tuple

import gradio as gr
import numpy as np
import pandas as pd
from dotenv import load_dotenv

load_dotenv()

# ------------------------
# Genetic Algorithm Logic
# ------------------------

class TimetableGA:
    def __init__(
        self,
        courses: List[str],
        teachers: List[str],
        rooms: List[str],
        days: List[str],
        slots: List[str],
        teacher_unavailable: Dict[str, List[Tuple[str, str]]],  # teacher -> list of (day,slot)
        course_teacher_pref: Dict[str, List[str]],  # course -> possible teachers
        room_constraints: Dict[str, List[str]],  # course -> allowed rooms (optional)
        population_size: int = 100,
        generations: int = 200,
        mutation_rate: float = 0.05,
    ):
        self.courses = courses
        self.teachers = teachers
        self.rooms = rooms
        self.days = days
        self.slots = slots
        self.teacher_unavailable = teacher_unavailable
        self.course_teacher_pref = course_teacher_pref
        self.room_constraints = room_constraints or {}
        self.population_size = population_size
        self.generations = generations
        self.mutation_rate = mutation_rate

        self.times = [(d, s) for d in days for s in slots]
        self.num_periods = len(self.times)
        self.num_courses = len(courses)

    def _random_individual(self):
        """
        Individual representation:
         - For each course, a tuple (period_index, room_index, teacher_index)
         - Represented as arrays: period_indices, room_indices, teacher_indices
        """
        period_indices = np.random.randint(0, self.num_periods, size=self.num_courses)
        room_indices = np.random.randint(0, len(self.rooms), size=self.num_courses)
        teacher_indices = np.zeros(self.num_courses, dtype=int)
        for i, c in enumerate(self.courses):
            # choose teacher from preferences if provided, else random
            prefs = self.course_teacher_pref.get(c, None)
            if prefs:
                # pick one of allowed teachers
                teacher_indices[i] = self.teachers.index(random.choice(prefs))
            else:
                teacher_indices[i] = np.random.randint(0, len(self.teachers))
        return (period_indices, room_indices, teacher_indices)

    def _fitness(self, individual):
        period_indices, room_indices, teacher_indices = individual
        score = 0
        penalties = 0

        # 1) Teacher conflicts: a teacher cannot teach more than one course in same period
        teacher_period = {}
        for i, t_idx in enumerate(teacher_indices):
            key = (t_idx, int(period_indices[i]))
            teacher_period.setdefault(key, 0)
            teacher_period[key] += 1
        teacher_conflicts = sum(max(0, cnt - 1) for cnt in teacher_period.values())
        penalties += teacher_conflicts * 5

        # 2) Room conflicts
        room_period = {}
        for i, r_idx in enumerate(room_indices):
            key = (r_idx, int(period_indices[i]))
            room_period.setdefault(key, 0)
            room_period[key] += 1
        room_conflicts = sum(max(0, cnt - 1) for cnt in room_period.values())
        penalties += room_conflicts * 5

        # 3) Teacher availability violations
        avail_violations = 0
        for i, t_idx in enumerate(teacher_indices):
            teacher = self.teachers[t_idx]
            period = self.times[int(period_indices[i])]
            if teacher in self.teacher_unavailable:
                if period in self.teacher_unavailable[teacher]:
                    avail_violations += 1
        penalties += avail_violations * 10

        # 4) Course-teacher preference violations
        pref_violations = 0
        for i, c in enumerate(self.courses):
            prefs = self.course_teacher_pref.get(c)
            if prefs:
                chosen_teacher = self.teachers[teacher_indices[i]]
                if chosen_teacher not in prefs:
                    pref_violations += 1
        penalties += pref_violations * 2

        # 5) Room constraint violations
        room_violations = 0
        for i, c in enumerate(self.courses):
            allowed = self.room_constraints.get(c)
            if allowed:
                chosen_room = self.rooms[room_indices[i]]
                if chosen_room not in allowed:
                    room_violations += 1
        penalties += room_violations * 4

        # 6) Spread penalty (optional): same course multiple occurrences in same day slot collisions (if course repeated)
        # For simple use-case assume each course appears once - no extra penalty.

        # Fitness: higher is better. Start from base and subtract penalties.
        base = 1000
        fitness = base - penalties
        # Provide components for debugging in return as well
        return fitness

    def _crossover(self, parent_a, parent_b):
        # single-point crossover on all arrays
        cut = np.random.randint(1, self.num_courses - 1)
        a_period, a_room, a_teacher = parent_a
        b_period, b_room, b_teacher = parent_b
        child1 = (
            np.concatenate([a_period[:cut], b_period[cut:]]),
            np.concatenate([a_room[:cut], b_room[cut:]]),
            np.concatenate([a_teacher[:cut], b_teacher[cut:]]),
        )
        child2 = (
            np.concatenate([b_period[:cut], a_period[cut:]]),
            np.concatenate([b_room[:cut], a_room[cut:]]),
            np.concatenate([b_teacher[:cut], a_teacher[cut:]]),
        )
        return child1, child2

    def _mutate(self, individual):
        period_indices, room_indices, teacher_indices = individual
        for i in range(self.num_courses):
            if random.random() < self.mutation_rate:
                period_indices[i] = random.randint(0, self.num_periods - 1)
            if random.random() < self.mutation_rate:
                room_indices[i] = random.randint(0, len(self.rooms) - 1)
            if random.random() < self.mutation_rate:
                prefs = self.course_teacher_pref.get(self.courses[i], None)
                if prefs:
                    teacher_indices[i] = self.teachers.index(random.choice(prefs))
                else:
                    teacher_indices[i] = random.randint(0, len(self.teachers) - 1)
        return (period_indices, room_indices, teacher_indices)

    def run(self, verbose=False):
        # Initialize population
        population = [self._random_individual() for _ in range(self.population_size)]
        fitnesses = [self._fitness(ind) for ind in population]
        best = population[np.argmax(fitnesses)]
        best_score = max(fitnesses)

        for gen in range(self.generations):
            # Selection (tournament)
            new_pop = []
            while len(new_pop) < self.population_size:
                i1, i2 = random.sample(range(self.population_size), 2)
                p1 = population[i1] if fitnesses[i1] > fitnesses[i2] else population[i2]
                i3, i4 = random.sample(range(self.population_size), 2)
                p2 = population[i3] if fitnesses[i3] > fitnesses[i4] else population[i4]

                c1, c2 = self._crossover(p1, p2)
                c1 = self._mutate(c1)
                c2 = self._mutate(c2)
                new_pop.extend([c1, c2])

            population = new_pop[: self.population_size]
            fitnesses = [self._fitness(ind) for ind in population]

            gen_best_idx = int(np.argmax(fitnesses))
            gen_best_score = fitnesses[gen_best_idx]
            if gen_best_score > best_score:
                best_score = gen_best_score
                best = population[gen_best_idx]

            # early exit if perfect
            if best_score >= 1000:
                break

            if verbose and gen % max(1, self.generations // 10) == 0:
                print(f"Gen {gen} best {best_score}")

        return {"best": best, "score": best_score, "times": self.times}

# ------------------------
# Helpers to convert individual -> dataframe
# ------------------------

def individual_to_dataframe(individual, courses, teachers, rooms, times):
    period_indices, room_indices, teacher_indices = individual
    rows = []
    for i, course in enumerate(courses):
        period_idx = int(period_indices[i])
        day, slot = times[period_idx]
        rows.append(
            {
                "Course": course,
                "Teacher": teachers[int(teacher_indices[i])],
                "Room": rooms[int(room_indices[i])],
                "Day": day,
                "Slot": slot,
            }
        )
    return pd.DataFrame(rows).sort_values(["Day", "Slot"]).reset_index(drop=True)

def dataframe_to_csv_bytes(df: pd.DataFrame):
    buf = io.StringIO()
    df.to_csv(buf, index=False)
    return buf.getvalue().encode("utf-8")

# ------------------------
# Gradio UI
# ------------------------

def parse_multiline_list(text: str) -> List[str]:
    return [line.strip() for line in text.splitlines() if line.strip()]

def parse_teacher_unavailability(text: str) -> Dict[str, List[Tuple[str,str]]]:
    # Format per line: Teacher,Day,Slot
    # Example: T1_Ali,Monday,Slot1
    d = {}
    for ln in text.splitlines():
        ln = ln.strip()
        if not ln: 
            continue
        parts = [p.strip() for p in ln.split(",")]
        if len(parts) >= 3:
            teacher, day, slot = parts[0], parts[1], parts[2]
            d.setdefault(teacher, []).append((day, slot))
    return d

def parse_course_teacher_pref(text: str) -> Dict[str, List[str]]:
    # Format per line: Course: T1,T2
    d = {}
    for ln in text.splitlines():
        ln = ln.strip()
        if not ln: 
            continue
        if ":" in ln:
            course, rest = ln.split(":", 1)
            teachers = [t.strip() for t in rest.split(",") if t.strip()]
            d[course.strip()] = teachers
    return d

def parse_room_constraints(text: str) -> Dict[str, List[str]]:
    # Format per line: Course: R1,R2
    d = {}
    for ln in text.splitlines():
        ln = ln.strip()
        if not ln:
            continue
        if ":" in ln:
            course, rest = ln.split(":", 1)
            rooms = [r.strip() for r in rest.split(",") if r.strip()]
            d[course.strip()] = rooms
    return d

def run_ga_and_return_csv(
    courses_text, teachers_text, rooms_text, days_text, slots_text,
    teacher_unavail_text, course_teacher_pref_text, room_constraints_text,
    pop_size, generations, mutation_rate, seed=42
):
    random.seed(int(seed))
    np.random.seed(int(seed))

    courses = parse_multiline_list(courses_text)
    teachers = parse_multiline_list(teachers_text)
    rooms = parse_multiline_list(rooms_text)
    days = parse_multiline_list(days_text)
    slots = parse_multiline_list(slots_text)
    if not (courses and teachers and rooms and days and slots):
        return "Please provide at least one course, teacher, room, day, and slot.", None, None

    teacher_unavail = parse_teacher_unavailability(teacher_unavail_text)
    course_teacher_pref = parse_course_teacher_pref(course_teacher_pref_text)
    room_constraints = parse_room_constraints(room_constraints_text)

    ga = TimetableGA(
        courses=courses,
        teachers=teachers,
        rooms=rooms,
        days=days,
        slots=slots,
        teacher_unavailable=teacher_unavail,
        course_teacher_pref=course_teacher_pref,
        room_constraints=room_constraints,
        population_size=int(pop_size),
        generations=int(generations),
        mutation_rate=float(mutation_rate),
    )
    result = ga.run(verbose=False)
    best = result["best"]
    score = result["score"]
    times = result["times"]
    df = individual_to_dataframe(best, courses, teachers, rooms, times)
    csv_bytes = dataframe_to_csv_bytes(df)
    summary = f"GA finished. Best fitness score: {score}. Generated timetable rows: {len(df)}"
    return summary, df, csv_bytes

# ------------------------
# Build Gradio app
# ------------------------

with gr.Blocks(title="Automatic Time Table Generation Agent (Genetic Algorithm)") as demo:
    gr.Markdown("# Automatic Time Table Generation Agent (Genetic Algorithm)")
    gr.Markdown(
        "Create an optimized timetable using a genetic algorithm. "
        "Enter courses, teachers, rooms, days and slots; optionally provide teacher unavailability, course-teacher preferences and room constraints."
    )

    with gr.Row():
        with gr.Column(scale=1):
            gr.Markdown("## Inputs")
            courses_in = gr.Textbox(label="Courses (one per line)", value="C1_Math\nC2_Physics\nC3_Chemistry\nC4_English\nC5_Biology\nC6_History", lines=6)
            teachers_in = gr.Textbox(label="Teachers (one per line)", value="T1_Ali\nT2_Sara\nT3_Omar\nT4_Fatima", lines=4)
            rooms_in = gr.Textbox(label="Rooms (one per line)", value="R1\nR2\nR3\nR4\nR5", lines=4)
            days_in = gr.Textbox(label="Days (one per line)", value="Monday\nTuesday\nWednesday\nThursday\nFriday", lines=5)
            slots_in = gr.Textbox(label="Slots (one per line)", value="Slot1\nSlot2\nSlot3\nSlot4\nSlot5\nSlot6", lines=6)

            gr.Markdown("### Optional constraints")
            teacher_unavail_in = gr.Textbox(label="Teacher unavailability (one per line: Teacher,Day,Slot)", value="", lines=4)
            course_teacher_pref_in = gr.Textbox(label="Course -> allowed teachers (one per line: Course: T1,T2)", value="", lines=4)
            room_constraints_in = gr.Textbox(label="Course -> allowed rooms (one per line: Course: R1,R2)", value="", lines=4)

            gr.Markdown("### GA parameters")
            pop_in = gr.Slider(label="Population size", minimum=10, maximum=500, value=120, step=10)
            gen_in = gr.Slider(label="Generations", minimum=10, maximum=2000, value=300, step=10)
            mut_in = gr.Slider(label="Mutation rate", minimum=0.0, maximum=0.5, value=0.05, step=0.01)
            seed_in = gr.Number(label="Random seed", value=42)

            run_btn = gr.Button("Run Genetic Algorithm")

        with gr.Column(scale=1):
            gr.Markdown("## Output")
            summary_out = gr.Textbox(label="Summary", lines=3)
            table_out = gr.Dataframe(headers=["Course","Teacher","Room","Day","Slot"], interactive=False)
            download_btn = gr.File(label="Download CSV")
            gr.Markdown("## Timetable Agent (Chat)")
            chat_in = gr.Textbox(label="Ask the Timetable Agent (explain conflicts, suggest fixes...)")
            chat_out = gr.Textbox(label="Agent response", lines=6)

    def run_and_prepare_download(*args):
        (
            courses_text, teachers_text, rooms_text, days_text, slots_text,
            teacher_unavail_text, course_teacher_pref_text, room_constraints_text,
            pop_size, generations, mutation_rate, seed
        ) = args
        summary, df, csv_bytes = run_ga_and_return_csv(
            courses_text, teachers_text, rooms_text, days_text, slots_text,
            teacher_unavail_text, course_teacher_pref_text, room_constraints_text,
            pop_size, generations, mutation_rate, seed
        )
        if df is None:
            return summary, None, None, None
        # prepare in-memory file for Gradio
        file_obj = io.BytesIO(csv_bytes)
        file_obj.name = "timetable.csv"
        return summary, df, file_obj, "Timetable generated. Ask the agent for an explanation."

    run_btn.click(
        run_and_prepare_download,
        inputs=[
            courses_in, teachers_in, rooms_in, days_in, slots_in,
            teacher_unavail_in, course_teacher_pref_in, room_constraints_in,
            pop_in, gen_in, mut_in, seed_in
        ],
        outputs=[summary_out, table_out, download_btn, chat_out],
        show_progress=True,
    )

    # Simple local "chat" behavior (not LLM)
    def simple_agent(query, summary_text, df):
        if not query:
            return "Type a question like: 'Which teachers have conflicts?' or 'Suggest 3 ways to reduce conflicts.'"
        if df is None or df.shape[0] == 0:
            return "No timetable available. Run the GA first."
        # Basic analysis
        text = query.lower()
        response_lines = []
        if "conflict" in text or "conflicts" in text or "problem" in text:
            # detect teacher conflicts
            tconf = df.groupby(["Teacher","Day","Slot"]).size().reset_index(name="count")
            tconf = tconf[tconf["count"]>1]
            if not tconf.empty:
                response_lines.append("Teacher conflicts found:")
                for _, row in tconf.iterrows():
                    response_lines.append(f"- {row['Teacher']} has {row['count']} assignments on {row['Day']} {row['Slot']}")
            else:
                response_lines.append("No teacher conflicts detected.")

            # room conflicts
            rconf = df.groupby(["Room","Day","Slot"]).size().reset_index(name="count")
            rconf = rconf[rconf["count"]>1]
            if not rconf.empty:
                response_lines.append("Room conflicts found:")
                for _, row in rconf.iterrows():
                    response_lines.append(f"- {row['Room']} has {row['count']} assignments on {row['Day']} {row['Slot']}")
            else:
                response_lines.append("No room conflicts detected.")
            response_lines.append("Suggested fixes: 1) add rooms or change room constraints, 2) change teacher availability for offending periods, 3) allow alternate teachers for affected courses.")
            return "\n".join(response_lines)

        if "suggest" in text or "improve" in text or "reduce" in text:
            return "Three quick suggestions:\n1) Increase number of rooms or relax room constraints.\n2) Allow more teachers per course (course-teacher preferences).\n3) Move a class to a different slot/day for teachers with conflicts."

        # default explanation: show top 5 assignments
        sample = df.head(8).to_string(index=False)
        return f"Timetable summary (first rows):\n{sample}\n\nAsk for 'conflicts' or 'suggestions' for improvement."

    chat_btn = gr.Button("Ask Agent")
    chat_btn.click(simple_agent, inputs=[chat_in, summary_out, table_out], outputs=[chat_out])

    gr.Markdown("### Notes\n- This genetic algorithm is a demonstrative solver. For production use you should add stronger constraints (room capacities, repeating lessons, student group clashes) and tune GA parameters.")

if __name__ == "__main__":
    demo.launch(server_name="0.0.0.0", share=False)