DocScoDoc/app/comp/moy_mod.py

# -*- mode: python -*-
# -*- coding: utf-8 -*-

##############################################################################
#
# Gestion scolarite IUT
#
# Copyright (c) 1999 - 2022 Emmanuel Viennet.  All rights reserved.
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
#
#   Emmanuel Viennet      emmanuel.viennet@viennet.net
#
##############################################################################

"""Fonctions de calcul des moyennes de modules (modules, ressources ou SAÉ)

Pour les formations classiques et le BUT

Rappel: pour éviter les confusions, on appelera *poids* les coefficients d'une
évaluation dans un module, et *coefficients* ceux utilisés pour le calcul de la
moyenne générale d'une UE.
"""
import dataclasses
from dataclasses import dataclass

import numpy as np
import pandas as pd

from app import db
from app.models import Evaluation, EvaluationUEPoids, ModuleImpl
from app.scodoc import sco_cache
from app.scodoc import sco_utils as scu
from app.scodoc.sco_codes_parcours import UE_SPORT
from app.scodoc.sco_exceptions import ScoBugCatcher
from app.scodoc.sco_utils import ModuleType


@dataclass
class EvaluationEtat:
    """Classe pour stocker quelques infos sur les résultats d'une évaluation"""

    evaluation_id: int
    nb_attente: int
    is_complete: bool

    def to_dict(self):
        "convert to dict"
        return dataclasses.asdict(self)


class ModuleImplResults:
    """Classe commune à toutes les formations (standard et APC).
    Les notes des étudiants d'un moduleimpl.
    Les poids des évals sont à part car on en a besoin sans les notes pour les
    tableaux de bord.
    Les attributs sont tous des objets simples cachables dans Redis;
    les caches sont gérés par  ResultatsSemestre.
    """

    def __init__(self, moduleimpl: ModuleImpl):
        self.moduleimpl_id = moduleimpl.id
        self.module_id = moduleimpl.module.id
        self.etudids = None
        "liste des étudiants inscrits au SEMESTRE (incluant dem et def)"

        self.nb_inscrits_module = None
        "nombre d'inscrits (non DEM) à ce module"
        self.evaluations_completes = []
        "séquence de booléens, indiquant les évals à prendre en compte."
        self.evaluations_completes_dict = {}
        "{ evaluation.id : bool } indique si à prendre en compte ou non."
        self.evaluations_etat = {}
        "{ evaluation_id: EvaluationEtat }"
        self.en_attente = False
        "Vrai si au moins une évaluation a une note en attente"
        #
        self.evals_notes = None
        """DataFrame, colonnes: EVALS, Lignes: etudid (inscrits au SEMESTRE)
            valeur: notes brutes, float ou NOTES_ATTENTE, NOTES_NEUTRALISE,
            NOTES_ABSENCE.
            Les NaN désignent les notes manquantes (non saisies).
        """
        self.etuds_moy_module = None
        """DataFrame, colonnes UE, lignes etud
            = la note de l'étudiant dans chaque UE pour ce module.
            ou NaN si les évaluations (dans lesquelles l'étudiant a des notes)
            ne donnent pas de coef vers cette UE.
        """
        self.evals_etudids_sans_note = {}
        """dict: evaluation_id : set des etudids non notés dans cette eval, sans les démissions."""
        self.load_notes()
        self.etuds_use_session2 = pd.Series(False, index=self.evals_notes.index)
        """1 bool par etud, indique si sa moyenne de module vient de la session2"""
        self.etuds_use_rattrapage = pd.Series(False, index=self.evals_notes.index)
        """1 bool par etud, indique si sa moyenne de module utilise la note de rattrapage"""

    def load_notes(self):  # ré-écriture de df_load_modimpl_notes
        """Charge toutes les notes de toutes les évaluations du module.
        Dataframe evals_notes
            colonnes: le nom de la colonne est l'evaluation_id (int)
            index (lignes): etudid (int)

        L'ensemble des étudiants est celui des inscrits au SEMESTRE.

        Les notes sont "brutes" (séries de floats) et peuvent prendre les valeurs:
            note : float (valeur enregistrée brute, NON normalisée sur 20)
            pas de note: NaN (rien en bd, ou étudiant non inscrit au module)
            absent: NOTES_ABSENCE (NULL en bd)
            excusé: NOTES_NEUTRALISE (voir sco_utils)
            attente: NOTES_ATTENTE

        Évaluation "complete" (prise en compte dans les calculs) si:
        - soit tous les étudiants inscrits au module ont des notes
        - soit elle a été déclarée "à prise en compte immédiate" (publish_incomplete)
            ou est une évaluation de rattrapage ou de session 2
        Évaluation "attente" (prise en compte dans les calculs, mais il y
        manque des notes) ssi il y a des étudiants inscrits au semestre et au module
        qui ont des notes ATT.
        """
        moduleimpl = ModuleImpl.query.get(self.moduleimpl_id)
        self.etudids = self._etudids()

        # --- Calcul nombre d'inscrits pour déterminer les évaluations "completes":
        # on prend les inscrits au module ET au semestre (donc sans démissionnaires)
        inscrits_module = {ins.etud.id for ins in moduleimpl.inscriptions}.intersection(
            moduleimpl.formsemestre.etudids_actifs
        )
        self.nb_inscrits_module = len(inscrits_module)

        # dataFrame vide, index = tous les inscrits au SEMESTRE
        evals_notes = pd.DataFrame(index=self.etudids, dtype=float)
        self.evaluations_completes = []
        self.evaluations_completes_dict = {}
        self.en_attente = False
        for evaluation in moduleimpl.evaluations:
            eval_df = self._load_evaluation_notes(evaluation)
            # is_complete ssi tous les inscrits (non dem) au semestre ont une note
            # ou évaluation déclarée "à prise en compte immédiate"
            # Les évaluations de rattrapage et 2eme session sont toujours complètes

            etudids_sans_note = inscrits_module - set(eval_df.index)  # sans les dem.
            is_complete = (
                (evaluation.evaluation_type == scu.EVALUATION_RATTRAPAGE)
                or (evaluation.evaluation_type == scu.EVALUATION_SESSION2)
                or (evaluation.publish_incomplete)
                or (not etudids_sans_note)
            )
            self.evaluations_completes.append(is_complete)
            self.evaluations_completes_dict[evaluation.id] = is_complete
            self.evals_etudids_sans_note[evaluation.id] = etudids_sans_note

            # NULL en base => ABS (= -999)
            eval_df.fillna(scu.NOTES_ABSENCE, inplace=True)
            # Ce merge ne garde que les étudiants inscrits au module
            # et met à NULL les notes non présentes
            #  (notes non saisies ou etuds non inscrits au module):
            evals_notes = evals_notes.merge(
                eval_df, how="left", left_index=True, right_index=True
            )
            # Notes en attente: (ne prend en compte que les inscrits, non démissionnaires)
            nb_att = sum(
                evals_notes[str(evaluation.id)][list(inscrits_module)]
                == scu.NOTES_ATTENTE
            )
            self.evaluations_etat[evaluation.id] = EvaluationEtat(
                evaluation_id=evaluation.id, nb_attente=nb_att, is_complete=is_complete
            )
            if nb_att > 0:
                self.en_attente = True

        # Force columns names to integers (evaluation ids)
        evals_notes.columns = pd.Index([int(x) for x in evals_notes.columns], dtype=int)
        self.evals_notes = evals_notes

    def _load_evaluation_notes(self, evaluation: Evaluation) -> pd.DataFrame:
        """Charge les notes de l'évaluation
        Resultat: dataframe, index: etudid ayant une note, valeur: note brute.
        """
        eval_df = pd.read_sql_query(
            """SELECT n.etudid, n.value AS "%(evaluation_id)s"
            FROM notes_notes n, notes_moduleimpl_inscription i
            WHERE evaluation_id=%(evaluation_id)s
            AND n.etudid = i.etudid
            AND i.moduleimpl_id = %(moduleimpl_id)s
            """,
            db.engine,
            params={
                "evaluation_id": evaluation.id,
                "moduleimpl_id": evaluation.moduleimpl.id,
            },
            index_col="etudid",
        )
        eval_df[str(evaluation.id)] = pd.to_numeric(eval_df[str(evaluation.id)])
        return eval_df

    def _etudids(self):
        """L'index du dataframe est la liste de tous les étudiants inscrits au semestre
        (incluant les DEM et DEF)
        """
        return [
            inscr.etudid
            for inscr in ModuleImpl.query.get(
                self.moduleimpl_id
            ).formsemestre.inscriptions
        ]

    def get_evaluations_coefs(self, moduleimpl: ModuleImpl) -> np.array:
        """Coefficients des évaluations.
        Les coefs des évals incomplètes et non "normales" (session 2, rattrapage)
        sont zéro.
        Résultat: 2d-array of floats, shape (nb_evals, 1)
        """
        return (
            np.array(
                [
                    e.coefficient
                    if e.evaluation_type == scu.EVALUATION_NORMALE
                    else 0.0
                    for e in moduleimpl.evaluations
                ],
                dtype=float,
            )
            * self.evaluations_completes
        ).reshape(-1, 1)

    # was _list_notes_evals_titles
    def get_evaluations_completes(self, moduleimpl: ModuleImpl) -> list:
        "Liste des évaluations complètes"
        return [
            e for e in moduleimpl.evaluations if self.evaluations_completes_dict[e.id]
        ]

    def get_eval_notes_sur_20(self, moduleimpl: ModuleImpl) -> np.array:
        """Les notes de toutes les évaluations du module, complètes ou non.
        Remplace les ATT, EXC, ABS, NaN par zéro et mets les notes sur 20.
        Résultat: 2d array of floats, shape nb_etuds x nb_evaluations
        """
        return np.where(
            self.evals_notes.values > scu.NOTES_ABSENCE, self.evals_notes.values, 0.0
        ) / [e.note_max / 20.0 for e in moduleimpl.evaluations]

    def get_eval_notes_dict(self, evaluation_id: int) -> dict:
        """Notes d'une évaulation, brutes, sous forme d'un dict
        { etudid : valeur }
        avec les valeurs float, ou "ABS" ou EXC
        """
        return {
            etudid: scu.fmt_note(x, keep_numeric=True)
            for (etudid, x) in self.evals_notes[evaluation_id].items()
        }

    def get_evaluation_rattrapage(self, moduleimpl: ModuleImpl):
        """L'évaluation de rattrapage de ce module, ou None s'il n'en a pas.
        Rattrapage: la moyenne du module est la meilleure note entre moyenne
        des autres évals et la note eval rattrapage.
        """
        eval_list = [
            e
            for e in moduleimpl.evaluations
            if e.evaluation_type == scu.EVALUATION_RATTRAPAGE
        ]
        if eval_list:
            return eval_list[0]
        return None

    def get_evaluation_session2(self, moduleimpl: ModuleImpl):
        """L'évaluation de deuxième session de ce module, ou None s'il n'en a pas.
        Session 2: remplace la note de moyenne des autres évals.
        """
        eval_list = [
            e
            for e in moduleimpl.evaluations
            if e.evaluation_type == scu.EVALUATION_SESSION2
        ]
        if eval_list:
            return eval_list[0]
        return None


class ModuleImplResultsAPC(ModuleImplResults):
    "Calcul des moyennes de modules à la mode BUT"

    def compute_module_moy(
        self,
        evals_poids_df: pd.DataFrame,
    ) -> pd.DataFrame:
        """Calcule les moyennes des étudiants dans ce module

        Argument: evals_poids: DataFrame, colonnes: UEs, Lignes: EVALs

        Résultat: DataFrame, colonnes UE, lignes etud
            = la note de l'étudiant dans chaque UE pour ce module.
            ou NaN si les évaluations (dans lesquelles l'étudiant a des notes)
            ne donnent pas de coef vers cette UE.
        """
        modimpl = ModuleImpl.query.get(self.moduleimpl_id)
        nb_etuds, nb_evals = self.evals_notes.shape
        nb_ues = evals_poids_df.shape[1]
        assert evals_poids_df.shape[0] == nb_evals  # compat notes/poids
        if nb_etuds == 0:
            return pd.DataFrame(index=[], columns=evals_poids_df.columns)
        if nb_ues == 0:
            return pd.DataFrame(index=self.evals_notes.index, columns=[])
        evals_coefs = self.get_evaluations_coefs(modimpl)
        evals_poids = evals_poids_df.values * evals_coefs
        # -> evals_poids shape : (nb_evals, nb_ues)
        assert evals_poids.shape == (nb_evals, nb_ues)
        evals_notes_20 = self.get_eval_notes_sur_20(modimpl)

        # Les poids des évals pour chaque étudiant: là où il a des notes
        # non neutralisées
        # (ABS n'est  pas neutralisée, mais ATTENTE et NEUTRALISE oui)
        # Note: les NaN sont remplacés par des 0 dans evals_notes
        #  et dans dans evals_poids_etuds
        #  (rappel: la comparaison est toujours false face à un NaN)
        # shape: (nb_etuds, nb_evals, nb_ues)
        poids_stacked = np.stack([evals_poids] * nb_etuds)
        evals_poids_etuds = np.where(
            np.stack([self.evals_notes.values] * nb_ues, axis=2) > scu.NOTES_NEUTRALISE,
            poids_stacked,
            0,
        )
        # Calcule la moyenne pondérée sur les notes disponibles:
        evals_notes_stacked = np.stack([evals_notes_20] * nb_ues, axis=2)
        with np.errstate(invalid="ignore"):  # ignore les 0/0 (-> NaN)
            etuds_moy_module = np.sum(
                evals_poids_etuds * evals_notes_stacked, axis=1
            ) / np.sum(evals_poids_etuds, axis=1)

        # Session2 : quand elle existe, remplace la note de module
        eval_session2 = self.get_evaluation_session2(modimpl)
        if eval_session2:
            notes_session2 = self.evals_notes[eval_session2.id].values
            # n'utilise que les notes valides (pas ATT, EXC, ABS, NaN)
            etuds_use_session2 = notes_session2 > scu.NOTES_ABSENCE
            etuds_moy_module = np.where(
                etuds_use_session2[:, np.newaxis],
                np.tile(
                    (notes_session2 / (eval_session2.note_max / 20.0))[:, np.newaxis],
                    nb_ues,
                ),
                etuds_moy_module,
            )
            self.etuds_use_session2 = pd.Series(
                etuds_use_session2, index=self.evals_notes.index
            )
        else:
            # Rattrapage: remplace la note de module ssi elle est supérieure
            eval_rat = self.get_evaluation_rattrapage(modimpl)
            if eval_rat:
                notes_rat = self.evals_notes[eval_rat.id].values
                # remplace les notes invalides (ATT, EXC...) par des NaN
                notes_rat = np.where(
                    notes_rat > scu.NOTES_ABSENCE,
                    notes_rat / (eval_rat.note_max / 20.0),
                    np.nan,
                )
                # "Étend" le rattrapage sur les UE: la note de rattrapage est la même
                # pour toutes les UE mais ne remplace que là où elle est supérieure
                notes_rat_ues = np.stack([notes_rat] * nb_ues, axis=1)
                # prend le max
                etuds_use_rattrapage = notes_rat_ues > etuds_moy_module
                etuds_moy_module = np.where(
                    etuds_use_rattrapage, notes_rat_ues, etuds_moy_module
                )
                # Serie indiquant que l'étudiant utilise une note de rattrapage sur l'une des UE:
                self.etuds_use_rattrapage = pd.Series(
                    etuds_use_rattrapage.any(axis=1), index=self.evals_notes.index
                )
        self.etuds_moy_module = pd.DataFrame(
            etuds_moy_module,
            index=self.evals_notes.index,
            columns=evals_poids_df.columns,
        )
        return self.etuds_moy_module


def load_evaluations_poids(moduleimpl_id: int) -> tuple[pd.DataFrame, list]:
    """Charge poids des évaluations d'un module et retourne un dataframe
    rows = evaluations, columns = UE, value = poids (float).
    Les valeurs manquantes (évaluations sans coef vers des UE) sont
    remplies: 1 si le coef de ce module dans l'UE est non nul, zéro sinon
    (sauf pour module bonus, defaut à 1)

    Si le module n'est pas une ressource ou une SAE, ne charge pas de poids
    et renvoie toujours les poids par défaut.

    Résultat: (evals_poids, liste de UEs du semestre sauf le sport)
    """
    modimpl: ModuleImpl = ModuleImpl.query.get(moduleimpl_id)
    evaluations = Evaluation.query.filter_by(moduleimpl_id=moduleimpl_id).all()
    ues = modimpl.formsemestre.query_ues(with_sport=False).all()
    ue_ids = [ue.id for ue in ues]
    evaluation_ids = [evaluation.id for evaluation in evaluations]
    evals_poids = pd.DataFrame(columns=ue_ids, index=evaluation_ids, dtype=float)
    if (
        modimpl.module.module_type == ModuleType.RESSOURCE
        or modimpl.module.module_type == ModuleType.SAE
    ):
        for ue_poids in EvaluationUEPoids.query.join(
            EvaluationUEPoids.evaluation
        ).filter_by(moduleimpl_id=moduleimpl_id):
            try:
                evals_poids[ue_poids.ue_id][ue_poids.evaluation_id] = ue_poids.poids
            except KeyError as exc:
                pass  # poids vers des UE qui n'existent plus ou sont dans un autre semestre...

    # Initialise poids non enregistrés:
    default_poids = (
        1.0
        if modimpl.module.ue.type == UE_SPORT
        or modimpl.module.module_type == ModuleType.MALUS
        else 0.0
    )

    if np.isnan(evals_poids.values.flat).any():
        ue_coefs = modimpl.module.get_ue_coef_dict()
        for ue in ues:
            evals_poids[ue.id][evals_poids[ue.id].isna()] = (
                1 if ue_coefs.get(ue.id, default_poids) > 0 else 0
            )

    return evals_poids, ues


def moduleimpl_is_conforme(
    moduleimpl, evals_poids: pd.DataFrame, modules_coefficients: pd.DataFrame
) -> bool:
    """Vérifie que les évaluations de ce moduleimpl sont bien conformes
    au PN.
    Un module est dit *conforme* si et seulement si la somme des poids de ses
    évaluations vers une UE de coefficient non nul est non nulle.

    Arguments:
        evals_poids: DataFrame, colonnes: UEs, Lignes: EVALs
        modules_coefficients: DataFrame, cols module_id, lignes UEs
    NB: les UEs dans evals_poids sont sans le bonus sport
    """
    nb_evals, nb_ues = evals_poids.shape
    if nb_evals == 0:
        return True  # modules vides conformes
    if nb_ues == 0:
        return False  # situation absurde (pas d'UE)
    if len(modules_coefficients) != nb_ues:
        # il arrive (#bug) que le cache ne soit pas à jour...
        sco_cache.invalidate_formsemestre()
        raise ScoBugCatcher("moduleimpl_is_conforme: nb ue incoherent")

    if moduleimpl.module_id not in modules_coefficients:
        # soupçon de bug cache coef ?
        sco_cache.invalidate_formsemestre()
        raise ScoBugCatcher("Erreur 454 - merci de ré-essayer")

    module_evals_poids = evals_poids.transpose().sum(axis=1) != 0
    return all((modules_coefficients[moduleimpl.module_id] != 0).eq(module_evals_poids))


class ModuleImplResultsClassic(ModuleImplResults):
    "Calcul des moyennes de modules des formations classiques"

    def compute_module_moy(self) -> pd.Series:
        """Calcule les moyennes des étudiants dans ce module

        Résultat: Series, lignes etud
            = la note (moyenne) de l'étudiant pour ce module.
            ou NaN si les évaluations (dans lesquelles l'étudiant a des notes)
            ne donnent pas de coef.
        """
        modimpl = ModuleImpl.query.get(self.moduleimpl_id)
        nb_etuds, nb_evals = self.evals_notes.shape
        if nb_etuds == 0:
            return pd.Series()
        evals_coefs = self.get_evaluations_coefs(modimpl).reshape(-1)
        assert evals_coefs.shape == (nb_evals,)
        evals_notes_20 = self.get_eval_notes_sur_20(modimpl)
        # Les coefs des évals pour chaque étudiant: là où il a des notes
        #  non neutralisées
        # (ABS n'est  pas neutralisée, mais ATTENTE et NEUTRALISE oui)
        # Note: les NaN sont remplacés par des 0 dans evals_notes
        #  et dans dans evals_poids_etuds
        #  (rappel: la comparaison est toujours False face à un NaN)
        # shape: (nb_etuds, nb_evals)
        coefs_stacked = np.stack([evals_coefs] * nb_etuds)
        evals_coefs_etuds = np.where(
            self.evals_notes.values > scu.NOTES_NEUTRALISE, coefs_stacked, 0
        )
        # Calcule la moyenne pondérée sur les notes disponibles:
        with np.errstate(invalid="ignore"):  # ignore les 0/0 (-> NaN)
            etuds_moy_module = np.sum(
                evals_coefs_etuds * evals_notes_20, axis=1
            ) / np.sum(evals_coefs_etuds, axis=1)

        # Session2 : quand elle existe, remplace la note de module
        eval_session2 = self.get_evaluation_session2(modimpl)
        if eval_session2:
            notes_session2 = self.evals_notes[eval_session2.id].values
            # n'utilise que les notes valides (pas ATT, EXC, ABS, NaN)
            etuds_use_session2 = notes_session2 > scu.NOTES_ABSENCE
            etuds_moy_module = np.where(
                etuds_use_session2,
                notes_session2 / (eval_session2.note_max / 20.0),
                etuds_moy_module,
            )
            self.etuds_use_session2 = pd.Series(
                etuds_use_session2, index=self.evals_notes.index
            )
        else:
            # Rattrapage: remplace la note de module ssi elle est supérieure
            eval_rat = self.get_evaluation_rattrapage(modimpl)
            if eval_rat:
                notes_rat = self.evals_notes[eval_rat.id].values
                # remplace les notes invalides (ATT, EXC...) par des NaN
                notes_rat = np.where(
                    notes_rat > scu.NOTES_ABSENCE,
                    notes_rat / (eval_rat.note_max / 20.0),
                    np.nan,
                )
                # prend le max
                etuds_use_rattrapage = notes_rat > etuds_moy_module
                etuds_moy_module = np.where(
                    etuds_use_rattrapage, notes_rat, etuds_moy_module
                )
                self.etuds_use_rattrapage = pd.Series(
                    etuds_use_rattrapage, index=self.evals_notes.index
                )
        self.etuds_moy_module = pd.Series(
            etuds_moy_module,
            index=self.evals_notes.index,
        )

        return self.etuds_moy_module