405 lines
14 KiB
Python
405 lines
14 KiB
Python
"""
|
|
Services découplés pour les opérations métier sur les évaluations.
|
|
|
|
Ce module applique les principes SOLID en séparant les responsabilités
|
|
de calcul, statistiques et progression qui étaient auparavant dans le modèle Assessment.
|
|
"""
|
|
from abc import ABC, abstractmethod
|
|
from typing import Dict, Any, List, Optional, Tuple, Protocol
|
|
from dataclasses import dataclass
|
|
from collections import defaultdict
|
|
import statistics
|
|
import math
|
|
|
|
# Type hints pour améliorer la lisibilité
|
|
StudentId = int
|
|
ExerciseId = int
|
|
GradingElementId = int
|
|
|
|
|
|
# =================== INTERFACES (Dependency Inversion Principle) ===================
|
|
|
|
class ConfigProvider(Protocol):
|
|
"""Interface pour l'accès à la configuration."""
|
|
|
|
def is_special_value(self, value: str) -> bool:
|
|
"""Vérifie si une valeur est spéciale (., d, etc.)"""
|
|
...
|
|
|
|
def get_special_values(self) -> Dict[str, Dict[str, Any]]:
|
|
"""Retourne la configuration des valeurs spéciales."""
|
|
...
|
|
|
|
|
|
class DatabaseProvider(Protocol):
|
|
"""Interface pour l'accès aux données."""
|
|
|
|
def get_grades_for_assessment(self, assessment_id: int) -> List[Any]:
|
|
"""Récupère toutes les notes d'une évaluation en une seule requête."""
|
|
...
|
|
|
|
def get_grading_elements_with_students(self, assessment_id: int) -> List[Any]:
|
|
"""Récupère les éléments de notation avec les étudiants associés."""
|
|
...
|
|
|
|
|
|
# =================== DATA TRANSFER OBJECTS ===================
|
|
|
|
@dataclass
|
|
class ProgressResult:
|
|
"""Résultat du calcul de progression."""
|
|
percentage: int
|
|
completed: int
|
|
total: int
|
|
status: str
|
|
students_count: int
|
|
|
|
|
|
@dataclass
|
|
class StudentScore:
|
|
"""Score d'un étudiant pour une évaluation."""
|
|
student_id: int
|
|
student_name: str
|
|
total_score: float
|
|
total_max_points: float
|
|
exercises: Dict[ExerciseId, Dict[str, Any]]
|
|
|
|
|
|
@dataclass
|
|
class StatisticsResult:
|
|
"""Résultat des calculs statistiques."""
|
|
count: int
|
|
mean: float
|
|
median: float
|
|
min: float
|
|
max: float
|
|
std_dev: float
|
|
|
|
|
|
# =================== STRATEGY PATTERN pour les types de notation ===================
|
|
|
|
class GradingStrategy(ABC):
|
|
"""Interface Strategy pour les différents types de notation."""
|
|
|
|
@abstractmethod
|
|
def calculate_score(self, grade_value: str, max_points: float) -> Optional[float]:
|
|
"""Calcule le score selon le type de notation."""
|
|
pass
|
|
|
|
@abstractmethod
|
|
def get_grading_type(self) -> str:
|
|
"""Retourne le type de notation."""
|
|
pass
|
|
|
|
|
|
class NotesStrategy(GradingStrategy):
|
|
"""Strategy pour la notation en points (notes)."""
|
|
|
|
def calculate_score(self, grade_value: str, max_points: float) -> Optional[float]:
|
|
try:
|
|
return float(grade_value)
|
|
except (ValueError, TypeError):
|
|
return 0.0
|
|
|
|
def get_grading_type(self) -> str:
|
|
return 'notes'
|
|
|
|
|
|
class ScoreStrategy(GradingStrategy):
|
|
"""Strategy pour la notation par compétences (score 0-3)."""
|
|
|
|
def calculate_score(self, grade_value: str, max_points: float) -> Optional[float]:
|
|
try:
|
|
score_int = int(grade_value)
|
|
if 0 <= score_int <= 3:
|
|
return (score_int / 3) * max_points
|
|
return 0.0
|
|
except (ValueError, TypeError):
|
|
return 0.0
|
|
|
|
def get_grading_type(self) -> str:
|
|
return 'score'
|
|
|
|
|
|
class GradingStrategyFactory:
|
|
"""Factory pour créer les strategies de notation."""
|
|
|
|
_strategies = {
|
|
'notes': NotesStrategy,
|
|
'score': ScoreStrategy
|
|
}
|
|
|
|
@classmethod
|
|
def create(cls, grading_type: str) -> GradingStrategy:
|
|
"""Crée une strategy selon le type."""
|
|
strategy_class = cls._strategies.get(grading_type)
|
|
if not strategy_class:
|
|
raise ValueError(f"Type de notation non supporté: {grading_type}")
|
|
return strategy_class()
|
|
|
|
@classmethod
|
|
def register_strategy(cls, grading_type: str, strategy_class: type):
|
|
"""Permet d'enregistrer de nouveaux types de notation."""
|
|
cls._strategies[grading_type] = strategy_class
|
|
|
|
|
|
# =================== SERVICES MÉTIER ===================
|
|
|
|
class UnifiedGradingCalculator:
|
|
"""
|
|
Calculateur unifié utilisant le pattern Strategy et l'injection de dépendances.
|
|
Remplace la classe GradingCalculator du modèle.
|
|
"""
|
|
|
|
def __init__(self, config_provider: ConfigProvider):
|
|
self.config_provider = config_provider
|
|
self._strategies = {}
|
|
|
|
def calculate_score(self, grade_value: str, grading_type: str, max_points: float) -> Optional[float]:
|
|
"""
|
|
Point d'entrée unifié pour tous les calculs de score.
|
|
Utilise l'injection de dépendances pour éviter les imports circulaires.
|
|
"""
|
|
# Valeurs spéciales en premier
|
|
if self.config_provider.is_special_value(grade_value):
|
|
special_config = self.config_provider.get_special_values()[grade_value]
|
|
special_value = special_config['value']
|
|
if special_value is None: # Dispensé
|
|
return None
|
|
return float(special_value) # 0 pour '.', etc.
|
|
|
|
# Utilisation du pattern Strategy
|
|
strategy = GradingStrategyFactory.create(grading_type)
|
|
return strategy.calculate_score(grade_value, max_points)
|
|
|
|
def is_counted_in_total(self, grade_value: str) -> bool:
|
|
"""Détermine si une note doit être comptée dans le total."""
|
|
if self.config_provider.is_special_value(grade_value):
|
|
special_config = self.config_provider.get_special_values()[grade_value]
|
|
return special_config['counts']
|
|
return True
|
|
|
|
|
|
class AssessmentProgressService:
|
|
"""
|
|
Service dédié au calcul de progression des notes.
|
|
Single Responsibility: calcul et formatage de la progression.
|
|
"""
|
|
|
|
def __init__(self, db_provider: DatabaseProvider):
|
|
self.db_provider = db_provider
|
|
|
|
def calculate_grading_progress(self, assessment) -> ProgressResult:
|
|
"""
|
|
Calcule la progression de saisie des notes pour une évaluation.
|
|
Optimisé pour éviter les requêtes N+1.
|
|
"""
|
|
total_students = len(assessment.class_group.students)
|
|
|
|
if total_students == 0:
|
|
return ProgressResult(
|
|
percentage=0,
|
|
completed=0,
|
|
total=0,
|
|
status='no_students',
|
|
students_count=0
|
|
)
|
|
|
|
# Requête optimisée : récupération en une seule fois
|
|
grading_elements_data = self.db_provider.get_grading_elements_with_students(assessment.id)
|
|
|
|
total_elements = 0
|
|
completed_elements = 0
|
|
|
|
for element_data in grading_elements_data:
|
|
total_elements += total_students
|
|
completed_elements += element_data['completed_grades_count']
|
|
|
|
if total_elements == 0:
|
|
return ProgressResult(
|
|
percentage=0,
|
|
completed=0,
|
|
total=0,
|
|
status='no_elements',
|
|
students_count=total_students
|
|
)
|
|
|
|
percentage = round((completed_elements / total_elements) * 100)
|
|
|
|
# Détermination du statut
|
|
status = self._determine_status(percentage)
|
|
|
|
return ProgressResult(
|
|
percentage=percentage,
|
|
completed=completed_elements,
|
|
total=total_elements,
|
|
status=status,
|
|
students_count=total_students
|
|
)
|
|
|
|
def _determine_status(self, percentage: int) -> str:
|
|
"""Détermine le statut basé sur le pourcentage."""
|
|
if percentage == 0:
|
|
return 'not_started'
|
|
elif percentage == 100:
|
|
return 'completed'
|
|
else:
|
|
return 'in_progress'
|
|
|
|
|
|
class StudentScoreCalculator:
|
|
"""
|
|
Service dédié au calcul des scores des étudiants.
|
|
Single Responsibility: calculs de notes avec logique métier.
|
|
"""
|
|
|
|
def __init__(self,
|
|
grading_calculator: UnifiedGradingCalculator,
|
|
db_provider: DatabaseProvider):
|
|
self.grading_calculator = grading_calculator
|
|
self.db_provider = db_provider
|
|
|
|
def calculate_student_scores(self, assessment) -> Tuple[Dict[StudentId, StudentScore], Dict[ExerciseId, Dict[StudentId, float]]]:
|
|
"""
|
|
Calcule les scores de tous les étudiants pour une évaluation.
|
|
Optimisé avec requête unique pour éviter N+1.
|
|
"""
|
|
# Requête optimisée : toutes les notes en une fois
|
|
grades_data = self.db_provider.get_grades_for_assessment(assessment.id)
|
|
|
|
# Organisation des données par étudiant et exercice
|
|
students_scores = {}
|
|
exercise_scores = defaultdict(lambda: defaultdict(float))
|
|
|
|
# Calcul pour chaque étudiant
|
|
for student in assessment.class_group.students:
|
|
student_score = self._calculate_single_student_score(
|
|
student, assessment, grades_data
|
|
)
|
|
students_scores[student.id] = student_score
|
|
|
|
# Mise à jour des scores par exercice
|
|
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 _calculate_single_student_score(self, student, assessment, grades_data) -> StudentScore:
|
|
"""Calcule le score d'un seul étudiant."""
|
|
total_score = 0
|
|
total_max_points = 0
|
|
student_exercises = {}
|
|
|
|
# Filtrage des notes pour cet étudiant
|
|
student_grades = {
|
|
grade['grading_element_id']: grade
|
|
for grade in grades_data
|
|
if grade['student_id'] == student.id
|
|
}
|
|
|
|
for exercise in assessment.exercises:
|
|
exercise_result = self._calculate_exercise_score(
|
|
exercise, student_grades
|
|
)
|
|
|
|
student_exercises[exercise.id] = exercise_result
|
|
total_score += exercise_result['score']
|
|
total_max_points += exercise_result['max_points']
|
|
|
|
return StudentScore(
|
|
student_id=student.id,
|
|
student_name=f"{student.first_name} {student.last_name}",
|
|
total_score=round(total_score, 2),
|
|
total_max_points=total_max_points,
|
|
exercises=student_exercises
|
|
)
|
|
|
|
def _calculate_exercise_score(self, exercise, student_grades) -> Dict[str, Any]:
|
|
"""Calcule le score pour un exercice spécifique."""
|
|
exercise_score = 0
|
|
exercise_max_points = 0
|
|
|
|
for element in exercise.grading_elements:
|
|
grade_data = student_grades.get(element.id)
|
|
|
|
if grade_data and grade_data['value'] and grade_data['value'] != '':
|
|
calculated_score = self.grading_calculator.calculate_score(
|
|
grade_data['value'].strip(),
|
|
element.grading_type,
|
|
element.max_points
|
|
)
|
|
|
|
if self.grading_calculator.is_counted_in_total(grade_data['value'].strip()):
|
|
if calculated_score is not None: # Pas dispensé
|
|
exercise_score += calculated_score
|
|
exercise_max_points += element.max_points
|
|
|
|
return {
|
|
'score': exercise_score,
|
|
'max_points': exercise_max_points,
|
|
'title': exercise.title
|
|
}
|
|
|
|
|
|
class AssessmentStatisticsService:
|
|
"""
|
|
Service dédié aux calculs statistiques.
|
|
Single Responsibility: analyses statistiques des résultats.
|
|
"""
|
|
|
|
def __init__(self, score_calculator: StudentScoreCalculator):
|
|
self.score_calculator = score_calculator
|
|
|
|
def get_assessment_statistics(self, assessment) -> StatisticsResult:
|
|
"""Calcule les statistiques descriptives pour une évaluation."""
|
|
students_scores, _ = self.score_calculator.calculate_student_scores(assessment)
|
|
scores = [score.total_score for score in students_scores.values()]
|
|
|
|
if not scores:
|
|
return StatisticsResult(
|
|
count=0,
|
|
mean=0,
|
|
median=0,
|
|
min=0,
|
|
max=0,
|
|
std_dev=0
|
|
)
|
|
|
|
return StatisticsResult(
|
|
count=len(scores),
|
|
mean=round(statistics.mean(scores), 2),
|
|
median=round(statistics.median(scores), 2),
|
|
min=min(scores),
|
|
max=max(scores),
|
|
std_dev=round(statistics.stdev(scores) if len(scores) > 1 else 0, 2)
|
|
)
|
|
|
|
|
|
# =================== FACADE pour simplifier l'utilisation ===================
|
|
|
|
class AssessmentServicesFacade:
|
|
"""
|
|
Facade qui regroupe tous les services pour faciliter l'utilisation.
|
|
Point d'entrée unique avec injection de dépendances.
|
|
"""
|
|
|
|
def __init__(self,
|
|
config_provider: ConfigProvider,
|
|
db_provider: DatabaseProvider):
|
|
# Création des services avec injection de dépendances
|
|
self.grading_calculator = UnifiedGradingCalculator(config_provider)
|
|
self.progress_service = AssessmentProgressService(db_provider)
|
|
self.score_calculator = StudentScoreCalculator(self.grading_calculator, db_provider)
|
|
self.statistics_service = AssessmentStatisticsService(self.score_calculator)
|
|
|
|
def get_grading_progress(self, assessment) -> ProgressResult:
|
|
"""Point d'entrée pour la progression."""
|
|
return self.progress_service.calculate_grading_progress(assessment)
|
|
|
|
def calculate_student_scores(self, assessment) -> Tuple[Dict[StudentId, StudentScore], Dict[ExerciseId, Dict[StudentId, float]]]:
|
|
"""Point d'entrée pour les scores étudiants."""
|
|
return self.score_calculator.calculate_student_scores(assessment)
|
|
|
|
def get_statistics(self, assessment) -> StatisticsResult:
|
|
"""Point d'entrée pour les statistiques."""
|
|
return self.statistics_service.get_assessment_statistics(assessment) |