ScoDoc-PE/app/but/bulletin_but.py

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

from collections import defaultdict
import datetime
from flask import url_for, g
import numpy as np
import pandas as pd

from app import db

from app.comp import moy_ue, moy_sem, inscr_mod
from app.models import ModuleImpl
from app.scodoc import sco_utils as scu
from app.scodoc.sco_cache import ResultatsSemestreBUTCache
from app.scodoc.sco_exceptions import ScoFormatError
from app.scodoc import sco_preferences
from app.scodoc.sco_utils import jsnan, fmt_note


class ResultatsSemestreBUT:
    """Structure légère pour stocker les résultats du semestre et
    générer les bulletins.
    __init__ : charge depuis le cache ou calcule
    invalidate(): invalide données cachées
    """

    _cached_attrs = (
        "sem_cube",
        "modimpl_inscr_df",
        "modimpl_coefs_df",
        "etud_moy_ue",
        "modimpls_evals_poids",
        "modimpls_evals_notes",
        "etud_moy_gen",
        "etud_moy_gen_ranks",
        "modimpls_evaluations_complete",
    )

    def __init__(self, formsemestre):
        self.formsemestre = formsemestre
        self.ues = formsemestre.query_ues().all()
        self.modimpls = formsemestre.modimpls.all()
        self.etuds = self.formsemestre.get_inscrits(include_dem=False)
        self.etud_index = {e.id: idx for idx, e in enumerate(self.etuds)}
        self.saes = [
            m for m in self.modimpls if m.module.module_type == scu.ModuleType.SAE
        ]
        self.ressources = [
            m for m in self.modimpls if m.module.module_type == scu.ModuleType.RESSOURCE
        ]
        if not self.load_cached():
            self.compute()
            self.store()

    def load_cached(self) -> bool:
        "Load cached dataframes, returns False si pas en cache"
        data = ResultatsSemestreBUTCache.get(self.formsemestre.id)
        if not data:
            return False
        for attr in self._cached_attrs:
            setattr(self, attr, data[attr])
        return True

    def store(self):
        "Cache our dataframes"
        ResultatsSemestreBUTCache.set(
            self.formsemestre.id,
            {attr: getattr(self, attr) for attr in self._cached_attrs},
        )

    def compute(self):
        "Charge les notes et inscriptions et calcule toutes les moyennes"
        (
            self.sem_cube,
            self.modimpls_evals_poids,
            self.modimpls_evals_notes,
            modimpls_evaluations,
            self.modimpls_evaluations_complete,
        ) = moy_ue.notes_sem_load_cube(self.formsemestre)
        self.modimpl_inscr_df = inscr_mod.df_load_modimpl_inscr(self.formsemestre)
        self.modimpl_coefs_df, _, _ = moy_ue.df_load_modimpl_coefs(
            self.formsemestre, ues=self.ues, modimpls=self.modimpls
        )
        # 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)
        self.etud_moy_ue = moy_ue.compute_ue_moys(
            self.sem_cube,
            self.etuds,
            self.modimpls,
            self.ues,
            self.modimpl_inscr_df,
            self.modimpl_coefs_df,
        )
        self.etud_moy_gen = moy_sem.compute_sem_moys(
            self.etud_moy_ue, self.modimpl_coefs_df
        )
        self.etud_moy_gen_ranks = moy_sem.comp_ranks_series(self.etud_moy_gen)

    def etud_ue_mod_results(self, etud, ue, modimpls) -> dict:
        "dict synthèse résultats dans l'UE pour les modules indiqués"
        d = {}
        etud_idx = self.etud_index[etud.id]
        ue_idx = self.modimpl_coefs_df.index.get_loc(ue.id)
        etud_moy_module = self.sem_cube[etud_idx]  # module x UE
        for mi in modimpls:
            coef = self.modimpl_coefs_df[mi.id][ue.id]
            if coef > 0:
                d[mi.module.code] = {
                    "id": mi.id,
                    "coef": coef,
                    "moyenne": fmt_note(
                        etud_moy_module[self.modimpl_coefs_df.columns.get_loc(mi.id)][
                            ue_idx
                        ]
                    ),
                }
        return d

    def etud_ue_results(self, etud, ue):
        "dict synthèse résultats UE"
        d = {
            "id": ue.id,
            "numero": ue.numero,
            "ECTS": {
                "acquis": 0,  # XXX TODO voir jury
                "total": ue.ects,
            },
            "competence": None,  # XXX TODO lien avec référentiel
            "moyenne": {
                "value": fmt_note(self.etud_moy_ue[ue.id][etud.id]),
                "min": fmt_note(self.etud_moy_ue[ue.id].min()),
                "max": fmt_note(self.etud_moy_ue[ue.id].max()),
                "moy": fmt_note(self.etud_moy_ue[ue.id].mean()),
            },
            "bonus": None,  # XXX TODO
            "malus": None,  # XXX TODO voir ce qui est ici
            "capitalise": None,  # "AAAA-MM-JJ" TODO
            "ressources": self.etud_ue_mod_results(etud, ue, self.ressources),
            "saes": self.etud_ue_mod_results(etud, ue, self.saes),
        }
        return d

    def etud_mods_results(self, etud, modimpls) -> dict:
        """dict synthèse résultats des modules indiqués,
        avec évaluations de chacun."""
        d = {}
        etud_idx = self.etud_index[etud.id]
        for mi in modimpls:
            mod_idx = self.modimpl_coefs_df.columns.get_loc(mi.id)
            # moyennes indicatives (moyennes de moyennes d'UE)
            try:
                moyennes_etuds = np.nan_to_num(
                    np.nanmean(self.sem_cube[:, mod_idx, :], axis=1),
                    copy=False,
                )
            except RuntimeWarning:  # all nans in np.nanmean (sur certains etuds sans notes valides)
                pass
            try:
                moy_indicative_mod = np.nanmean(self.sem_cube[etud_idx, mod_idx])
            except RuntimeWarning:  # all nans in np.nanmean
                pass
            d[mi.module.code] = {
                "id": mi.id,
                "titre": mi.module.titre,
                "code_apogee": mi.module.code_apogee,
                "url": url_for(
                    "notes.moduleimpl_status",
                    scodoc_dept=g.scodoc_dept,
                    moduleimpl_id=mi.id,
                ),
                "moyenne": {
                    # moyenne indicative de module: moyenne des UE, ignorant celles sans notes (nan)
                    "value": fmt_note(moy_indicative_mod),
                    "min": fmt_note(moyennes_etuds.min()),
                    "max": fmt_note(moyennes_etuds.max()),
                    "moy": fmt_note(moyennes_etuds.mean()),
                },
                "evaluations": [
                    self.etud_eval_results(etud, e)
                    for eidx, e in enumerate(mi.evaluations)
                    if e.visibulletin
                    and self.modimpls_evaluations_complete[mi.id][eidx]
                ],
            }
        return d

    def etud_eval_results(self, etud, e) -> dict:
        "dict resultats d'un étudiant à une évaluation"
        eval_notes = self.modimpls_evals_notes[e.moduleimpl_id][e.id]  # pd.Series
        notes_ok = eval_notes.where(eval_notes > scu.NOTES_ABSENCE).dropna()
        d = {
            "id": e.id,
            "description": e.description,
            "date": e.jour.isoformat() if e.jour else None,
            "heure_debut": e.heure_debut.strftime("%H:%M") if e.heure_debut else None,
            "heure_fin": e.heure_fin.strftime("%H:%M") if e.heure_debut else None,
            "coef": e.coefficient,
            "poids": {p.ue.acronyme: p.poids for p in e.ue_poids},
            "note": {
                "value": fmt_note(
                    self.modimpls_evals_notes[e.moduleimpl_id][e.id][etud.id]
                ),
                "min": fmt_note(notes_ok.min()),
                "max": fmt_note(notes_ok.max()),
                "moy": fmt_note(notes_ok.mean()),
            },
            "url": url_for(
                "notes.evaluation_listenotes",
                scodoc_dept=g.scodoc_dept,
                evaluation_id=e.id,
            ),
        }
        return d

    def bulletin_etud(self, etud, formsemestre) -> dict:
        """Le bulletin de l'étudiant dans ce semestre"""
        d = {
            "version": "0",
            "type": "BUT",
            "date": datetime.datetime.utcnow().isoformat() + "Z",
            "etudiant": etud.to_dict_bul(),
            "formation": {
                "id": formsemestre.formation.id,
                "acronyme": formsemestre.formation.acronyme,
                "titre_officiel": formsemestre.formation.titre_officiel,
                "titre": formsemestre.formation.titre,
            },
            "formsemestre_id": formsemestre.id,
            "ressources": self.etud_mods_results(etud, self.ressources),
            "saes": self.etud_mods_results(etud, self.saes),
            "ues": {ue.acronyme: self.etud_ue_results(etud, ue) for ue in self.ues},
            "semestre": {
                "notes": {  # moyenne des moyennes générales du semestre
                    "value": fmt_note(self.etud_moy_gen[etud.id]),
                    "min": fmt_note(self.etud_moy_gen.min()),
                    "moy": fmt_note(self.etud_moy_gen.mean()),
                    "max": fmt_note(self.etud_moy_gen.max()),
                },
                "rang": {  # classement wrt moyenne général, indicatif
                    "value": self.etud_moy_gen_ranks[etud.id],
                    "total": len(self.etuds),
                },
                "absences": {  # XXX TODO
                    "injustifie": 1,
                    "total": 33,
                },
                "etapes": {  # XXX TODO
                    # à spécifier: liste des étapes Apogée
                },
                "date_debut": formsemestre.date_debut.isoformat(),
                "date_fin": formsemestre.date_fin.isoformat(),
                "annee_universitaire": self.formsemestre.annee_scolaire_str(),
                "inscription": "TODO-MM-JJ",  # XXX TODO
                "numero": formsemestre.semestre_id,
                "decision": None,  # XXX TODO
                "situation": "Décision jury: Validé.  Diplôme obtenu.",  # XXX TODO
                "date_jury": "AAAA-MM-JJ",  # XXX TODO
                "groupes": [],  # XXX TODO
            },
            "options": bulletin_option_affichage(formsemestre),
        }
        return d


def bulletin_option_affichage(formsemestre):
    "dict avec les options d'affichages (préférences) pour ce semestre"
    prefs = sco_preferences.SemPreferences(formsemestre.id)
    fields = (
        "bul_show_abs",
        "bul_show_abs_modules",
        "bul_show_ects",
        "bul_show_codemodules",
        "bul_show_matieres",
        "bul_show_rangs",
        "bul_show_ue_rangs",
        "bul_show_mod_rangs",
        "bul_show_moypromo",
        "bul_show_minmax",
        "bul_show_minmax_mod",
        "bul_show_minmax_eval",
        "bul_show_coef",
        "bul_show_ue_cap_details",
        "bul_show_ue_cap_current",
        "bul_show_temporary",
        "bul_temporary_txt",
        "bul_show_uevalid",
        "bul_show_date_inscr",
    )
    # on enlève le "bul_" de la clé:
    return {field[4:]: prefs[field] for field in fields}


# Pour raccorder le code des anciens bulletins qui attendent une NoteTable
class APCNotesTableCompat:
    """Implementation partielle de NotesTable pour les formations APC
    Accès aux notes et rangs.
    """

    def __init__(self, formsemestre):
        self.results = ResultatsSemestreBUT(formsemestre)
        nb_etuds = len(self.results.etuds)
        self.rangs = self.results.etud_moy_gen_ranks
        self.moy_min = self.results.etud_moy_gen.min()
        self.moy_max = self.results.etud_moy_gen.max()
        self.moy_moy = self.results.etud_moy_gen.mean()
        self.bonus = defaultdict(lambda: 0.0)  # XXX
        self.ue_rangs = {
            u.id: (defaultdict(lambda: 0.0), nb_etuds) for u in self.results.ues
        }
        self.mod_rangs = {
            m.id: (defaultdict(lambda: 0), nb_etuds) for m in self.results.modimpls
        }

    def get_ues(self):
        ues = []
        for ue in self.results.ues:
            d = ue.to_dict()
            d.update(
                {
                    "max": self.results.etud_moy_ue[ue.id].max(),
                    "min": self.results.etud_moy_ue[ue.id].min(),
                    "moy": self.results.etud_moy_ue[ue.id].mean(),
                    "nb_moy": len(self.results.etud_moy_ue),
                }
            )
            ues.append(d)
        return ues

    def get_modimpls(self):
        return [m.to_dict() for m in self.results.modimpls]

    def get_etud_moy_gen(self, etudid):
        return self.results.etud_moy_gen[etudid]

    def get_moduleimpls_attente(self):
        return []  # XXX TODO

    def get_etud_rang(self, etudid):
        return self.rangs[etudid]

    def get_etud_rang_group(self, etudid, group_id):
        return (None, 0)  # XXX unimplemented TODO

    def get_etud_ue_status(self, etudid, ue_id):
        return {
            "cur_moy_ue": self.results.etud_moy_ue[ue_id][etudid],
            "is_capitalized": False,  # XXX TODO
        }

    def get_etud_mod_moy(self, moduleimpl_id, etudid):
        mod_idx = self.results.modimpl_coefs_df.columns.get_loc(moduleimpl_id)
        etud_idx = self.results.etud_index[etudid]
        # moyenne sur les UE:
        self.results.sem_cube[etud_idx, mod_idx].mean()

    def get_mod_stats(self, moduleimpl_id):
        return {
            "moy": "-",
            "max": "-",
            "min": "-",
            "nb_notes": "-",
            "nb_missing": "-",
            "nb_valid_evals": "-",
        }

    def get_evals_in_mod(self, moduleimpl_id):
        mi = ModuleImpl.query.get(moduleimpl_id)
        evals_results = []
        for e in mi.evaluations:
            d = e.to_dict()
            d["heure_debut"] = e.heure_debut  # datetime.time
            d["heure_fin"] = e.heure_fin
            d["jour"] = e.jour  # datetime
            d["notes"] = {
                etud.id: {
                    "etudid": etud.id,
                    "value": self.results.modimpls_evals_notes[e.moduleimpl_id][e.id][
                        etud.id
                    ],
                }
                for etud in self.results.etuds
            }
            evals_results.append(d)
        return evals_results