Files
notytex/backend/domain/services/class_statistics_service.py

332 lines
14 KiB
Python

"""
Service de calcul des statistiques de classe.
Calcule les statistiques complètes pour le dashboard de classe:
- Moyennes par élève
- Scores par évaluation pour chaque élève
- Statistiques par domaine et compétence
"""
from typing import List, Dict, Optional, Tuple
from collections import defaultdict
from infrastructure.database.models import (
Student,
Assessment,
Exercise,
GradingElement,
Grade,
Domain,
Competence,
)
from domain.services.grading_calculator import GradingCalculator
from schemas.class_group import (
StudentAverage,
AssessmentScore,
DomainStudentStats,
CompetenceStudentStats,
DomainStats,
CompetenceStats,
)
class ClassStatisticsService:
"""Service de calcul des statistiques de classe."""
def __init__(self):
self.calculator = GradingCalculator()
async def calculate_student_statistics(
self,
students: List[Student],
assessments: List[Assessment],
grades_by_student_assessment: Dict[Tuple[int, int], List[Tuple[Grade, GradingElement]]],
domains: List[Domain],
competences: List[Competence],
) -> List[StudentAverage]:
"""
Calcule les statistiques complètes pour chaque élève.
Args:
students: Liste des élèves
assessments: Liste des évaluations du trimestre
grades_by_student_assessment: Dict[(student_id, assessment_id)] -> [(grade, element)]
domains: Liste des domaines
competences: Liste des compétences
Returns:
Liste des StudentAverage avec toutes les statistiques
"""
student_averages = []
for student in students:
# Initialiser les statistiques par domaine/compétence
domain_stats: Dict[int, DomainStudentStats] = {
domain.id: DomainStudentStats(
domain_id=domain.id,
evaluation_count=0,
total_points_obtained=0.0,
total_points_possible=0.0,
)
for domain in domains
}
competence_stats: Dict[int, CompetenceStudentStats] = {}
# Calculer les scores par évaluation
assessment_scores: Dict[int, AssessmentScore] = {}
weighted_sum = 0.0
total_coefficient = 0.0
assessment_count = 0
for assessment in assessments:
grades_data = grades_by_student_assessment.get((student.id, assessment.id), [])
if not grades_data:
continue
# Calculer le score total pour cette évaluation
total_score = 0.0
total_max_points = 0.0
for grade, element in grades_data:
if grade.value:
score = self.calculator.calculate_score(
grade.value, element.grading_type, element.max_points
)
if score is not None and self.calculator.is_counted_in_total(grade.value):
total_score += score
total_max_points += element.max_points
# Statistiques par domaine
if element.domain_id and element.domain_id in domain_stats:
domain_stats[element.domain_id].evaluation_count += 1
domain_stats[element.domain_id].total_points_obtained += score
domain_stats[element.domain_id].total_points_possible += element.max_points
# Statistiques par compétence (skill)
# Note: On utilise element.skill pour identifier la compétence
if element.skill:
# Trouver la compétence correspondante
matching_competence = next(
(c for c in competences if c.name == element.skill),
None
)
if matching_competence:
if matching_competence.id not in competence_stats:
competence_stats[matching_competence.id] = CompetenceStudentStats(
competence_id=matching_competence.id,
evaluation_count=0,
total_points_obtained=0.0,
total_points_possible=0.0,
)
competence_stats[matching_competence.id].evaluation_count += 1
competence_stats[matching_competence.id].total_points_obtained += score
competence_stats[matching_competence.id].total_points_possible += element.max_points
# Calculer le score sur 20
score_on_20 = None
if total_max_points > 0:
score_on_20 = round(total_score / total_max_points * 20, 2)
weighted_sum += score_on_20 * assessment.coefficient
total_coefficient += assessment.coefficient
assessment_count += 1
# Sauvegarder le score de cette évaluation
assessment_scores[assessment.id] = AssessmentScore(
assessment_id=assessment.id,
assessment_title=assessment.title,
score=round(total_score, 2) if total_score > 0 else None,
max_points=round(total_max_points, 2),
score_on_20=score_on_20,
)
# Calculer la moyenne pondérée
average = None
if total_coefficient > 0:
average = round(weighted_sum / total_coefficient, 2)
student_averages.append(StudentAverage(
student_id=student.id,
first_name=student.first_name,
last_name=student.last_name,
full_name=f"{student.first_name} {student.last_name}",
average=average,
assessment_count=assessment_count,
assessment_scores=assessment_scores,
domain_stats=domain_stats,
competence_stats=competence_stats,
))
return student_averages
def aggregate_domain_competence_stats(
self,
student_averages: List[StudentAverage],
domains: List[Domain],
competences: List[Competence],
) -> Tuple[List[DomainStats], List[CompetenceStats]]:
"""
Agrège les statistiques par domaine et compétence pour tous les élèves.
Args:
student_averages: Liste des statistiques par élève
domains: Liste des domaines
competences: Liste des compétences
Returns:
Tuple (domains_stats, competences_stats)
"""
# Agréger par domaine
domain_aggregates: Dict[int, Dict] = defaultdict(
lambda: {
"evaluation_count": 0,
"total_points_obtained": 0.0,
"total_points_possible": 0.0,
}
)
for student in student_averages:
for domain_id, stats in student.domain_stats.items():
domain_aggregates[domain_id]["evaluation_count"] += stats.evaluation_count
domain_aggregates[domain_id]["total_points_obtained"] += stats.total_points_obtained
domain_aggregates[domain_id]["total_points_possible"] += stats.total_points_possible
domains_stats = []
for domain in domains:
agg = domain_aggregates.get(domain.id, {
"evaluation_count": 0,
"total_points_obtained": 0.0,
"total_points_possible": 0.0,
})
domains_stats.append(DomainStats(
id=domain.id,
name=domain.name,
color=domain.color,
evaluation_count=agg["evaluation_count"],
total_points_obtained=round(agg["total_points_obtained"], 2),
total_points_possible=round(agg["total_points_possible"], 2),
))
# Agréger par compétence
competence_aggregates: Dict[int, Dict] = defaultdict(
lambda: {
"evaluation_count": 0,
"total_points_obtained": 0.0,
"total_points_possible": 0.0,
}
)
for student in student_averages:
for competence_id, stats in student.competence_stats.items():
competence_aggregates[competence_id]["evaluation_count"] += stats.evaluation_count
competence_aggregates[competence_id]["total_points_obtained"] += stats.total_points_obtained
competence_aggregates[competence_id]["total_points_possible"] += stats.total_points_possible
competences_stats = []
for competence in competences:
agg = competence_aggregates.get(competence.id, {
"evaluation_count": 0,
"total_points_obtained": 0.0,
"total_points_possible": 0.0,
})
competences_stats.append(CompetenceStats(
id=competence.id,
name=competence.name,
color=competence.color,
evaluation_count=agg["evaluation_count"],
total_points_obtained=round(agg["total_points_obtained"], 2),
total_points_possible=round(agg["total_points_possible"], 2),
))
return domains_stats, competences_stats
def calculate_domain_competence_from_elements(
self,
assessments: List[Assessment],
domains: List[Domain],
competences: List[Competence],
) -> Tuple[List[DomainStats], List[CompetenceStats]]:
"""
Calcule les statistiques domaines/compétences depuis les GradingElements.
Perspective enseignant : ce qui a été évalué, pas les résultats des élèves.
Args:
assessments: Liste des évaluations (avec exercises et grading_elements chargés)
domains: Liste des domaines
competences: Liste des compétences
Returns:
Tuple (domains_stats, competences_stats)
"""
# Compter les GradingElements par domaine
domain_aggregates: Dict[int, Dict] = defaultdict(
lambda: {
"evaluation_count": 0,
"total_points_possible": 0.0,
}
)
competence_aggregates: Dict[int, Dict] = defaultdict(
lambda: {
"evaluation_count": 0,
"total_points_possible": 0.0,
}
)
# Parcourir tous les éléments de notation
for assessment in assessments:
for exercise in assessment.exercises:
for element in exercise.grading_elements:
# Compter par domaine
if element.domain_id:
domain_aggregates[element.domain_id]["evaluation_count"] += 1
domain_aggregates[element.domain_id]["total_points_possible"] += element.max_points
# Compter par compétence (via skill)
if element.skill:
matching_competence = next(
(c for c in competences if c.name == element.skill),
None
)
if matching_competence:
competence_aggregates[matching_competence.id]["evaluation_count"] += 1
competence_aggregates[matching_competence.id]["total_points_possible"] += element.max_points
# Créer les stats par domaine
domains_stats = []
for domain in domains:
agg = domain_aggregates.get(domain.id, {
"evaluation_count": 0,
"total_points_possible": 0.0,
})
domains_stats.append(DomainStats(
id=domain.id,
name=domain.name,
color=domain.color,
evaluation_count=agg["evaluation_count"],
total_points_obtained=0.0, # Non utilisé dans cette perspective
total_points_possible=round(agg["total_points_possible"], 2),
))
# Créer les stats par compétence
competences_stats = []
for competence in competences:
agg = competence_aggregates.get(competence.id, {
"evaluation_count": 0,
"total_points_possible": 0.0,
})
competences_stats.append(CompetenceStats(
id=competence.id,
name=competence.name,
color=competence.color,
evaluation_count=agg["evaluation_count"],
total_points_obtained=0.0, # Non utilisé
total_points_possible=round(agg["total_points_possible"], 2),
))
return domains_stats, competences_stats