ScoDoc/app/pe/pe_jury.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
"""

# ----------------------------------------------------------
# Ensemble des fonctions et des classes
# permettant les calculs preliminaires (hors affichage)
# a l'edition d'un jury de poursuites d'etudes
# ----------------------------------------------------------

import io
import os
from zipfile import ZipFile


from app.comp import res_sem
from app.comp.res_compat import NotesTableCompat
from app.models import FormSemestre
from app.models.etudiants import Identite

from app.scodoc.gen_tables import GenTable, SeqGenTable
import app.scodoc.sco_utils as scu
from app.pe.pe_etudiant import EtudiantsJuryPE
from app.pe.pe_trajectoire import TrajectoiresJuryPE, Trajectoire
import app.pe.pe_comp as pe_comp
from app.pe.pe_semtag import SemestreTag
from app.pe.pe_interclasstag import AggregatInterclasseTag
from app.pe.pe_trajectoiretag import TrajectoireTag
import app.pe.pe_affichage as pe_affichage

import pandas as pd
import numpy as np

# ----------------------------------------------------------------------------------------


# ----------------------------------------------------------------------------------------
class JuryPE(object):
    """Classe mémorisant toutes les informations nécessaires pour établir un jury de PE.
        Modèle basé sur NotesTable.

        Attributs :

        * diplome : l'année d'obtention du diplome BUT et du jury de PE (généralement février XXXX)
        * juryEtudDict : dictionnaire récapitulant les étudiants participant au jury PE (données administratives +
          celles des semestres valides à prendre en compte permettant le calcul des moyennes  ...
          ``{'etudid : { 'nom', 'prenom', 'civilite', 'diplome', '',  }}``
    a
          Rq: il contient à la fois les étudiants qui vont être diplomés à la date prévue
          et ceux qui sont éliminés (abandon, redoublement, ...) pour affichage alternatif
    """

    # Variables de classe décrivant les aggrégats, leur ordre d'apparition temporelle et
    # leur affichage dans les avis latex

    # ------------------------------------------------------------------------------------------------------------------
    def __init__(self, diplome, formation_id):
        """
        Création d'une table PE sur la base d'un semestre selectionné. De ce semestre est déduit :
        1. l'année d'obtention du DUT,
        2. tous les étudiants susceptibles à ce stade (au regard de leur parcours) d'être diplomés.

        Args:
            sem_base: le FormSemestre donnant le semestre à la base du jury PE
            semBase: le dictionnaire sem donnant la base du jury (CB: TODO: A supprimer à long term)
            meme_programme: si True, impose un même programme pour tous les étudiants participant au jury,
                            si False, permet des programmes differents
        """
        self.diplome = diplome
        "L'année du diplome"

        self.formation_id = formation_id
        "La formation associée au diplome"

        self.nom_export_zip = f"Jury_PE_{self.diplome}"
        "Nom du zip où ranger les fichiers générés"

        self.zipdata = io.BytesIO()
        with ZipFile(self.zipdata, "w") as zipfile:
            # Chargement des étudiants à prendre en compte dans le jury
            pe_comp.pe_print(
                f"""*** Recherche et chargement des étudiants diplômés en {
                    self.diplome} pour la formation {self.formation_id}"""
            )
            self.etudiants = EtudiantsJuryPE(
                self.diplome
            )  # Les infos sur les étudiants
            self.etudiants.find_etudiants(self.formation_id)
            self.diplomes_ids = self.etudiants.diplomes_ids

            # Génère les semestres taggués (avec le calcul des moyennes) pour le jury PE
            pe_comp.pe_print("*** Génère les semestres taggués")
            self.semestres_taggues = compute_semestres_tag(self.etudiants)

            if pe_comp.PE_DEBUG:
                # Intègre le bilan des semestres taggués au zip final
                for formsemestretag in self.semestres_taggues.values():
                    filename = formsemestretag.nom.replace(" ", "_") + ".csv"
                    pe_comp.pe_print(f"    - Export csv de {filename} ")
                    self.add_file_to_zip(
                        zipfile,
                        filename,
                        formsemestretag.str_tagtable(),
                        path="details_semestres",
                    )

            # Génère les trajectoires (combinaison de semestres suivis
            # par un étudiant pour atteindre le semestre final d'un aggrégat)
            pe_comp.pe_print(
                "*** Génère les trajectoires (différentes combinaisons de semestres) des étudiants"
            )
            self.trajectoires = TrajectoiresJuryPE(self.diplome)
            self.trajectoires.cree_trajectoires(self.etudiants)

            # Génère les moyennes par tags des trajectoires
            pe_comp.pe_print(
                "*** Calcule les moyennes par tag des trajectoires possibles"
            )
            self.trajectoires_tagguees = compute_trajectoires_tag(
                self.trajectoires, self.etudiants, self.semestres_taggues
            )

            if pe_comp.PE_DEBUG:
                # Intègre le bilan des trajectoires tagguées au zip final
                for trajectoire_tagguee in self.trajectoires_tagguees.values():
                    filename = trajectoire_tagguee.get_repr().replace(" ", "_") + ".csv"
                    pe_comp.pe_print(f"    - Export csv de {filename} ")
                    self.add_file_to_zip(
                        zipfile,
                        filename,
                        trajectoire_tagguee.str_tagtable(),
                        path="details_semestres",
                    )

            # Génère les interclassements (par promo et) par (nom d') aggrégat
            pe_comp.pe_print("*** Génère les interclassements par aggrégat")
            self.interclassements_taggues = compute_interclassements(
                self.etudiants, self.trajectoires, self.trajectoires_tagguees
            )

            if pe_comp.PE_DEBUG:
                # Intègre le bilan des aggrégats (par promo) au zip final
                for interclass_tag in self.interclassements_taggues.values():
                    filename = interclass_tag.get_repr().replace(" ", "_") + ".csv"
                    pe_comp.pe_print(f"    - Export csv de {filename} ")
                    self.add_file_to_zip(
                        zipfile,
                        filename,
                        interclass_tag.str_tagtable(),
                        path="details_semestres",
                    )

            # Synthèse des éléments du jury PE
            self.synthese = self.synthetise_juryPE()

            # Export des données => mode 1 seule feuille -> supprimé
            pe_comp.pe_print("*** Export du jury de synthese")
            output = io.BytesIO()

            with pd.ExcelWriter(output, engine="openpyxl") as writer:
                for onglet, df in self.synthese.items():
                    # écriture dans l'onglet:
                    df.to_excel(writer, onglet, index=True, header=True)
            output.seek(0)

            self.add_file_to_zip(
                zipfile, f"synthese_jury_{self.diplome}.xlsx", output.read()
            )

        # Fin !!!! Tada :)

    def add_file_to_zip(self, zipfile: ZipFile, filename: str, data, path=""):
        """Add a file to given zip
        All files under NOM_EXPORT_ZIP/
        path may specify a subdirectory

        Args:
             zipfile: ZipFile
             filename: Le nom du fichier à intégrer au zip
             data: Les données du fichier
             path: Un dossier dans l'arborescence du zip
        """
        path_in_zip = os.path.join(path, filename)  # self.nom_export_zip,
        zipfile.writestr(path_in_zip, data)

    def get_zipped_data(self) -> io.BytesIO | None:
        """returns file-like data with a zip of all generated (CSV) files.
        Warning: reset stream to the begining.
        """
        self.zipdata.seek(0)
        return self.zipdata

    def do_tags_list(self, interclassements: dict[str, AggregatInterclasseTag]):
        """La liste des tags extraites des interclassements"""
        tags = []
        for aggregat in interclassements:
            interclass = interclassements[aggregat]
            if interclass.tags_sorted:
                tags.extend(interclass.tags_sorted)
        tags = sorted(set(tags))
        return tags

    # **************************************************************************************************************** #
    # Méthodes pour la synthèse du juryPE
    # *****************************************************************************************************************

    def synthetise_juryPE(self):
        """Synthétise tous les résultats du jury PE dans des dataframes"""

        pe_comp.pe_print("*** Synthèse finale des moyennes ***")

        synthese = {}
        pe_comp.pe_print("  -> Synthèse des données administratives")
        synthese["administratif"] = self.df_administratif()

        tags = self.do_tags_list(self.interclassements_taggues)
        for tag in tags:
            pe_comp.pe_print(f"  -> Synthèse du tag {tag}")
            synthese[tag] = self.df_tag(tag)
        return synthese

    def df_administratif(self):
        """Synthétise toutes les données administratives des étudiants"""

        etudids = list(self.diplomes_ids)

        # Récupération des  données des étudiants
        administratif = {}
        nbre_semestres_max = self.etudiants.nbre_etapes_max_diplomes()

        for etudid in etudids:
            etudiant = self.etudiants.identites[etudid]
            cursus = self.etudiants.cursus[etudid]
            formsemestres = cursus["formsemestres"]

            if cursus["diplome"]:
                diplome = cursus["diplome"]
            else:
                diplome = "indéterminé"

            administratif[etudid] = {
                "Nom": etudiant.nom,
                "Prenom": etudiant.prenom,
                "Civilite": etudiant.civilite_str,
                "Age": pe_comp.calcul_age(etudiant.date_naissance),
                "Date d'entree": cursus["entree"],
                "Date de diplome": diplome,
                "Nbre de semestres": len(formsemestres),
            }

            # Ajout des noms de semestres parcourus
            etapes = pe_affichage.etapes_du_cursus(formsemestres, nbre_semestres_max)
            administratif[etudid] |= etapes

        # Construction du dataframe
        df = pd.DataFrame.from_dict(administratif, orient="index")

        # Tri par nom/prénom
        df.sort_values(by=["Nom", "Prenom"], inplace=True)
        return df

    def df_tag(self, tag):
        """Génère le DataFrame synthétisant les moyennes/classements (groupe,
        interclassement promo) pour tous les aggrégats prévus,
        tels que fourni dans l'excel final.

        Args:
            tag: Un des tags (a minima `but`)

        Returns:
        """

        etudids = list(self.diplomes_ids)
        aggregats = pe_comp.TOUS_LES_PARCOURS

        donnees = {}

        for etudid in etudids:
            etudiant = self.etudiants.identites[etudid]
            donnees[etudid] = {
                "Nom": etudiant.nom,
                "Prenom": etudiant.prenom,
                "Civilite": etudiant.civilite_str,
            }

            for aggregat in aggregats:
                """La trajectoire de l'étudiant sur l'aggrégat"""
                trajectoire = self.trajectoires.suivi[etudid][aggregat]
                """Les moyennes par tag de cette trajectoire"""
                if trajectoire:
                    trajectoire_tagguee = self.trajectoires_tagguees[
                        trajectoire.trajectoire_id
                    ]
                    bilan = trajectoire_tagguee.moyennes_tags[tag]

                    donnees[etudid] |= {
                        f"{aggregat} notes ": f"{bilan['notes'].loc[etudid]:.1f}",
                        f"{aggregat} class. (groupe)": f"{bilan['classements'].loc[etudid]}/{bilan['nb_inscrits']}",
                        f"{aggregat} min/moy/max (groupe)": f"{bilan['min']:.1f}/{bilan['moy']:.1f}/{bilan['max']:.1f}",
                    }
                else:
                    donnees[etudid] |= {
                        f"{aggregat} notes ": "-",
                        f"{aggregat} class. (groupe)": "-",
                        f"{aggregat} min/moy/max (groupe)": "-",
                    }

                """L'interclassement"""
                interclass = self.interclassements_taggues[aggregat]
                if tag in interclass.moyennes_tags:
                    bilan = interclass.moyennes_tags[tag]

                    donnees[etudid] |= {
                        f"{aggregat} class. (promo)": f"{bilan['classements'].loc[etudid]}/{bilan['nb_inscrits']}",
                        f"{aggregat} min/moy/max (promo)": f"{bilan['min']:.1f}/{bilan['moy']:.1f}/{bilan['max']:.1f}",
                    }
                else:
                    donnees[etudid] |= {
                        f"{aggregat} class. (promo)": "-",
                        f"{aggregat} min/moy/max (promo)": "-",
                    }

            # Fin de l'aggrégat
        # Construction du dataFrame
        df = pd.DataFrame.from_dict(donnees, orient="index")

        # Tri par nom/prénom
        df.sort_values(by=["Nom", "Prenom"], inplace=True)
        return df

    def table_syntheseJury(self, mode="singlesheet"):  #  was str_syntheseJury
        """Table(s) du jury
        mode: singlesheet ou multiplesheet pour export excel
        """
        sT = SeqGenTable()  # le fichier excel à générer

        if mode == "singlesheet":
            return sT.get_genTable("singlesheet")
        else:
            return sT


def compute_semestres_tag(etudiants: EtudiantsJuryPE) -> dict:
    """Créé les semestres taggués, de type 'S1', 'S2', ..., pour un groupe d'étudiants donnés.
    Chaque semestre taggué est rattaché à l'un des FormSemestre faisant partie du cursus scolaire
    des étudiants (cf. attribut etudiants.cursus).
    En crééant le semestre taggué, sont calculées les moyennes/classements par tag associé.
        .

    Args:
        etudiants: Un groupe d'étudiants participant au jury

    Returns:
        Un dictionnaire {fid: SemestreTag(fid)}
    """

    """Création des semestres taggués, de type 'S1', 'S2', ..."""
    pe_comp.pe_print("*** Création des semestres taggués")

    formsemestres = etudiants.get_formsemestres(
        semestres_recherches=pe_comp.TOUS_LES_SEMESTRES
    )

    semestres_tags = {}
    for frmsem_id, formsemestre in formsemestres.items():
        """Choix d'un nom pour le semestretag"""
        nom = "S%d %d %d-%d" % (
            formsemestre.semestre_id,
            frmsem_id,
            formsemestre.date_debut.year,
            formsemestre.date_fin.year,
        )

        pe_comp.pe_print(f"  --> Semestre taggué {nom} sur la base de {formsemestre}")

        # Crée le semestre_tag et exécute les calculs de moyennes
        formsemestretag = SemestreTag(nom, frmsem_id)

        # Stocke le semestre taggué
        semestres_tags[frmsem_id] = formsemestretag

    return semestres_tags


def compute_trajectoires_tag(
    trajectoires: TrajectoiresJuryPE,
    etudiants: EtudiantsJuryPE,
    semestres_taggues: dict[int, SemestreTag],
):
    """Créée les trajectoires tagguées (combinaison aggrégeant plusieurs semestres au sens
    d'un aggrégat (par ex: '3S')),
    en calculant les moyennes et les classements par tag pour chacune.

    Pour rappel : Chaque trajectoire est identifiée un nom d'aggrégat et par un formsemestre terminal.

    Par exemple :

    * combinaisons '3S' : S1+S2+S3 en prenant en compte tous les S3 qu'ont fréquenté les
      étudiants du jury PE. Ces S3 marquent les formsemestre terminal de chaque combinaison.

    * combinaisons 'S2' : 1 seul S2 pour des étudiants n'ayant pas redoublé, 2 pour des redoublants (dont les
      notes seront moyennées sur leur 2 semestres S2). Ces combinaisons ont pour formsemestre le dernier S2 en
      date (le S2 redoublé par les redoublants est forcément antérieur)


    Args:
        etudiants: Les données des étudiants
        semestres_tag: Les semestres tag (pour lesquels des moyennes par tag ont été calculés)

    Return:
        Un dictionnaire de la forme {nom_aggregat: {fid_terminal: SetTag(fid_terminal)} }
    """

    pe_comp.pe_print(" *** Création des aggrégats ")

    trajectoires_tagguees = {}

    for trajectoire_id in trajectoires.trajectoires:
        trajectoire = trajectoires.trajectoires[trajectoire_id]
        nom = trajectoire.get_repr()

        pe_comp.pe_print(f"   --> Fusion {nom}")

        """Création de la trajectoire_tagguee associée"""
        trajectoire_tagguee = TrajectoireTag(
            nom, trajectoire, semestres_taggues, etudiants
        )

        """Mémorise le résultat"""
        trajectoires_tagguees[trajectoire_id] = trajectoire_tagguee

    return trajectoires_tagguees


def compute_interclassements(
    etudiants: EtudiantsJuryPE,
    trajectoires_jury_pe: TrajectoiresJuryPE,
    trajectoires_tagguees: dict[tuple, Trajectoire],
):
    """Interclasse les étudiants, (nom d') aggrégat par aggrégat,
    pour fournir un classement sur la promo. Le classement est établi au regard du nombre
    d'étudiants ayant participé au même aggrégat.
    """
    pe_comp.pe_print(" Interclassement sur la promo")

    aggregats_interclasses_taggues = {}
    for nom_aggregat in pe_comp.TOUS_LES_SEMESTRES + pe_comp.TOUS_LES_AGGREGATS:
        pe_comp.pe_print(f"   --> {nom_aggregat}")
        interclass = AggregatInterclasseTag(
            nom_aggregat, etudiants, trajectoires_jury_pe, trajectoires_tagguees
        )
        aggregats_interclasses_taggues[nom_aggregat] = interclass
    return aggregats_interclasses_taggues