397 lines
16 KiB
Python
397 lines
16 KiB
Python
import random
|
|
import copy
|
|
from datetime import datetime, timedelta
|
|
from typing import List, Dict, Tuple, Optional
|
|
from dataclasses import dataclass
|
|
from app import db
|
|
from app.models.classroom import Classroom
|
|
from app.models.reservation import Reservation, ReservationStatus
|
|
from app.models.subject import Commission
|
|
|
|
|
|
@dataclass
|
|
class ReservationRequest:
|
|
"""Represents a reservation request from the teaching committee"""
|
|
commission_id: int
|
|
expected_attendees: int
|
|
purpose: str
|
|
preferred_start_time: datetime
|
|
preferred_end_time: datetime
|
|
priority: int = 1 # 1=highest, 5=lowest
|
|
flexibility_hours: int = 2 # How flexible the start time can be
|
|
subject_requirements: Dict = None # Special requirements for the subject
|
|
|
|
def __post_init__(self):
|
|
if self.subject_requirements is None:
|
|
self.subject_requirements = {}
|
|
|
|
|
|
@dataclass
|
|
class Gene:
|
|
"""Represents a single reservation assignment"""
|
|
commission_id: int
|
|
classroom_id: int
|
|
start_time: datetime
|
|
end_time: datetime
|
|
expected_attendees: int
|
|
purpose: str
|
|
|
|
|
|
class Individual:
|
|
"""Represents a complete reservation schedule (chromosome)"""
|
|
|
|
def __init__(self, genes: List[Gene]):
|
|
self.genes = genes
|
|
self.fitness = 0.0
|
|
self.conflicts = []
|
|
|
|
def calculate_fitness(self, classrooms: Dict[int, Classroom], requests: Dict[int, ReservationRequest]) -> float:
|
|
"""Calculate fitness score based on multiple factors"""
|
|
score = 0.0
|
|
self.conflicts = []
|
|
|
|
for gene in self.genes:
|
|
classroom = classrooms.get(gene.classroom_id)
|
|
request = requests.get(gene.commission_id)
|
|
|
|
if not classroom or not request:
|
|
continue
|
|
|
|
# Factor 1: Capacity efficiency (40% weight)
|
|
capacity_score = self._calculate_capacity_score(classroom, gene.expected_attendees)
|
|
score += capacity_score * 0.4
|
|
|
|
# Factor 2: Time preference satisfaction (30% weight)
|
|
time_score = self._calculate_time_score(gene, request)
|
|
score += time_score * 0.3
|
|
|
|
# Factor 3: Conflict penalty (20% weight)
|
|
conflict_score = self._calculate_conflict_penalty(gene, self.genes)
|
|
score += conflict_score * 0.2
|
|
|
|
# Factor 4: Resource matching (10% weight)
|
|
resource_score = self._calculate_resource_score(classroom, request.subject_requirements)
|
|
score += resource_score * 0.1
|
|
|
|
# Store conflicts for debugging
|
|
if conflict_score < 0:
|
|
self.conflicts.append({
|
|
'gene': gene,
|
|
'conflict_type': 'time_overlap'
|
|
})
|
|
|
|
self.fitness = score
|
|
return score
|
|
|
|
def _calculate_capacity_score(self, classroom: Classroom, expected_attendees: int) -> float:
|
|
"""Calculate how well the classroom capacity matches the expected attendees"""
|
|
if classroom.capacity < expected_attendees:
|
|
return 0.0 # Overfilled classroom
|
|
|
|
# Calculate efficiency: closer capacity match = higher score
|
|
waste_ratio = (classroom.capacity - expected_attendees) / expected_attendees
|
|
if waste_ratio <= 0.1: # Less than 10% waste
|
|
return 1.0
|
|
elif waste_ratio <= 0.3: # Less than 30% waste
|
|
return 0.8
|
|
elif waste_ratio <= 0.5: # Less than 50% waste
|
|
return 0.6
|
|
else:
|
|
return 0.4
|
|
|
|
def _calculate_time_score(self, gene: Gene, request: ReservationRequest) -> float:
|
|
"""Calculate how well the assigned time matches preferences"""
|
|
# Calculate time difference from preferred time
|
|
time_diff = abs((gene.start_time - request.preferred_start_time).total_seconds() / 3600)
|
|
|
|
if time_diff <= 0.5: # Within 30 minutes
|
|
return 1.0
|
|
elif time_diff <= request.flexibility_hours: # Within flexible window
|
|
return 0.8 - (time_diff / request.flexibility_hours) * 0.3
|
|
else:
|
|
return max(0.0, 0.5 - (time_diff - request.flexibility_hours) * 0.1)
|
|
|
|
def _calculate_conflict_penalty(self, gene: Gene, all_genes: List[Gene]) -> float:
|
|
"""Check for time conflicts in the same classroom"""
|
|
conflicts = 0
|
|
for other in all_genes:
|
|
if gene != other and gene.classroom_id == other.classroom_id:
|
|
if self._times_overlap(gene.start_time, gene.end_time, other.start_time, other.end_time):
|
|
conflicts += 1
|
|
|
|
if conflicts > 0:
|
|
return -conflicts * 0.5 # Penalty for each conflict
|
|
return 1.0
|
|
|
|
def _calculate_resource_score(self, classroom: Classroom, requirements: Dict) -> float:
|
|
"""Check if classroom meets subject requirements"""
|
|
if not requirements:
|
|
return 1.0
|
|
|
|
# This would need to be implemented based on actual classroom resources
|
|
# For now, return a default score
|
|
return 0.8
|
|
|
|
def _times_overlap(self, start1: datetime, end1: datetime, start2: datetime, end2: datetime) -> bool:
|
|
"""Check if two time periods overlap"""
|
|
return start1 < end2 and end1 > start2
|
|
|
|
|
|
class GeneticAlgorithm:
|
|
"""Main genetic algorithm for room reservation optimization"""
|
|
|
|
def __init__(self, population_size: int = 50, generations: int = 100,
|
|
mutation_rate: float = 0.1, crossover_rate: float = 0.8):
|
|
self.population_size = population_size
|
|
self.generations = generations
|
|
self.mutation_rate = mutation_rate
|
|
self.crossover_rate = crossover_rate
|
|
|
|
def optimize_reservations(self, requests: List[ReservationRequest],
|
|
available_classrooms: List[Classroom],
|
|
start_date: datetime, end_date: datetime) -> Individual:
|
|
"""Run genetic algorithm to find optimal reservation schedule"""
|
|
classrooms_dict = {c.id: c for c in available_classrooms}
|
|
requests_dict = {r.commission_id: r for r in requests}
|
|
|
|
# Initialize population
|
|
population = self._initialize_population(requests, available_classrooms, start_date, end_date)
|
|
|
|
# Evolve population
|
|
for generation in range(self.generations):
|
|
# Calculate fitness for all individuals
|
|
for individual in population:
|
|
individual.calculate_fitness(classrooms_dict, requests_dict)
|
|
|
|
# Sort by fitness (best first)
|
|
population.sort(key=lambda x: x.fitness, reverse=True)
|
|
|
|
# Create new generation
|
|
new_population = []
|
|
|
|
# Elitism: keep best 10%
|
|
elite_size = int(self.population_size * 0.1)
|
|
new_population.extend(population[:elite_size])
|
|
|
|
# Crossover and mutation for remaining
|
|
while len(new_population) < self.population_size:
|
|
if random.random() < self.crossover_rate:
|
|
parent1 = self._tournament_selection(population)
|
|
parent2 = self._tournament_selection(population)
|
|
child1, child2 = self._crossover(parent1, parent2)
|
|
|
|
# Mutate children
|
|
if random.random() < self.mutation_rate:
|
|
child1 = self._mutate(child1, available_classrooms, start_date, end_date)
|
|
if random.random() < self.mutation_rate:
|
|
child2 = self._mutate(child2, available_classrooms, start_date, end_date)
|
|
|
|
new_population.extend([child1, child2])
|
|
else:
|
|
# Direct copy with potential mutation
|
|
selected = self._tournament_selection(population)
|
|
if random.random() < self.mutation_rate:
|
|
selected = self._mutate(selected, available_classrooms, start_date, end_date)
|
|
new_population.append(selected)
|
|
|
|
population = new_population[:self.population_size]
|
|
|
|
# Return best individual from final generation
|
|
population.sort(key=lambda x: x.fitness, reverse=True)
|
|
return population[0]
|
|
|
|
def _initialize_population(self, requests: List[ReservationRequest],
|
|
classrooms: List[Classroom],
|
|
start_date: datetime, end_date: datetime) -> List[Individual]:
|
|
"""Create initial population with random assignments"""
|
|
population = []
|
|
|
|
for _ in range(self.population_size):
|
|
genes = []
|
|
for request in requests:
|
|
# Random classroom selection (with capacity constraint)
|
|
suitable_classrooms = [c for c in classrooms if c.capacity >= request.expected_attendees]
|
|
if not suitable_classrooms:
|
|
continue
|
|
|
|
classroom = random.choice(suitable_classrooms)
|
|
|
|
# Random time slot (within flexible window)
|
|
time_offset = random.uniform(-request.flexibility_hours, request.flexibility_hours)
|
|
start_time = request.preferred_start_time + timedelta(hours=time_offset)
|
|
duration = request.preferred_end_time - request.preferred_start_time
|
|
end_time = start_time + duration
|
|
|
|
gene = Gene(
|
|
commission_id=request.commission_id,
|
|
classroom_id=classroom.id,
|
|
start_time=start_time,
|
|
end_time=end_time,
|
|
expected_attendees=request.expected_attendees,
|
|
purpose=request.purpose
|
|
)
|
|
genes.append(gene)
|
|
|
|
population.append(Individual(genes))
|
|
|
|
return population
|
|
|
|
def _tournament_selection(self, population: List[Individual], tournament_size: int = 3) -> Individual:
|
|
"""Select individual using tournament selection"""
|
|
tournament = random.sample(population, min(tournament_size, len(population)))
|
|
return max(tournament, key=lambda x: x.fitness)
|
|
|
|
def _crossover(self, parent1: Individual, parent2: Individual) -> Tuple[Individual, Individual]:
|
|
"""Perform crossover between two parents"""
|
|
# Simple one-point crossover
|
|
if len(parent1.genes) <= 1 or len(parent2.genes) <= 1:
|
|
return Individual(copy.deepcopy(parent1.genes)), Individual(copy.deepcopy(parent2.genes))
|
|
|
|
crossover_point = random.randint(1, min(len(parent1.genes), len(parent2.genes)) - 1)
|
|
|
|
child1_genes = copy.deepcopy(parent1.genes[:crossover_point] + parent2.genes[crossover_point:])
|
|
child2_genes = copy.deepcopy(parent2.genes[:crossover_point] + parent1.genes[crossover_point:])
|
|
|
|
return Individual(child1_genes), Individual(child2_genes)
|
|
|
|
def _mutate(self, individual: Individual, classrooms: List[Classroom],
|
|
start_date: datetime, end_date: datetime) -> Individual:
|
|
"""Apply mutation to an individual"""
|
|
if not individual.genes:
|
|
return individual
|
|
|
|
# Random gene mutation
|
|
gene_to_mutate = random.choice(individual.genes)
|
|
|
|
# Mutation types
|
|
mutation_type = random.choice(['classroom', 'time'])
|
|
|
|
if mutation_type == 'classroom':
|
|
# Change classroom
|
|
suitable_classrooms = [c for c in classrooms if c.capacity >= gene_to_mutate.expected_attendees]
|
|
if suitable_classrooms:
|
|
gene_to_mutate.classroom_id = random.choice(suitable_classrooms).id
|
|
|
|
elif mutation_type == 'time':
|
|
# Slightly adjust time
|
|
time_adjustment = timedelta(hours=random.uniform(-1, 1))
|
|
gene_to_mutate.start_time += time_adjustment
|
|
gene_to_mutate.end_time += time_adjustment
|
|
|
|
return individual
|
|
|
|
|
|
class ReservationOptimizer:
|
|
"""High-level interface for the reservation optimization system"""
|
|
|
|
def __init__(self):
|
|
self.ga = GeneticAlgorithm()
|
|
|
|
def optimize_schedule(self, commission_ids: List[int],
|
|
admin_user_id: int,
|
|
start_date: datetime = None,
|
|
end_date: datetime = None) -> Dict:
|
|
"""Optimize reservation schedule for given commissions"""
|
|
|
|
# Default date range if not provided
|
|
if not start_date:
|
|
start_date = datetime.utcnow()
|
|
if not end_date:
|
|
end_date = start_date + timedelta(days=30)
|
|
|
|
# Get commissions and create requests
|
|
commissions = Commission.query.filter(
|
|
Commission.id.in_(commission_ids),
|
|
Commission.active == True
|
|
).all()
|
|
|
|
requests = []
|
|
for commission in commissions:
|
|
# Calculate preferred time based on semester/year
|
|
preferred_time = self._calculate_preferred_time(commission)
|
|
|
|
request = ReservationRequest(
|
|
commission_id=commission.id,
|
|
expected_attendees=commission.max_students,
|
|
purpose=f"Regular class - {commission.subject.name if commission.subject else 'Unknown subject'}",
|
|
preferred_start_time=preferred_time['start'],
|
|
preferred_end_time=preferred_time['end'],
|
|
priority=1, # All regular classes have same priority
|
|
flexibility_hours=2
|
|
)
|
|
requests.append(request)
|
|
|
|
# Get available classrooms
|
|
classrooms = Classroom.query.filter_by(is_active=True).all()
|
|
|
|
# Run genetic algorithm
|
|
best_individual = self.ga.optimize_reservations(requests, classrooms, start_date, end_date)
|
|
|
|
# Convert to reservation format
|
|
optimized_reservations = []
|
|
for gene in best_individual.genes:
|
|
reservation_data = {
|
|
'commission_id': gene.commission_id,
|
|
'classroom_id': gene.classroom_id,
|
|
'user_id': admin_user_id,
|
|
'start_time': gene.start_time,
|
|
'end_time': gene.end_time,
|
|
'purpose': gene.purpose,
|
|
'expected_attendees': gene.expected_attendees,
|
|
'status': ReservationStatus.PENDING
|
|
}
|
|
optimized_reservations.append(reservation_data)
|
|
|
|
return {
|
|
'success': True,
|
|
'fitness_score': best_individual.fitness,
|
|
'reservations': optimized_reservations,
|
|
'conflicts': best_individual.conflicts,
|
|
'total_requests': len(requests),
|
|
'assigned_reservations': len(optimized_reservations)
|
|
}
|
|
|
|
def _calculate_preferred_time(self, commission: Commission) -> Dict:
|
|
"""Calculate preferred time slots for a commission"""
|
|
# This is a simplified version - would need to be based on actual semester schedule
|
|
current_date = datetime.utcnow()
|
|
|
|
# Default to weekday mornings for regular classes
|
|
days_ahead = (0 - current_date.weekday()) % 7 # Next Monday
|
|
if days_ahead == 0: # If today is Monday, use next week
|
|
days_ahead = 7
|
|
|
|
preferred_date = current_date + timedelta(days=days_ahead)
|
|
|
|
return {
|
|
'start': preferred_date.replace(hour=9, minute=0, second=0, microsecond=0),
|
|
'end': preferred_date.replace(hour=11, minute=0, second=0, microsecond=0)
|
|
}
|
|
|
|
def apply_optimized_reservations(self, reservation_data: List[Dict]) -> List[Reservation]:
|
|
"""Apply the optimized reservations to the database"""
|
|
applied_reservations = []
|
|
|
|
for data in reservation_data:
|
|
# Check for conflicts before creating
|
|
conflicts = Reservation.find_conflicts(
|
|
data['classroom_id'],
|
|
data['start_time'],
|
|
data['end_time']
|
|
)
|
|
|
|
if not conflicts:
|
|
reservation = Reservation(**data)
|
|
db.session.add(reservation)
|
|
applied_reservations.append(reservation)
|
|
else:
|
|
# Log conflict - could be handled differently
|
|
print(f"Conflict detected for reservation data: {data}")
|
|
|
|
try:
|
|
db.session.commit()
|
|
except Exception as e:
|
|
db.session.rollback()
|
|
raise e
|
|
|
|
return applied_reservations |