ScoDoc/app/comp/res_common.py

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

"""Résultats semestre: méthodes communes aux formations classiques et APC
"""

from collections import Counter, defaultdict
from collections.abc import Generator
import datetime
from functools import cached_property
from operator import attrgetter

import numpy as np
import pandas as pd
import sqlalchemy as sa
from flask import g, url_for

from app import db
from app.comp import res_sem
from app.comp.res_cache import ResultatsCache
from app.comp.jury import ValidationsSemestre
from app.comp.moy_mod import ModuleImplResults
from app.models import (
    Evaluation,
    FormSemestre,
    FormSemestreUECoef,
    Identite,
    ModuleImpl,
    ModuleImplInscription,
    ScolarAutorisationInscription,
    UniteEns,
)
from app.scodoc.sco_cache import ResultatsSemestreCache
from app.scodoc.codes_cursus import UE_SPORT
from app.scodoc.sco_exceptions import ScoValueError, ScoTemporaryError
from app.scodoc import sco_utils as scu


# Il faut bien distinguer
#  - ce qui est caché de façon persistente (via redis):
#      ce sont les attributs listés dans `_cached_attrs`
#      le stockage et l'invalidation sont gérés dans sco_cache.py
#
#  - les valeurs cachées durant le temps d'une requête
#      (durée de vie de l'instance de ResultatsSemestre)
#      qui sont notamment les attributs décorés par `@cached_property``
#
class ResultatsSemestre(ResultatsCache):
    """Les résultats (notes, ...) d'un formsemestre
    Classe commune à toutes les formations (classiques, BUT)
    """

    _cached_attrs = (
        "bonus",
        "bonus_ues",
        "dispense_ues",
        "etud_coef_ue_df",
        "etud_moy_gen_ranks",
        "etud_moy_gen",
        "etud_moy_ue",
        "modimpl_inscr_df",
        "modimpls_results",
        "moyennes_matieres",
    )

    def __init__(self, formsemestre: FormSemestre):
        super().__init__(formsemestre, ResultatsSemestreCache)
        # BUT ou standard ? (apc == "approche par compétences")
        self.is_apc: bool = formsemestre.formation.is_apc()
        # Attributs "virtuels", définis dans les sous-classes
        self.bonus: pd.Series = None  # virtuel
        "Bonus sur moy. gen. Series de float, index etudid"
        self.bonus_ues: pd.DataFrame = None  # virtuel
        "DataFrame de float, index etudid, columns: ue.id"
        self.dispense_ues: set[tuple[int, int]] = set()
        """set des dispenses d'UE: (etudid, ue_id), en APC seulement."""
        #  ResultatsSemestreBUT ou ResultatsSemestreClassic
        self.etud_moy_ue = {}
        "etud_moy_ue: DataFrame columns UE, rows etudid"
        self.etud_moy_gen: pd.Series = None
        self.etud_moy_gen_ranks = {}
        self.etud_moy_gen_ranks_int = {}
        self.moy_gen_rangs_by_group = None  # virtual
        self.modimpl_inscr_df: pd.DataFrame = None
        "Inscriptions: row etudid, col modimlpl_id"
        self.modimpls_results: dict[int, ModuleImplResults] = None
        "Résultats de chaque modimpl (classique ou BUT)"
        self.etud_coef_ue_df = None
        """coefs d'UE effectifs pour chaque étudiant (pour form. classiques)"""
        self.modimpl_coefs_df: pd.DataFrame = None
        """Coefs APC: rows = UEs (sans bonus), columns = modimpl, value = coef."""

        self.validations = None
        self.autorisations_inscription = None
        self.moyennes_matieres = {}
        """Moyennes de matières, si calculées. { matiere_id : Series, index etudid }"""
        # self._ues_by_id_cache: dict[int, UniteEns] = {}  # per-instance cache

    def __repr__(self):
        return f"<{self.__class__.__name__}(formsemestre='{self.formsemestre}')>"

    def compute(self):
        "Charge les notes et inscriptions et calcule toutes les moyennes"
        # voir ce qui est chargé / calculé ici et dans les sous-classes
        raise NotImplementedError()

    def get_inscriptions_counts(self) -> Counter:
        """Nombre d'inscrits, défaillants, démissionnaires.

        Exemple: res.get_inscriptions_counts()[scu.INSCRIT]

        Result: a collections.Counter instance
        """
        return Counter(ins.etat for ins in self.formsemestre.inscriptions)

    @cached_property
    def etuds(self) -> list[Identite]:
        "Liste des inscrits au semestre, avec les démissionnaires et les défaillants"
        # nb: si la liste des inscrits change, ResultatsSemestre devient invalide
        return self.formsemestre.get_inscrits(include_demdef=True)

    @cached_property
    def etud_index(self) -> dict[int, int]:
        "dict { etudid : indice dans les inscrits }"
        return {e.id: idx for idx, e in enumerate(self.etuds)}

    def etud_ues_ids(self, etudid: int) -> list[int]:
        """Liste des UE auxquelles l'etudiant est inscrit, sans bonus
        (surchargée en BUT pour prendre en compte les parcours)
        """
        # Pour les formations classiques, etudid n'est pas utilisé
        # car tous les étudiants sont inscrits à toutes les UE
        return [ue.id for ue in self.ues if ue.type != UE_SPORT]

    def etud_parcours_ues_ids(self, etudid: int) -> set[int]:
        """Ensemble des UEs que l'étudiant "doit" valider.
        En formations classiques, c'est la même chose (en set) que etud_ues_ids.
        Surchargée en BUT pour donner les UEs du parcours de l'étudiant.
        """
        return {ue.id for ue in self.ues if ue.type != UE_SPORT}

    def etud_ues(self, etudid: int) -> Generator[UniteEns]:
        """Liste des UE auxquelles l'étudiant est inscrit
        (sans bonus, en BUT prend en compte le parcours de l'étudiant)."""
        return (db.session.get(UniteEns, ue_id) for ue_id in self.etud_ues_ids(etudid))

    def etud_ects_tot_sem(self, etudid: int) -> float:
        """Le total des ECTS associées à ce semestre (que l'étudiant peut ou non valider)"""
        etud_ues = self.etud_ues(etudid)
        return sum([ue.ects or 0.0 for ue in etud_ues]) if etud_ues else 0.0

    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
        """
        # différent en BUT et classique: virtuelle
        raise NotImplementedError

    @cached_property
    def etuds_dict(self) -> dict[int, Identite]:
        """dict { etudid : Identite } inscrits au semestre,
        avec les démissionnaires et defs."""
        return {etud.id: etud for etud in self.etuds}

    @cached_property
    def ues(self) -> list[UniteEns]:
        """Liste des UEs du semestre (avec les UE bonus sport)
        (indices des DataFrames).
        Note: un étudiant n'est pas nécessairement inscrit dans toutes ces UEs.
        """
        return self.formsemestre.get_ues(with_sport=True)

    @cached_property
    def ressources(self):
        "Liste des ressources du semestre, triées par numéro de module"
        return [
            m
            for m in self.formsemestre.modimpls_sorted
            if m.module.module_type == scu.ModuleType.RESSOURCE
        ]

    @cached_property
    def saes(self):
        "Liste des SAÉs du semestre, triées par numéro de module"
        return [
            m
            for m in self.formsemestre.modimpls_sorted
            if m.module.module_type == scu.ModuleType.SAE
        ]

    def get_etudids_attente(self) -> set[int]:
        """L'ensemble des etudids ayant au moins une note en ATTente"""
        return set().union(
            *[mr.etudids_attente for mr in self.modimpls_results.values()]
        )

    # Etat des évaluations
    def get_evaluation_etat(self, evaluation: Evaluation) -> dict:
        """État d'une évaluation
        {
            "coefficient" : float, # 0 si None
            "description" : str, # de l'évaluation, "" si None
            "etat" {
                "blocked" : bool, # vrai si prise en compte bloquée
                "evalcomplete" : bool,
                "last_modif" : datetime.datetime | None, # saisie de note la plus récente
                "nb_notes" : int, # nb notes d'étudiants inscrits
                "nb_attente" : int, # nb de notes en ATTente (même si bloquée)
            },
            "evaluation_id" : int,
            "jour" : datetime.datetime, # e.date_debut or datetime.datetime(1900, 1, 1)
            "publish_incomplete" : bool,
        }
        """
        mod_results = self.modimpls_results.get(evaluation.moduleimpl_id)
        if mod_results is None:
            raise ScoTemporaryError()  # argh !
        etat = mod_results.evaluations_etat.get(evaluation.id)
        if etat is None:
            raise ScoTemporaryError()  # argh !
        # Date de dernière saisie de note
        cursor = db.session.execute(
            sa.text(
                "SELECT MAX(date) FROM notes_notes WHERE evaluation_id = :evaluation_id"
            ),
            {"evaluation_id": evaluation.id},
        )
        date_modif = cursor.one_or_none()
        last_modif = date_modif[0] if date_modif else None
        return {
            "coefficient": evaluation.coefficient,
            "description": evaluation.description,
            "etat": {
                "blocked": evaluation.is_blocked(),
                "evalcomplete": etat.is_complete,
                "nb_attente": etat.nb_attente,
                "nb_notes": etat.nb_notes,
                "last_modif": last_modif,
            },
            "evaluation_id": evaluation.id,
            "jour": evaluation.date_debut or datetime.datetime(1900, 1, 1),
            "publish_incomplete": evaluation.publish_incomplete,
        }

    def get_mod_evaluation_etat_list(self, modimpl: ModuleImpl) -> list[dict]:
        """Liste des états des évaluations de ce module
        [ evaluation_etat, ... ] (voir get_evaluation_etat)
        trié par (numero desc, date_debut desc)
        """
        # nouvelle version 2024-02-02
        return list(
            reversed(
                [
                    self.get_evaluation_etat(evaluation)
                    for evaluation in modimpl.evaluations
                ]
            )
        )

    # modernisation de get_mod_evaluation_etat_list
    # utilisé par:
    # sco_evaluations.do_evaluation_etat_in_mod
    #   e["etat"]["evalcomplete"]
    #   e["etat"]["nb_notes"]
    #   e["etat"]["last_modif"]
    #
    # sco_formsemestre_status.formsemestre_description_table
    #   "jour" (qui est e.date_debut or datetime.date(1900, 1, 1))
    #   "description"
    #   "coefficient"
    #   e["etat"]["evalcomplete"]
    #   publish_incomplete
    #
    # sco_formsemestre_status.formsemestre_tableau_modules
    #   e["etat"]["nb_notes"]
    #

    # --- JURY...
    def get_formsemestre_validations(self) -> ValidationsSemestre:
        """Load validations if not already stored, set attribute and return value"""
        if not self.validations:
            self.validations = res_sem.load_formsemestre_validations(self.formsemestre)
        return self.validations

    def get_autorisations_inscription(self) -> dict[int : list[int]]:
        """Les autorisations d'inscription venant de ce formsemestre.
        Lit en base et cache le résultat.
        Resultat: { etudid : [ indices de semestres ]}
        Note: les etudids peuvent ne plus être inscrits ici.
        Seuls ceux avec des autorisations enregistrées sont présents dans le résultat.
        """
        if not self.autorisations_inscription:
            autorisations = ScolarAutorisationInscription.query.filter_by(
                origin_formsemestre_id=self.formsemestre.id
            )
            self.autorisations_inscription = defaultdict(list)
            for aut in autorisations:
                self.autorisations_inscription[aut.etudid].append(aut.semestre_id)
        return self.autorisations_inscription

    def get_etud_ue_validables(self, etudid: int) -> list[UniteEns]:
        """Liste des UEs du semestre qui doivent être validées

        Rappel: l'étudiant est inscrit à des modimpls et non à des UEs.

        - En BUT: on considère que l'étudiant va (ou non) valider toutes les UEs des modules
        du parcours.

        - En classique: toutes les UEs des modimpls auxquels l'étudiant est inscrit sont
        susceptibles d'être validées.

        Les UE "bonus" (sport) ne sont jamais "validables".
        """
        if self.is_apc:
            return list(self.etud_ues(etudid))
        # Formations classiques:
        # restreint aux UE auxquelles l'étudiant est inscrit (dans l'un des modimpls)
        ues = {
            modimpl.module.ue
            for modimpl in self.formsemestre.modimpls_sorted
            if self.modimpl_inscr_df[modimpl.id][etudid]
        }
        ues = sorted(list(ues), key=attrgetter("numero"))
        return ues

    def modimpls_in_ue(self, ue: UniteEns, etudid, with_bonus=True) -> list[ModuleImpl]:
        """Liste des modimpl de cette UE auxquels l'étudiant est inscrit.
        Utile en formations classiques, surchargée pour le BUT.
        Inclus modules bonus le cas échéant.
        """
        # Utilisée pour l'affichage ou non de l'UE sur le bulletin
        # Méthode surchargée en BUT
        modimpls = [
            modimpl
            for modimpl in self.formsemestre.modimpls_sorted
            if modimpl.module.ue.id == ue.id
            and self.modimpl_inscr_df[modimpl.id][etudid]
        ]
        if not with_bonus:
            return [
                modimpl for modimpl in modimpls if modimpl.module.ue.type != UE_SPORT
            ]
        return modimpls

    @cached_property
    def ue_au_dessus(self, seuil=10.0) -> pd.DataFrame:
        """DataFrame columns UE, rows etudid, valeurs: bool
        Par exemple, pour avoir le nombre d'UE au dessus de 10 pour l'étudiant etudid
        nb_ues_ok = sum(res.ue_au_dessus().loc[etudid])
        """
        return self.etud_moy_ue > (seuil - scu.NOTES_TOLERANCE)

    def apply_capitalisation(self):
        """Recalcule la moyenne générale pour prendre en compte d'éventuelles
        UE capitalisées.
        """
        # Supposant qu'il y a peu d'UE capitalisées,
        # on recalcule les moyennes gen des etuds ayant des UEs capitalisées.
        self.get_formsemestre_validations()
        ue_capitalisees = self.validations.ue_capitalisees
        for etudid in ue_capitalisees.index:
            recompute_mg = False
            # ue_codes = set(ue_capitalisees.loc[etudid]["ue_code"])
            # for ue_code in ue_codes:
            #     ue = ue_by_code.get(ue_code)
            #     if ue is None:
            #         ue = self.formsemestre.query_ues.filter_by(ue_code=ue_code)
            #         ue_by_code[ue_code] = ue

            # Quand il y a une capitalisation, vérifie toutes les UEs
            sum_notes_ue = 0.0
            sum_coefs_ue = 0.0
            for ue in self.formsemestre.get_ues():
                ue_cap = self.get_etud_ue_status(etudid, ue.id)
                if ue_cap is None:
                    continue
                if ue_cap["is_capitalized"]:
                    recompute_mg = True
                coef = ue_cap["coef_ue"]
                if not np.isnan(ue_cap["moy"]) and coef:
                    sum_notes_ue += ue_cap["moy"] * coef
                    sum_coefs_ue += coef

            if recompute_mg and sum_coefs_ue > 0.0:
                # On doit prendre en compte une ou plusieurs UE capitalisées
                # et donc recalculer la moyenne générale
                self.etud_moy_gen[etudid] = sum_notes_ue / sum_coefs_ue
                # Ajoute le bonus sport
                if self.bonus is not None and self.bonus[etudid]:
                    self.etud_moy_gen[etudid] += self.bonus[etudid]
                    self.etud_moy_gen[etudid] = max(
                        0.0, min(self.etud_moy_gen[etudid], 20.0)
                    )

    def get_etud_etat(self, etudid: int) -> str:
        "Etat de l'etudiant: 'I', 'D', DEF ou '' (si pas connu dans ce semestre)"
        ins = self.formsemestre.etuds_inscriptions.get(etudid)
        if ins is None:
            return ""
        return ins.etat

    def _get_etud_ue_cap(self, etudid: int, ue: UniteEns) -> dict:
        """Donne les informations sur la capitalisation de l'UE ue pour cet étudiant.
        Résultat:
            Si pas capitalisée: None
            Si capitalisée: un dict, avec les colonnes de validation.
        """
        capitalisations = self.validations.ue_capitalisees.loc[etudid]
        if isinstance(capitalisations, pd.DataFrame):
            ue_cap = capitalisations[capitalisations["ue_code"] == ue.ue_code]
            if ue_cap.empty:
                return None
            if isinstance(ue_cap, pd.DataFrame):
                # si plusieurs fois capitalisée, prend le max
                cap_idx = ue_cap["moy_ue"].values.argmax()
                ue_cap = ue_cap.iloc[cap_idx]
        else:
            if capitalisations["ue_code"] == ue.ue_code:
                ue_cap = capitalisations
            else:
                return None
        # converti la Series en dict, afin que les np.int64 reviennent en int
        # et remplace les NaN (venant des NULL en base) par des None
        ue_cap_dict = ue_cap.to_dict()
        if ue_cap_dict["formsemestre_id"] is not None and np.isnan(
            ue_cap_dict["formsemestre_id"]
        ):
            ue_cap_dict["formsemestre_id"] = None
        if ue_cap_dict["compense_formsemestre_id"] is not None and np.isnan(
            ue_cap_dict["compense_formsemestre_id"]
        ):
            ue_cap_dict["compense_formsemestre_id"] = None
        return ue_cap_dict

    def get_etud_ue_status(self, etudid: int, ue_id: int) -> dict | None:
        """L'état de l'UE pour cet étudiant.
        Result: dict, ou None si l'UE n'existe pas ou n'est pas dans ce semestre.
        {
            "is_capitalized": # vrai si la version capitalisée est celle prise en compte (meilleure)
            "was_capitalized":# si elle a été capitalisée (meilleure ou pas)
            "is_external": # si UE externe
            "coef_ue": 0.0,
            "cur_moy_ue": 0.0, # moyenne de l'UE courante
            "moy": 0.0, # moyenne prise en compte
            "event_date": # date de la capiltalisation éventuelle (ou None)
            "ue": ue_dict, # l'UE, comme un dict
            "formsemestre_id": None,
            "capitalized_ue_id": None, # l'id de l'UE capitalisée, ou None
            "ects_pot": 0.0, # deprecated (les ECTS liés à cette UE)
            "ects": 0.0, # les ECTS acquis grace à cette UE
            "ects_ue": # les ECTS liés à cette UE
        }
        """
        ue: UniteEns = db.session.get(UniteEns, ue_id)
        if not ue:
            return None
        ue_dict = ue.to_dict()

        if ue.type == UE_SPORT:
            return {
                "is_capitalized": False,
                "was_capitalized": False,
                "is_external": False,
                "coef_ue": 0.0,
                "cur_moy_ue": 0.0,
                "moy": 0.0,
                "event_date": None,
                "ue": ue_dict,
                "formsemestre_id": None,
                "capitalized_ue_id": None,
                "ects_pot": 0.0,
                "ects": 0.0,
                "ects_ue": ue.ects,
            }
        if not ue_id in self.etud_moy_ue or not etudid in self.etud_moy_ue[ue_id]:
            return None
        if not self.validations:
            self.validations = res_sem.load_formsemestre_validations(self.formsemestre)
        cur_moy_ue = self.etud_moy_ue[ue_id][etudid]
        moy_ue = cur_moy_ue
        is_capitalized = False  # si l'UE prise en compte est une UE capitalisée
        # s'il y a precedemment une UE capitalisée (pas forcement meilleure):
        was_capitalized = False
        if etudid in self.validations.ue_capitalisees.index:
            ue_cap = self._get_etud_ue_cap(etudid, ue)
            if (
                ue_cap
                and (ue_cap["moy_ue"] is not None)
                and not np.isnan(ue_cap["moy_ue"])
            ):
                was_capitalized = True
                if ue_cap["moy_ue"] > cur_moy_ue or np.isnan(cur_moy_ue):
                    moy_ue = ue_cap["moy_ue"]
                    is_capitalized = True

        # Coef l'UE dans le semestre courant:
        if self.is_apc:
            # utilise les ECTS comme coef.
            coef_ue = ue.ects
        else:
            # formations classiques
            coef_ue = self.etud_coef_ue_df[ue_id][etudid]
        if (not coef_ue) and is_capitalized:  # étudiant non inscrit dans l'UE courante
            if self.is_apc:
                # Coefs de l'UE capitalisée en formation APC: donné par ses ECTS
                ue_capitalized = db.session.get(UniteEns, ue_cap["ue_id"])
                coef_ue = ue_capitalized.ects
                if coef_ue is None:
                    orig_sem = FormSemestre.get_formsemestre(ue_cap["formsemestre_id"])
                    raise ScoValueError(
                        f"""L'UE capitalisée {ue_capitalized.acronyme}
                    du semestre {orig_sem.titre_annee()}
                    n'a pas d'indication d'ECTS.
                    Corrigez ou faite corriger le programme
                    <a href="{url_for('notes.ue_table', scodoc_dept=g.scodoc_dept,
                    formation_id=ue_capitalized.formation_id)}">via cette page</a>.
                    """,
                        safe=True,
                    )
            else:
                # Coefs de l'UE capitalisée en formation classique:
                # va chercher le coef dans le semestre d'origine
                coef_ue = ModuleImplInscription.sum_coefs_modimpl_ue(
                    ue_cap["formsemestre_id"], etudid, ue_cap["ue_id"]
                )

        return {
            "is_capitalized": is_capitalized,
            "was_capitalized": was_capitalized,
            "is_external": ue_cap["is_external"] if is_capitalized else ue.is_external,
            "coef_ue": coef_ue,
            "ects_pot": ue.ects or 0.0,
            "ects": (
                self.validations.decisions_jury_ues.get(etudid, {})
                .get(ue.id, {})
                .get("ects", 0.0)
                if self.validations.decisions_jury_ues
                else 0.0
            ),
            "ects_ue": ue.ects,
            "cur_moy_ue": cur_moy_ue,
            "moy": moy_ue,
            "event_date": ue_cap["event_date"] if is_capitalized else None,
            "ue": ue_dict,
            "formsemestre_id": ue_cap["formsemestre_id"] if is_capitalized else None,
            "capitalized_ue_id": ue_cap["ue_id"] if is_capitalized else None,
        }

    def compute_etud_ue_coef(self, etudid: int, ue: UniteEns) -> float:
        "Détermine le coefficient de l'UE pour cet étudiant."
        # calcul différent en classique et BUT
        raise NotImplementedError()

    def get_etud_ue_cap_coef(self, etudid, ue, ue_cap):  # UNUSED in ScoDoc 9
        """Calcule le coefficient d'une UE capitalisée, pour cet étudiant,
        injectée dans le semestre courant.

        ue : ue du semestre courant

        ue_cap = resultat de formsemestre_get_etud_capitalisation
        { 'ue_id' (dans le semestre source),
          'ue_code', 'moy', 'event_date','formsemestre_id' }
        """
        # 1- Coefficient explicitement déclaré dans le semestre courant pour cette UE ?
        ue_coef_db = FormSemestreUECoef.query.filter_by(
            formsemestre_id=self.formsemestre.id, ue_id=ue.id
        ).first()
        if ue_coef_db is not None:
            return ue_coef_db.coefficient

        # En APC: somme des coefs des modules vers cette UE
        # En classique:  Capitalisation UE externe: quel coef appliquer ?
        # En ScoDoc 7, calculait la somme des coefs dans l'UE du semestre d'origine
        # ici si l'étudiant est inscrit dans le semestre courant,
        # somme des coefs des modules de l'UE auxquels il est inscrit
        return self.compute_etud_ue_coef(etudid, ue)