ScoDoc-PE/app/pe/moys/pe_rcstag.py

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

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

##############################################################################
#  Module "Avis de poursuite d'étude"
#  conçu et développé par Cléo Baras (IUT de Grenoble)
##############################################################################

"""
Created on Fri Sep  9 09:15:05 2016

@author: barasc
"""

from app.models import FormSemestre
from app.pe import pe_affichage
import pandas as pd
import numpy as np
from app.pe.rcss import pe_rcs, pe_rcsemx
import app.pe.moys.pe_sxtag as pe_sxtag
import app.pe.pe_comp as pe_comp
from app.pe.moys import pe_tabletags, pe_moytag
from app.scodoc.sco_utils import ModuleType


class RCSemXTag(pe_tabletags.TableTag):
    def __init__(
        self,
        rcsemx: pe_rcsemx.RCSemX,
        sxstags: dict[(str, int) : pe_sxtag.SxTag],
        semXs_suivis: dict[int, dict],
    ):
        """Calcule les moyennes par tag (orientées compétences)
        d'un regroupement de SxTag, pour extraire les classements par tag pour un
        groupe d'étudiants donnés. Le groupe d'étudiants est formé par ceux ayant tous
        participé au même semestre terminal.

        Args:
            rcsemx: Le RCSemX (identifié par un nom et l'id de son semestre terminal)
            sxstags: Les données sur les SemX taggués
            semXs_suivis: Les données indiquant quels SXTags sont à prendre en compte
                          pour chaque étudiant
        """
        pe_tabletags.TableTag.__init__(self)

        self.rcs_id: tuple(str, int) = rcsemx.rcs_id
        """Identifiant du RCSemXTag (identique au RCSemX sur lequel il s'appuie)"""

        self.rcsemx: pe_rcsemx.RCSemX = rcsemx
        """Le regroupement RCSemX associé au RCSemXTag"""

        self.semXs_suivis = semXs_suivis
        """Les semXs suivis par les étudiants"""

        self.nom = self.get_repr()
        """Représentation textuelle du RSCtag"""

        # Les données du semestre final
        self.formsemestre_final: FormSemestre = rcsemx.formsemestre_final
        """Le semestre final"""
        self.fid_final: int = rcsemx.formsemestre_final.formsemestre_id
        """Le fid du semestre final"""

        # Affichage pour debug
        pe_affichage.pe_print(f"*** {self.get_repr(verbose=True)}")

        # Les données aggrégés (RCRCF + SxTags)
        self.semXs_aggreges: dict[(str, int) : pe_rcsemx.RCSemX] = rcsemx.semXs_aggreges
        """Les SemX aggrégés"""
        self.sxstags_aggreges = {}
        """Les SxTag associés aux SemX aggrégés"""
        try:
            for rcf_id in self.semXs_aggreges:
                self.sxstags_aggreges[rcf_id] = sxstags[rcf_id]
        except:
            raise ValueError("Semestres SxTag manquants")
        self.sxtags_connus = sxstags  # Tous les sxstags connus

        # Les étudiants (etuds, états civils & etudis)
        sems_dans_aggregat = rcsemx.aggregat
        sxtag_final = self.sxstags_aggreges[(sems_dans_aggregat[-1], self.rcs_id[1])]
        self.etuds = sxtag_final.etuds
        """Les étudiants (extraits du semestre final)"""
        self.add_etuds(self.etuds)
        self.etudids_sorted = sorted(self.etudids)
        """Les étudids triés"""

        # Les compétences (extraites de tous les Sxtags)
        self.acronymes_ues_to_competences = self._do_acronymes_to_competences()
        """L'association acronyme d'UEs -> compétence (extraites des SxTag aggrégés)"""

        self.competences_sorted = sorted(
            set(self.acronymes_ues_to_competences.values())
        )
        """Compétences (triées par nom, extraites des SxTag aggrégés)"""
        aff = pe_affichage.repr_comp_et_ues(self.acronymes_ues_to_competences)
        pe_affichage.pe_print(f"--> Compétences : {aff}")

        # Les tags
        self.tags_sorted = self._do_taglist()
        """Tags extraits de tous les SxTag aggrégés"""
        aff_tag = ["👜" + tag for tag in self.tags_sorted]
        pe_affichage.pe_print(f"--> Tags : {', '.join(aff_tag)}")

        # Les moyennes
        self.moyennes_tags: dict[str, pe_moytag.MoyennesTag] = {}

        """Synthétise les moyennes/classements par tag (qu'ils soient personnalisé ou de compétences)"""
        for tag in self.tags_sorted:
            pe_affichage.pe_print(f"--> Moyennes du tag 👜{tag}")

            # Cubes d'inscription (etudids_sorted x compétences_sorted x sxstags),
            # de notes et de coeffs pour la moyenne générale
            # en "aggrégant" les données des sxstags, compétence par compétence
            (
                inscr_df,
                inscr_cube,  # données d'inscriptions
                notes_df,
                notes_cube,  # notes
                coeffs_df,
                coeffs_cube,  # coeffs pour la moyenne générale (par UEs)
                coeffs_rcues_df,
                coeffs_rcues_cube,  # coeffs pour la moyenne de regroupement d'UEs
            ) = self.assemble_cubes(tag)

            # Calcule les moyennes, et synthétise les coeffs
            (
                moys_competences,
                matrice_coeffs_moy_gen,
            ) = self.compute_notes_et_coeffs_competences(
                notes_cube, coeffs_cube, coeffs_rcues_cube, inscr_cube
            )

            # Affichage des coeffs
            aff = pe_affichage.repr_profil_coeffs(
                matrice_coeffs_moy_gen, with_index=True
            )
            pe_affichage.pe_print(f"     > Moyenne calculée avec pour coeffs : {aff}")

            # Mémorise les moyennes et les coeff associés
            self.moyennes_tags[tag] = pe_moytag.MoyennesTag(
                tag,
                pe_moytag.CODE_MOY_COMPETENCES,
                moys_competences,
                matrice_coeffs_moy_gen,
            )

    def __eq__(self, other):
        """Egalité de 2 RCS taggués sur la base de leur identifiant"""
        return self.rcs_id == other.sxtag_id

    def get_repr(self, verbose=True) -> str:
        """Renvoie une représentation textuelle (celle de la trajectoire sur laquelle elle
        est basée)"""
        if verbose:
            return f"{self.__class__.__name__} basé sur " + self.rcsemx.get_repr(
                verbose=verbose
            )
        else:
            return f"{self.__class__.__name__} {self.rcs_id}"

    def assemble_cubes(self, tag):
        """Pour un tag donné, construit les cubes :
        * d'inscriptions aux compétences (etudid x competences x SxTag)
        * de notes (etudid x competences x SxTag)
        * des coeffs pour le calcul des moyennes générales par UE (etudid x competences x SxTag)
        * des coeffs pour le calcul des regroupements cohérents d'UE/compétences

        nécessaires au calcul des moyennes, en :

        * transformant les données des UEs en données de compétences (changement de noms)
        * fusionnant les données d'un même semestre, lorsque plusieurs UEs traitent d'une même compétence (cas des RCSx = Sx)
        * aggrégeant les données de compétences sur plusieurs semestres (cas des RCSx = xA ou xS)

        Args:
            tag: Le tag visé
        """
        #         etudids_sorted: list[int],
        #         competences_sorted: list[str],
        #         sxstags: dict[(str, int) : pe_sxtag.SxTag],

        inscriptions_dfs = {}
        notes_dfs = {}
        coeffs_moy_gen_dfs = {}
        coeffs_rcue_dfs = {}

        for sxtag_id, sxtag in self.sxstags_aggreges.items():
            # Partant de dataframes vierges
            inscription_df = pd.DataFrame(
                np.nan, index=self.etudids_sorted, columns=self.competences_sorted
            )
            notes_df = pd.DataFrame(
                np.nan, index=self.etudids_sorted, columns=self.competences_sorted
            )
            coeffs_moy_gen_df = pd.DataFrame(
                np.nan, index=self.etudids_sorted, columns=self.competences_sorted
            )
            coeffs_rcue_df = pd.DataFrame(
                np.nan, index=self.etudids_sorted, columns=self.competences_sorted
            )

            # Charge les données du semestre tag (copie car changement de nom de colonnes à venir)
            if tag in sxtag.moyennes_tags:  # si le tag est présent dans le semestre
                moys_tag = sxtag.moyennes_tags[tag]

                # Les inscr, les notes, les coeffs
                acro_ues_inscr_parcours = sxtag.acro_ues_inscr_parcours
                notes = moys_tag.matrice_notes
                coeffs_moy_gen = moys_tag.matrice_coeffs  # les coeffs
                coeffs_rcues = sxtag.coefs_rcue  # dictionnaire UE -> coeff

                # Traduction des acronymes d'UE en compétences
                # comp_to_ues = pe_comp.asso_comp_to_accronymes(self.acronymes_ues_to_competences)
                acronymes_ues_columns = notes.columns
                for acronyme in acronymes_ues_columns:
                    # La compétence visée
                    competence = self.acronymes_ues_to_competences[acronyme]  # La comp

                    # Les étud inscrits à la comp reportés dans l'inscription au RCSemX
                    comp_inscr = acro_ues_inscr_parcours[
                        acro_ues_inscr_parcours.notnull()
                    ].index
                    etudids_communs = list(
                        inscription_df.index.intersection(comp_inscr)
                    )
                    inscription_df.loc[
                        etudids_communs, competence
                    ] = acro_ues_inscr_parcours.loc[etudids_communs, acronyme]

                    # Les étud ayant une note à l'acronyme de la comp (donc à la comp)
                    etuds_avec_notes = notes[notes[acronyme].notnull()].index
                    etudids_communs = list(
                        notes_df.index.intersection(etuds_avec_notes)
                    )
                    notes_df.loc[etudids_communs, competence] = notes.loc[
                        etudids_communs, acronyme
                    ]

                    # Les coeffs pour la moyenne générale
                    etuds_avec_coeffs = coeffs_moy_gen[
                        coeffs_moy_gen[acronyme].notnull()
                    ].index
                    etudids_communs = list(
                        coeffs_moy_gen_df.index.intersection(etuds_avec_coeffs)
                    )
                    coeffs_moy_gen_df.loc[
                        etudids_communs, competence
                    ] = coeffs_moy_gen.loc[etudids_communs, acronyme]

                    # Les coeffs des RCUE reportés là où les étudiants ont des notes
                    etuds_avec_notes = notes[notes[acronyme].notnull()].index
                    etudids_communs = list(
                        notes_df.index.intersection(etuds_avec_notes)
                    )
                    coeffs_rcue_df.loc[etudids_communs, competence] = coeffs_rcues[
                        acronyme
                    ]

            # Supprime tout ce qui n'est pas numérique
            # for col in notes_df.columns:
            #    notes_df[col] = pd.to_numeric(notes_df[col], errors="coerce")

            # Stocke les dfs
            inscriptions_dfs[sxtag_id] = inscription_df
            notes_dfs[sxtag_id] = notes_df
            coeffs_moy_gen_dfs[sxtag_id] = coeffs_moy_gen_df
            coeffs_rcue_dfs[sxtag_id] = coeffs_rcue_df

        # Réunit les inscriptions sous forme d'un cube etudids x competences x semestres
        sxtag_x_etudids_x_comps = [
            inscriptions_dfs[sxtag_id] for sxtag_id in self.sxstags_aggreges
        ]
        inscriptions_etudids_x_comps_x_sxtag = np.stack(
            sxtag_x_etudids_x_comps, axis=-1
        )

        # Réunit les notes sous forme d'un cube etudids x competences x semestres
        sxtag_x_etudids_x_comps = [
            notes_dfs[sxtag_id] for sxtag_id in self.sxstags_aggreges
        ]
        notes_etudids_x_comps_x_sxtag = np.stack(sxtag_x_etudids_x_comps, axis=-1)

        # Réunit les coeffs sous forme d'un cube etudids x competences x semestres
        sxtag_x_etudids_x_comps = [
            coeffs_moy_gen_dfs[sxtag_id] for sxtag_id in self.sxstags_aggreges
        ]
        coeffs_etudids_x_comps_x_sxtag = np.stack(sxtag_x_etudids_x_comps, axis=-1)

        # Normalise les coeffs de rcue par année (pour que le poids des années soit le
        # même)
        # Réunit les coeffs sous forme d'un cube etudids x competences x semestres
        sxtag_x_etudids_x_comps = [
            coeffs_rcue_dfs[sxtag_id] for sxtag_id in self.sxstags_aggreges
        ]
        coeffs_rcues_etudids_x_comps_x_sxtag = np.stack(
            sxtag_x_etudids_x_comps, axis=-1
        )

        return (
            inscriptions_dfs,
            inscriptions_etudids_x_comps_x_sxtag,
            notes_dfs,
            notes_etudids_x_comps_x_sxtag,
            coeffs_moy_gen_dfs,
            coeffs_etudids_x_comps_x_sxtag,
            coeffs_rcue_dfs,
            coeffs_rcues_etudids_x_comps_x_sxtag,
        )

    def _do_taglist(self) -> list[str]:
        """Synthétise les tags à partir des Sxtags aggrégés.

        Returns:
            Liste de tags triés par ordre alphabétique
        """
        tags = []
        for frmsem_id in self.sxstags_aggreges:
            tags.extend(self.sxstags_aggreges[frmsem_id].tags_sorted)
        return sorted(set(tags))

    def _do_acronymes_to_competences(self) -> dict[str:str]:
        """Synthétise l'association complète {acronyme_ue: competences}
        extraite de toutes les données/associations des SxTags
        aggrégés.

        Returns:
            Un dictionnaire {'acronyme_ue' : 'compétences'}
        """
        dict_competences = {}
        for sxtag_id, sxtag in self.sxstags_aggreges.items():
            dict_competences |= sxtag.acronymes_ues_to_competences
        return dict_competences

    def compute_notes_et_coeffs_competences(
        self,
        notes_cube: np.array,
        coeffs_cube: np.array,
        coeffs_rcue: np.array,
        inscr_mask: np.array,
    ):
        """Calcule la moyenne par UEs|Compétences en moyennant sur les semestres et renvoie les résultats (notes
        et coeffs) sous la forme de DataFrame"""
        (etud_moy_tag, coeff_tag) = compute_moyennes_par_RCS(
            notes_cube, coeffs_cube, coeffs_rcue, inscr_mask
        )

        etud_moy_tag_df = pd.DataFrame(
            etud_moy_tag,
            index=self.etudids_sorted,  # les etudids
            columns=self.competences_sorted,  # les competences
        )

        coeffs_df = pd.DataFrame(
            coeff_tag, index=self.etudids_sorted, columns=self.competences_sorted
        )

        return etud_moy_tag_df, coeffs_df


def compute_moyennes_par_RCS(
    notes_cube: np.array,
    coeffs_cube: np.array,
    coeffs_rcue: np.array,
    inscr_mask: np.array,
):
    """Partant d'une série de notes (fournie sous forme d'un cube
    etudids_sorted x UEs|Compétences x semestres)
    chaque note étant pondérée par un coeff dépendant du semestre (fourni dans coeffs_rcue),
    et pondérée par un coeff dépendant de l'UE|Compétence pour calculer une moyenne générale
    (fourni dans coeffs_cube),
    calcule :
    * la moyenne par UEs|Compétences sur plusieurs semestres (partant du set_cube).
    * les coeffs "cumulés" à appliquer pour le calcul de la moyenne générale

    La moyenne est un nombre (note/20), ou NaN si pas de notes disponibles

    Args:
        notes_cube: notes moyennes aux compétences ndarray
                    (etuds_sorted x UEs|compétences x sxtags),
                    des floats avec des NaN
        coeffs_cube: coeffs appliqués aux compétences dans le calcul des moyennes générales,
                    (etuds_sorted x UEs|compétences x sxtags),
                    des floats avec des NaN
        coeffs_rcue_cube: coeffs des RCUE appliqués dans les moyennes de RCS
        inscr_mask: inscriptions aux compétences ndarray
                    (etuds_sorted x UEs|compétences x sxtags),
                    des 1.0 (si inscrit) et des NaN (si non inscrit)
    Returns:
        Un DataFrame avec pour columns les moyennes par tags,
        et pour rows les etudid
    """
    # Applique le masque d'inscriptions aux notes et aux coeffs
    notes_significatives = notes_cube * inscr_mask
    coeffs_moy_gen_significatifs = coeffs_cube * inscr_mask
    coeffs_rcues_significatifs = coeffs_rcue * inscr_mask

    # Enlève les NaN des cubes pour les entrées manquantes
    notes_no_nan = np.nan_to_num(notes_significatives, nan=0.0)
    coeffs_no_nan = np.nan_to_num(coeffs_moy_gen_significatifs, nan=0.0)

    # Les moyennes par tag
    with np.errstate(invalid="ignore"):  # ignore les 0/0 (-> NaN)
        # Quelles entrées contiennent des notes et des coeffs?
        mask = ~np.isnan(notes_significatives) & ~np.isnan(coeffs_rcues_significatifs)

        # La moyenne (pondérée) sur le regroupement cohérent de semestres
        coeffs_rcues_non_nan = np.nan_to_num(coeffs_rcues_significatifs * mask, nan=0.0)
        notes_ponderes = notes_no_nan * coeffs_rcues_non_nan
        etud_moy_tag = np.sum(notes_ponderes, axis=2) / np.sum(
            coeffs_rcues_non_nan, axis=2
        )

        # Les coeffs pour la moyenne générale
        coeffs_pris_en_compte = coeffs_moy_gen_significatifs * mask
        coeffs_no_nan = np.nan_to_num(coeffs_pris_en_compte, nan=0.0)
        coeff_tag = np.sum(coeffs_no_nan, axis=2)

        # Le masque des inscriptions prises en compte
        inscr_prise_en_compte = inscr_mask * mask
        inscr_prise_en_compte = np.nan_to_num(inscr_prise_en_compte, nan=-1.0)
        inscr_tag = np.max(inscr_prise_en_compte, axis=2)
        inscr_tag[inscr_tag < 0] = np.NaN  # fix les max non calculés (-1) -> Na?

    # Le dataFrame des notes moyennes, en réappliquant le masque des inscriptions
    etud_moy_tag = etud_moy_tag * inscr_tag
    # Le dataFrame des coeffs pour la moyenne générale, en réappliquant le masque des inscriptions
    coeff_tag = coeff_tag * inscr_tag  # Réapplique le masque des inscriptions

    return (etud_moy_tag, coeff_tag)