ScoDoc-Lille/app/comp/res_but.py

##############################################################################
# ScoDoc
# Copyright (c) 1999 - 2023 Emmanuel Viennet.  All rights reserved.
# See LICENSE
##############################################################################

"""Résultats semestres BUT
"""
import time
import numpy as np
import pandas as pd

from app import db, log
from app.comp import moy_ue, moy_sem, inscr_mod
from app.comp.res_compat import NotesTableCompat
from app.comp.bonus_spo import BonusSport
from app.models import Formation, FormSemestreInscription, ScoDocSiteConfig
from app.models.moduleimpls import ModuleImpl
from app.models.but_refcomp import ApcParcours, ApcNiveau
from app.models.ues import DispenseUE, UniteEns
from app.models.but_validations import ApcValidationAnnee, ApcValidationRCUE
from app.scodoc import sco_preferences
from app.scodoc.codes_cursus import BUT_CODES_ORDER, UE_SPORT
from app.scodoc.sco_utils import ModuleType


class ResultatsSemestreBUT(NotesTableCompat):
    """Résultats BUT: organisation des calculs"""

    _cached_attrs = NotesTableCompat._cached_attrs + (
        "modimpl_coefs_df",
        "modimpls_evals_poids",
        "sem_cube",
        "etuds_parcour_id",  # parcours de chaque étudiant
        "ues_inscr_parcours_df",  # inscriptions aux UE / parcours
    )

    def __init__(self, formsemestre):
        super().__init__(formsemestre)

        self.sem_cube = None
        """ndarray (etuds x modimpl x ue)"""
        self.etuds_parcour_id = None
        """Parcours de chaque étudiant { etudid : parcour_id }"""
        self.ues_ids_by_parcour: dict[set[int]] = {}
        """{ parcour_id : set }, ue_id de chaque parcours"""
        self.validations_annee: dict[int, ApcValidationAnnee] = {}
        """chargé par get_validations_annee: jury annuel BUT"""
        if not self.load_cached():
            t0 = time.time()
            self.compute()
            t1 = time.time()
            self.store()
            t2 = time.time()
            log(
                f"""ResultatsSemestreBUT: cached formsemestre_id={formsemestre.id
                } ({(t1-t0):g}s +{(t2-t1):g}s)"""
            )

    def compute(self):
        "Charge les notes et inscriptions et calcule les moyennes d'UE et gen."
        (
            self.sem_cube,
            self.modimpls_evals_poids,
            self.modimpls_results,
        ) = moy_ue.notes_sem_load_cube(self.formsemestre)
        self.modimpl_inscr_df = inscr_mod.df_load_modimpl_inscr(self.formsemestre)
        self.ues_inscr_parcours_df = self.load_ues_inscr_parcours()
        self.modimpl_coefs_df, _, _ = moy_ue.df_load_modimpl_coefs(
            self.formsemestre, modimpls=self.formsemestre.modimpls_sorted
        )
        # l'idx de la colonne du mod modimpl.id est
        #       modimpl_coefs_df.columns.get_loc(modimpl.id)
        # idx de l'UE: modimpl_coefs_df.index.get_loc(ue.id)

        # Masque de tous les modules _sauf_ les bonus (sport)
        modimpls_mask = [
            modimpl.module.ue.type != UE_SPORT
            for modimpl in self.formsemestre.modimpls_sorted
        ]
        self.dispense_ues = DispenseUE.load_formsemestre_dispense_ues_set(
            self.formsemestre, self.modimpl_inscr_df.index, self.ues
        )
        self.etud_moy_ue = moy_ue.compute_ue_moys_apc(
            self.sem_cube,
            self.etuds,
            self.formsemestre.modimpls_sorted,
            self.modimpl_inscr_df,
            self.modimpl_coefs_df,
            modimpls_mask,
            self.dispense_ues,
            block=self.formsemestre.block_moyennes,
        )
        # Les coefficients d'UE ne sont pas utilisés en APC
        self.etud_coef_ue_df = pd.DataFrame(
            0.0, index=self.etud_moy_ue.index, columns=self.etud_moy_ue.columns
        )

        # --- Modules de MALUS sur les UEs
        self.malus = moy_ue.compute_malus(
            self.formsemestre, self.sem_cube, self.ues, self.modimpl_inscr_df
        )
        self.etud_moy_ue -= self.malus

        # --- Bonus Sport & Culture
        if not all(modimpls_mask):  # au moins un module bonus
            bonus_class = ScoDocSiteConfig.get_bonus_sport_class()
            if bonus_class is not None:
                bonus: BonusSport = bonus_class(
                    self.formsemestre,
                    self.sem_cube,
                    self.ues,
                    self.modimpl_inscr_df,
                    self.modimpl_coefs_df.transpose(),
                    self.etud_moy_gen,
                    self.etud_moy_ue,
                )
                self.bonus_ues = bonus.get_bonus_ues()
                if self.bonus_ues is not None:
                    self.etud_moy_ue += self.bonus_ues  # somme les dataframes

        # Clippe toutes les moyennes d'UE dans [0,20]
        self.etud_moy_ue.clip(lower=0.0, upper=20.0, inplace=True)

        # Nanifie les moyennes d'UE hors parcours pour chaque étudiant
        self.etud_moy_ue *= self.ues_inscr_parcours_df
        # Les ects (utilisés comme coefs) sont nuls pour les UE hors parcours:
        ects = self.ues_inscr_parcours_df.fillna(0.0) * [
            ue.ects for ue in self.ues if ue.type != UE_SPORT
        ]

        # Moyenne générale indicative:
        # (note: le bonus sport a déjà été appliqué aux moyennes d'UE, et impacte
        # donc la moyenne indicative)
        # self.etud_moy_gen = moy_sem.compute_sem_moys_apc_using_coefs(
        #    self.etud_moy_ue, self.modimpl_coefs_df
        # )
        if self.formsemestre.block_moyenne_generale or self.formsemestre.block_moyennes:
            self.etud_moy_gen = pd.Series(
                index=self.etud_moy_ue.index, dtype=float
            )  # NaNs
        else:
            self.etud_moy_gen = moy_sem.compute_sem_moys_apc_using_ects(
                self.etud_moy_ue,
                ects,
                formation_id=self.formsemestre.formation_id,
                skip_empty_ues=sco_preferences.get_preference(
                    "but_moy_skip_empty_ues", self.formsemestre.id
                ),
            )
        # --- UE capitalisées
        self.apply_capitalisation()

        # --- Classements:
        self.compute_rangs()

    def get_etud_mod_moy(self, moduleimpl_id: int, etudid: int) -> float:
        """La moyenne de l'étudiant dans le moduleimpl
        En APC, il s'agit d'une moyenne indicative sans valeur.
        Result: valeur float (peut être naN) ou chaîne "NI" (non inscrit ou DEM)
        """
        mod_idx = self.modimpl_coefs_df.columns.get_loc(moduleimpl_id)
        etud_idx = self.etud_index[etudid]
        # moyenne sur les UE:
        if len(self.sem_cube[etud_idx, mod_idx]):
            return np.nanmean(self.sem_cube[etud_idx, mod_idx])
            # note: si toutes les valeurs sont nan, on va déclencher ici
            # un RuntimeWarning: Mean of empty slice
        return np.nan

    def compute_etud_ue_coef(self, etudid: int, ue: UniteEns) -> float:
        """Détermine le coefficient de l'UE pour cet étudiant.
        N'est utilisé que pour l'injection des UE capitalisées dans la
        moyenne générale.
        En BUT, c'est simple: Coef = somme des coefs des modules vers cette UE.
        (ne dépend pas des modules auxquels est inscrit l'étudiant, ).
        """
        return self.modimpl_coefs_df.loc[ue.id].sum()

    def modimpls_in_ue(self, ue: UniteEns, etudid, with_bonus=True) -> list[ModuleImpl]:
        """Liste des modimpl ayant des coefs non nuls vers cette UE
        et auxquels l'étudiant est inscrit. Inclus modules bonus le cas échéant.
        """
        # sert pour l'affichage ou non de l'UE sur le bulletin et la table recap
        if ue.type == UE_SPORT:
            return [
                modimpl
                for modimpl in self.formsemestre.modimpls_sorted
                if modimpl.module.ue.id == ue.id
                and self.modimpl_inscr_df[modimpl.id][etudid]
            ]
        coefs = self.modimpl_coefs_df  # row UE (sans bonus), cols modimpl
        modimpls = [
            modimpl
            for modimpl in self.formsemestre.modimpls_sorted
            if (
                modimpl.module.ue.type != UE_SPORT
                and (coefs[modimpl.id][ue.id] != 0)
                and self.modimpl_inscr_df[modimpl.id][etudid]
            )
            or (
                modimpl.module.module_type == ModuleType.MALUS
                and modimpl.module.ue_id == ue.id
            )
        ]
        if not with_bonus:
            return [
                modimpl for modimpl in modimpls if modimpl.module.ue.type != UE_SPORT
            ]
        return modimpls

    def modimpl_notes(self, modimpl_id: int, ue_id: int) -> np.ndarray:
        """Les notes moyennes des étudiants du sem. à ce modimpl dans cette ue.
        Utile pour stats bottom tableau recap.
        Résultat: 1d array of float
        """
        i = self.modimpl_coefs_df.columns.get_loc(modimpl_id)
        j = self.modimpl_coefs_df.index.get_loc(ue_id)
        return self.sem_cube[:, i, j]

    def load_ues_inscr_parcours(self) -> pd.DataFrame:
        """Chargement des inscriptions aux parcours et calcul de la
        matrice d'inscriptions (etuds, ue).
        S'il n'y pas de référentiel de compétence, donc pas de parcours,
        on considère l'étudiant inscrit à toutes les ue.
        La matrice avec ue ne comprend que les UE non bonus.
        1.0 si étudiant inscrit à l'UE, NaN sinon.
        """
        etuds_parcour_id = {
            inscr.etudid: inscr.parcour_id for inscr in self.formsemestre.inscriptions
        }
        self.etuds_parcour_id = etuds_parcour_id
        ue_ids = [ue.id for ue in self.ues if ue.type != UE_SPORT]
        ue_ids_set = set(ue_ids)
        if self.formsemestre.formation.referentiel_competence is None:
            return pd.DataFrame(
                1.0, index=etuds_parcour_id.keys(), columns=ue_ids, dtype=float
            )
        # matrice de NaN: inscrits par défaut à AUCUNE UE:
        ues_inscr_parcours_df = pd.DataFrame(
            np.nan, index=etuds_parcour_id.keys(), columns=ue_ids, dtype=float
        )
        # Construit pour chaque parcours du référentiel l'ensemble de ses UE
        #  - considère aussi le cas des semestres sans parcours (clé parcour None)
        #  - retire les UEs qui ont un parcours mais qui ne sont pas dans l'un des
        #    parcours du semestre

        ue_by_parcours = {}  # parcours_id : {ue_id:0|1}
        for (
            parcour
        ) in self.formsemestre.formation.referentiel_competence.parcours.all() + [None]:
            ue_by_parcours[None if parcour is None else parcour.id] = {
                ue.id: 1.0
                for ue in self.formsemestre.formation.query_ues_parcour(parcour).filter(
                    UniteEns.semestre_idx == self.formsemestre.semestre_id
                )
                if ue.id in ue_ids_set
            }
        #
        for etudid in etuds_parcour_id:
            parcour_id = etuds_parcour_id[etudid]
            if parcour_id in ue_by_parcours:
                if ue_by_parcours[parcour_id]:
                    ues_inscr_parcours_df.loc[etudid] = ue_by_parcours[parcour_id]
        return ues_inscr_parcours_df

    def etud_ues_ids(self, etudid: int) -> list[int]:
        """Liste des id d'UE auxquelles l'étudiant est inscrit (sans bonus).
        (surchargée ici pour prendre en compte les parcours)
        Ne prend pas en compte les éventuelles DispenseUE (pour le moment ?)
        """
        s = self.ues_inscr_parcours_df.loc[etudid]
        return s.index[s.notna()]

    def etud_parcours_ues_ids(self, etudid: int) -> set[int]:
        """Ensemble les id des UEs que l'étudiant doit valider dans ce semestre compte tenu
        du parcours dans lequel il est inscrit.
        Se base sur le parcours dans ce semestre, et le référentiel de compétences.
        Note: il n'est pas nécessairement inscrit à toutes ces UEs.
        Ensemble vide si pas de référentiel.
        Si l'étudiant n'est pas inscrit dans un parcours, toutes les UEs du semestre.
        La requête est longue, les ue_ids par parcour sont donc cachés.
        """
        parcour_id = self.etuds_parcour_id[etudid]
        if parcour_id in self.ues_ids_by_parcour:  # cache
            return self.ues_ids_by_parcour[parcour_id]
        # Hors cache:
        ref_comp = self.formsemestre.formation.referentiel_competence
        if ref_comp is None:
            return set()
        if parcour_id is None:
            ues_ids = {ue.id for ue in self.ues if ue.type != UE_SPORT}
        else:
            parcour: ApcParcours = ApcParcours.query.get(parcour_id)
            annee = (self.formsemestre.semestre_id + 1) // 2
            niveaux = ApcNiveau.niveaux_annee_de_parcours(parcour, annee, ref_comp)
            # Les UEs du formsemestre associées à ces niveaux:
            ues_parcour = self.formsemestre.formation.query_ues_parcour(parcour)
            ues_ids = set()
            for niveau in niveaux:
                ue = ues_parcour.filter(UniteEns.niveau_competence == niveau).first()
                if ue:
                    ues_ids.add(ue.id)

        # memoize
        self.ues_ids_by_parcour[parcour_id] = ues_ids

        return ues_ids

    def etud_has_decision(self, etudid) -> bool:
        """True s'il y a une décision (quelconque) de jury
        émanant de ce formsemestre pour cet étudiant.
        prend aussi en compte les autorisations de passage.
        Ici sous-classée (BUT) pour les RCUEs et années.
        """
        return bool(
            super().etud_has_decision(etudid)
            or ApcValidationAnnee.query.filter_by(
                formsemestre_id=self.formsemestre.id, etudid=etudid
            ).count()
            or ApcValidationRCUE.query.filter_by(
                formsemestre_id=self.formsemestre.id, etudid=etudid
            ).count()
        )

    def get_validations_annee(self) -> dict[int, ApcValidationAnnee]:
        """Les validations des étudiants de ce semestre
        pour l'année BUT d'une formation compatible avec celle de ce semestre.
        Attention:
        1) la validation ne provient pas nécessairement de ce semestre
        (redoublants, pair/impair, extérieurs).
        2) l'étudiant a pu démissionner ou défaillir.
        3) S'il y a plusieurs validations pour le même étudiant, prend la "meilleure".

        Mémorise le résultat (dans l'instance, pas en cache: TODO voir au profiler)
        """
        if self.validations_annee:
            return self.validations_annee
        annee_but = (self.formsemestre.semestre_id + 1) // 2
        validations = (
            ApcValidationAnnee.query.filter_by(ordre=annee_but)
            .join(Formation)
            .filter_by(formation_code=self.formsemestre.formation.formation_code)
            .join(
                FormSemestreInscription,
                db.and_(
                    FormSemestreInscription.etudid == ApcValidationAnnee.etudid,
                    FormSemestreInscription.formsemestre_id == self.formsemestre.id,
                ),
            )
        )
        validation_by_etud = {}
        for validation in validations:
            if validation.etudid in validation_by_etud:
                # keep the "best"
                if BUT_CODES_ORDER.get(validation.code, 0) > BUT_CODES_ORDER.get(
                    validation_by_etud[validation.etudid].code, 0
                ):
                    validation_by_etud[validation.etudid] = validation
            else:
                validation_by_etud[validation.etudid] = validation
        self.validations_annee = validation_by_etud
        return self.validations_annee