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

456 lines
19 KiB
Python

# -*- 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.noms_semestres_aggreges
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
infos = {"aggregat": self.rcs_id[0], "cohorte": pe_moytag.CHAMP_GROUPE}
self.moyennes_tags[tag] = pe_moytag.MoyennesTag(
tag,
pe_moytag.CODE_MOY_COMPETENCES,
moys_competences,
matrice_coeffs_moy_gen,
infos,
)
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)