MonScoDocEssai/app/comp/moy_mod.py

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

##############################################################################
#
# Gestion scolarite IUT
#
# Copyright (c) 1999 - 2021 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É)

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 numpy as np
import pandas as pd
from pandas.core.frame import DataFrame

from app import db
from app import models
from app.models import ModuleImpl, Evaluation, EvaluationUEPoids
from app.scodoc import sco_utils as scu


def df_load_evaluations_poids(
    moduleimpl_id: int, default_poids=1.0
) -> 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 par default_poids.
    Résultat: (evals_poids, liste de UE du semestre)
    """
    modimpl = 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]
    df = pd.DataFrame(columns=ue_ids, index=evaluation_ids, dtype=float)
    for eval_poids in EvaluationUEPoids.query.join(
        EvaluationUEPoids.evaluation
    ).filter_by(moduleimpl_id=moduleimpl_id):
        df[eval_poids.ue_id][eval_poids.evaluation_id] = eval_poids.poids
    if default_poids is not None:
        df.fillna(value=default_poids, inplace=True)
    return df, ues


def check_moduleimpl_conformity(
    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.
    """
    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:
        raise ValueError("check_moduleimpl_conformity: nb ue incoherent")
    module_evals_poids = evals_poids.transpose().sum(axis=1).to_numpy() != 0
    check = all(
        (modules_coefficients[moduleimpl.module.id].to_numpy() != 0)
        == module_evals_poids
    )
    return check


def df_load_modimpl_notes(moduleimpl_id: int) -> tuple:
    """Construit un dataframe avec toutes les notes de toutes les évaluations du module.
    colonnes: le nom de la colonne est l'evaluation_id (int)
    index (lignes): etudid (int)

    Résultat: (evals_notes, liste de évaluations du moduleimpl,
              liste de booleens indiquant si l'évaluation est "complete")

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

    Les notes renvoyées 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

    L'é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 ne compte immédiate" (publish_incomplete)

    N'utilise pas de cache ScoDoc.
    """
    # L'index du dataframe est la liste des étudiants inscrits au semestre, sans les démissionnaires
    etudids = {
        e.etudid
        for e in ModuleImpl.query.get(moduleimpl_id).formsemestre.get_inscrits(
            include_dem=False
        )
    }
    evaluations = Evaluation.query.filter_by(moduleimpl_id=moduleimpl_id).all()
    # --- Calcul nombre d'inscrits pour détermnier si évaluation "complete":
    if evaluations:
        # on prend les inscrits au module ET au semestre (donc sans démissionnaires)
        inscrits_module = {
            ins.etud.id for ins in evaluations[0].moduleimpl.inscriptions
        }.intersection(etudids)
        nb_inscrits_module = len(inscrits_module)
    else:
        nb_inscrits_module = 0
    evals_notes = pd.DataFrame(index=etudids, dtype=float)  # empty df with all students
    evaluations_completes = []
    for evaluation in evaluations:
        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)])
        # is_complete ssi tous les inscrits (non dem) au semestre ont une note
        is_complete = (
            len(set(eval_df.index).intersection(etudids)) == nb_inscrits_module
        ) or evaluation.publish_incomplete
        evaluations_completes.append(is_complete)
        # NULL en base => ABS (= -999)
        eval_df.fillna(scu.NOTES_ABSENCE, inplace=True)
        # Ce merge met à NULL les élements non présents
        #  (notes non saisies ou etuds non inscrits au module):
        evals_notes = evals_notes.merge(
            eval_df, how="left", left_index=True, right_index=True
        )
    # Force columns names to integers (evaluation ids)
    evals_notes.columns = pd.Int64Index(
        [int(x) for x in evals_notes.columns], dtype="int64"
    )
    return evals_notes, evaluations, evaluations_completes


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

     - evals_notes : DataFrame, colonnes: EVALS, Lignes: etudid
        valeur: notes brutes, float ou NOTES_ATTENTE, NOTES_NEUTRALISE,
        NOTES_ABSENCE.
        Les NaN désignent les notes manquantes (non saisies).

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

     - evaluations: séquence d'évaluations (utilisées pour le coef et
        le barème)

     - evaluations_completes: séquence de booléens indiquant les
        évals à prendre en compte.

    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 à des notes)
         ne donnent pas de coef vers cette UE.
    """
    nb_etuds, nb_evals = evals_notes_df.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)
    # Coefficients des évaluations, met à zéro ceux des évals incomplètes:
    evals_coefs = (
        np.array(
            [e.coefficient for e in evaluations],
            dtype=float,
        )
        * evaluations_completes
    ).reshape(-1, 1)
    evals_poids = evals_poids_df.values * evals_coefs
    # -> evals_poids shape : (nb_evals, nb_ues)
    assert evals_poids.shape == (nb_evals, nb_ues)
    # Remplace les notes ATT, EXC, ABS, NaN par zéro et mets les notes sur 20:
    evals_notes = np.where(
        evals_notes_df.values > scu.NOTES_ABSENCE, evals_notes_df.values, 0.0
    ) / [e.note_max / 20.0 for e in evaluations]
    # Les poids des évals pour les é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([evals_notes_df.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] * 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)
    etuds_moy_module_df = pd.DataFrame(
        etuds_moy_module, index=evals_notes_df.index, columns=evals_poids_df.columns
    )
    return etuds_moy_module_df