207 lines
8.2 KiB
Python
207 lines
8.2 KiB
Python
from dataclasses import dataclass
|
|
from typing import List, Dict, Tuple
|
|
from collections import defaultdict
|
|
from models import db, Grade, GradingElement, Exercise
|
|
from repositories.temporal_student_repository import TemporalStudentRepository
|
|
from services.assessment_services import AssessmentProgressService, StudentScoreCalculator, ProgressResult, StudentScore
|
|
from providers.concrete_providers import SQLAlchemyDatabaseProvider
|
|
from sqlalchemy import and_, func
|
|
|
|
|
|
@dataclass
|
|
class TemporalProgressResult(ProgressResult):
|
|
"""Extension avec information temporelle."""
|
|
eligible_students_count: int
|
|
total_students_in_class: int
|
|
|
|
|
|
class TemporalAssessmentProgressService(AssessmentProgressService):
|
|
"""
|
|
Service de progression avec logique temporelle.
|
|
Hérite du service existant et ajoute la logique temporelle.
|
|
"""
|
|
|
|
def __init__(self, db_provider: SQLAlchemyDatabaseProvider, student_repo: TemporalStudentRepository):
|
|
super().__init__(db_provider)
|
|
self.student_repo = student_repo
|
|
|
|
def calculate_grading_progress(self, assessment) -> TemporalProgressResult:
|
|
"""
|
|
Calcule la progression en ne comptant que les élèves éligibles.
|
|
"""
|
|
# Étudiants éligibles à la date de l'évaluation
|
|
eligible_students = self.student_repo.find_eligible_for_assessment(assessment)
|
|
current_students = self.student_repo.find_current_students_in_class(assessment.class_group_id)
|
|
|
|
if not eligible_students:
|
|
return TemporalProgressResult(
|
|
percentage=0,
|
|
completed=0,
|
|
total=0,
|
|
status='no_eligible_students',
|
|
students_count=0,
|
|
eligible_students_count=0,
|
|
total_students_in_class=len(current_students)
|
|
)
|
|
|
|
# Calcul basé uniquement sur les élèves éligibles
|
|
eligible_student_ids = [s.id for s in eligible_students]
|
|
|
|
# Requête optimisée pour compter les notes des élèves éligibles uniquement
|
|
grading_elements_data = self._get_grading_elements_for_eligible_students(
|
|
assessment.id, eligible_student_ids
|
|
)
|
|
|
|
total_elements = 0
|
|
completed_elements = 0
|
|
|
|
for element_data in grading_elements_data:
|
|
total_elements += len(eligible_students)
|
|
completed_elements += element_data.completed_grades_count
|
|
|
|
if total_elements == 0:
|
|
return TemporalProgressResult(
|
|
percentage=0,
|
|
completed=0,
|
|
total=0,
|
|
status='no_elements',
|
|
students_count=len(eligible_students),
|
|
eligible_students_count=len(eligible_students),
|
|
total_students_in_class=len(current_students)
|
|
)
|
|
|
|
percentage = round((completed_elements / total_elements) * 100)
|
|
status = self._determine_status(percentage)
|
|
|
|
return TemporalProgressResult(
|
|
percentage=percentage,
|
|
completed=completed_elements,
|
|
total=total_elements,
|
|
status=status,
|
|
students_count=len(eligible_students),
|
|
eligible_students_count=len(eligible_students),
|
|
total_students_in_class=len(current_students)
|
|
)
|
|
|
|
def _get_grading_elements_for_eligible_students(self, assessment_id: int, eligible_student_ids: List[int]):
|
|
"""Requête optimisée pour récupérer les notes des élèves éligibles."""
|
|
|
|
return db.session.query(
|
|
GradingElement.id,
|
|
func.count(Grade.id).label('completed_grades_count')
|
|
).select_from(GradingElement)\
|
|
.join(Exercise, GradingElement.exercise_id == Exercise.id)\
|
|
.outerjoin(
|
|
Grade,
|
|
and_(
|
|
Grade.grading_element_id == GradingElement.id,
|
|
Grade.student_id.in_(eligible_student_ids),
|
|
Grade.value.isnot(None),
|
|
Grade.value != ''
|
|
)
|
|
)\
|
|
.filter(Exercise.assessment_id == assessment_id)\
|
|
.group_by(GradingElement.id)\
|
|
.all()
|
|
|
|
|
|
class TemporalStudentScoreCalculator(StudentScoreCalculator):
|
|
"""Calculateur de scores avec logique temporelle."""
|
|
|
|
def __init__(self,
|
|
grading_calculator,
|
|
db_provider: SQLAlchemyDatabaseProvider,
|
|
student_repo: TemporalStudentRepository):
|
|
super().__init__(grading_calculator, db_provider)
|
|
self.student_repo = student_repo
|
|
|
|
def calculate_student_scores(self, assessment) -> Tuple[Dict[int, StudentScore], Dict[int, Dict[int, float]]]:
|
|
"""
|
|
Calcule les scores uniquement pour les élèves éligibles.
|
|
"""
|
|
# Récupérer uniquement les élèves éligibles
|
|
eligible_students = self.student_repo.find_eligible_for_assessment(assessment)
|
|
|
|
if not eligible_students:
|
|
return {}, {}
|
|
|
|
# Requête optimisée pour les notes des élèves éligibles
|
|
eligible_student_ids = [s.id for s in eligible_students]
|
|
grades_data = self._get_grades_for_eligible_students(assessment.id, eligible_student_ids)
|
|
|
|
students_scores = {}
|
|
exercise_scores = defaultdict(lambda: defaultdict(float))
|
|
|
|
# Calcul pour chaque élève éligible
|
|
for student in eligible_students:
|
|
student_score = self._calculate_single_student_score(
|
|
student, assessment, grades_data
|
|
)
|
|
students_scores[student.id] = student_score
|
|
|
|
for exercise_id, exercise_data in student_score.exercises.items():
|
|
exercise_scores[exercise_id][student.id] = exercise_data['score']
|
|
|
|
return students_scores, dict(exercise_scores)
|
|
|
|
def _get_grades_for_eligible_students(self, assessment_id: int, eligible_student_ids: List[int]):
|
|
"""Requête optimisée pour les notes des élèves éligibles."""
|
|
|
|
grades = db.session.query(Grade)\
|
|
.join(GradingElement, Grade.grading_element_id == GradingElement.id)\
|
|
.join(Exercise, GradingElement.exercise_id == Exercise.id)\
|
|
.filter(
|
|
Exercise.assessment_id == assessment_id,
|
|
Grade.student_id.in_(eligible_student_ids)
|
|
).all()
|
|
|
|
# Convertir en format dict compatible avec le service parent
|
|
return [
|
|
{
|
|
'student_id': grade.student_id,
|
|
'grading_element_id': grade.grading_element_id,
|
|
'value': grade.value,
|
|
'grading_type': grade.grading_element.grading_type,
|
|
'max_points': grade.grading_element.max_points
|
|
}
|
|
for grade in grades
|
|
]
|
|
|
|
|
|
class TemporalClassStatisticsService:
|
|
"""Service pour calculer les statistiques de classe avec logique temporelle."""
|
|
|
|
def __init__(self, student_repo: TemporalStudentRepository):
|
|
self.student_repo = student_repo
|
|
|
|
def get_class_enrollment_summary(self, class_group_id: int, trimester: int = None):
|
|
"""Résumé des inscriptions pour une classe."""
|
|
current_students = self.student_repo.find_current_students_in_class(class_group_id)
|
|
|
|
# Calcul des mouvements si on a un trimestre spécifique
|
|
movements = {}
|
|
if trimester:
|
|
from datetime import date
|
|
# Approximation des dates de trimestre
|
|
trimester_dates = {
|
|
1: (date(2024, 9, 1), date(2024, 12, 31)),
|
|
2: (date(2025, 1, 1), date(2025, 3, 31)),
|
|
3: (date(2025, 4, 1), date(2025, 6, 30))
|
|
}
|
|
|
|
if trimester in trimester_dates:
|
|
start_date, end_date = trimester_dates[trimester]
|
|
students_with_movements = self.student_repo.find_students_with_movements_in_period(
|
|
start_date, end_date
|
|
)
|
|
movements = {
|
|
'arrivals': [],
|
|
'departures': [],
|
|
'total_movements': len(students_with_movements)
|
|
}
|
|
|
|
return {
|
|
'current_count': len(current_students),
|
|
'current_students': current_students,
|
|
'movements': movements
|
|
} |