fix contract
This commit is contained in:
Alejandro Vazquez
+15
-4
@@ -1,6 +1,17 @@
|
||||
from .user import User
|
||||
from .classroom import Classroom
|
||||
from .subject import Subject
|
||||
from .reservation import Reservation
|
||||
from .classroom import Classroom, ClassroomResource
|
||||
from .subject import Subject, Commission
|
||||
from .reservation import Reservation, ReservationStatus
|
||||
from .genetic_algorithm import GeneticAlgorithm, ReservationOptimizer
|
||||
|
||||
__all__ = ['User', 'Classroom', 'Subject', 'Reservation']
|
||||
__all__ = [
|
||||
'User',
|
||||
'Classroom',
|
||||
'ClassroomResource',
|
||||
'Subject',
|
||||
'Commission',
|
||||
'Reservation',
|
||||
'ReservationStatus',
|
||||
'GeneticAlgorithm',
|
||||
'ReservationOptimizer'
|
||||
]
|
||||
@@ -0,0 +1,397 @@
|
||||
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
|
||||
Reference in New Issue
Block a user