ScoDoc/app/models/ues.py

"""ScoDoc 9 models : Unités d'Enseignement (UE)
"""

from flask import abort, g
import pandas as pd

from app import db, log
from app import models
from app.models import APO_CODE_STR_LEN
from app.models import SHORT_STR_LEN
from app.models.but_refcomp import ApcNiveau, ApcParcours
from app.models.modules import Module
from app.scodoc import sco_utils as scu


class UniteEns(models.ScoDocModel):
    """Unité d'Enseignement (UE)"""

    __tablename__ = "notes_ue"

    id = db.Column(db.Integer, primary_key=True)
    ue_id = db.synonym("id")
    formation_id = db.Column(db.Integer, db.ForeignKey("notes_formations.id"))
    acronyme = db.Column(db.Text(), nullable=False)
    numero = db.Column(db.Integer, nullable=False, default=0)  # ordre de présentation
    titre = db.Column(db.Text())
    # Le semestre_idx n'est pas un id mais le numéro du semestre: 1, 2, ...
    # En ScoDoc7 et pour les formations classiques, il est NULL
    # (le numéro du semestre étant alors déterminé par celui des modules de l'UE)
    # Pour les formations APC, il est obligatoire (de 1 à 6 pour le BUT):
    semestre_idx = db.Column(db.Integer, nullable=True, index=True)
    # Type d'UE: 0 normal ("fondamentale"), 1 "sport", 2 "projet et stage (LP)",
    # 4 "élective"
    type = db.Column(db.Integer, default=0, server_default="0")
    # Les UE sont "compatibles" (pour la capitalisation) ssi elles ont ^m code
    # note: la fonction SQL notes_newid_ucod doit être créée à part
    ue_code = db.Column(
        db.String(SHORT_STR_LEN),
        server_default=db.text("notes_newid_ucod()"),
        nullable=False,
    )
    ects = db.Column(db.Float)  # nombre de credits ECTS (sauf si parcours spécifié)
    is_external = db.Column(db.Boolean(), default=False, server_default="false")
    # id de l'element pedagogique Apogee correspondant:
    code_apogee = db.Column(db.String(APO_CODE_STR_LEN))
    # coef UE, utilise seulement si l'option use_ue_coefs est activée:
    coefficient = db.Column(db.Float)

    # id de l'élément Apogée du RCUE (utilisé pour les UEs de sem. pair du BUT)
    code_apogee_rcue = db.Column(db.String(APO_CODE_STR_LEN))
    # coef. pour le calcul de moyennes de RCUE. Par défaut, 1.
    coef_rcue = db.Column(db.Float, nullable=False, default=1.0, server_default="1.0")

    color = db.Column(db.Text())

    # BUT
    niveau_competence_id = db.Column(
        db.Integer, db.ForeignKey("apc_niveau.id", ondelete="SET NULL")
    )
    niveau_competence = db.relationship("ApcNiveau", back_populates="ues")

    # Une UE appartient soit à tous les parcours (tronc commun), soit à un sous-ensemble
    parcours = db.relationship(
        ApcParcours,
        secondary="ue_parcours",
        backref=db.backref("ues", lazy=True),
        order_by="ApcParcours.numero, ApcParcours.code",
    )

    # relations
    matieres = db.relationship("Matiere", lazy="dynamic", backref="ue")
    modules = db.relationship("Module", lazy="dynamic", backref="ue")
    dispense_ues = db.relationship(
        "DispenseUE",
        back_populates="ue",
        cascade="all, delete",
        passive_deletes=True,
    )

    def __repr__(self):
        return f"""<{self.__class__.__name__}(id={self.id}, formation_id={
            self.formation_id}, acronyme='{self.acronyme}', semestre_idx={
            self.semestre_idx} {
            'EXTERNE' if self.is_external else ''})>"""

    def clone(self):
        """Create a new copy of this ue, add to session.
        Ne copie pas le code, ni le code Apogée, ni les liens au réf. de comp.
        (parcours et niveau).
        """
        ue = UniteEns(
            formation_id=self.formation_id,
            acronyme=self.acronyme + "-copie",
            numero=self.numero,
            titre=self.titre,
            semestre_idx=self.semestre_idx,
            type=self.type,
            ue_code="",  # ne duplique pas le code
            ects=self.ects,
            is_external=self.is_external,
            code_apogee="",  # ne copie pas les codes Apo
            coefficient=self.coefficient,
            coef_rcue=self.coef_rcue,
            color=self.color,
        )
        db.session.add(ue)
        return ue

    @classmethod
    def convert_dict_fields(cls, args: dict) -> dict:
        """Convert fields from the given dict to model's attributes values. No side effect.

        args: dict with args in application.
        returns: dict to store in model's db.
        """
        args = args.copy()
        if "type" in args:
            args["type"] = int(args["type"] or 0)
        if "is_external" in args:
            args["is_external"] = scu.to_bool(args["is_external"])
        if "ects" in args:
            args["ects"] = float(args["ects"])

        return args

    def from_dict(self, args: dict, excluded: set[str] | None = None) -> bool:
        """Update object's fields given in dict. Add to session but don't commit.
        True if modification.
        - can't change formation nor niveau_competence
        """
        return super().from_dict(
            args,
            excluded=(excluded or set()) | {"formation_id", "niveau_competence_id"},
        )

    def to_dict(self, convert_objects=False, with_module_ue_coefs=True):
        """as a dict, with the same conversions as in ScoDoc7.
        If convert_objects, convert all attributes to native types
            (suitable for json encoding).
        """
        # cache car très utilisé par anciens codes
        key = (self.id, convert_objects, with_module_ue_coefs)
        _cache = getattr(g, "_ue_to_dict_cache", None)
        if _cache:
            result = g._ue_to_dict_cache.get(key, False)
            if result is not False:
                return result
        else:
            g._ue_to_dict_cache = {}
            _cache = g._ue_to_dict_cache

        e = dict(self.__dict__)
        e.pop("_sa_instance_state", None)
        e.pop("evaluation_ue_poids", None)
        # ScoDoc7 output_formators
        e["ue_id"] = self.id
        e["numero"] = e["numero"] if e["numero"] else 0
        e["coefficient"] = e["coefficient"] if e["coefficient"] else 0.0
        e["code_apogee"] = e["code_apogee"] or ""  # pas de None
        e["ects_by_parcours"] = {
            parcour.code: self.get_ects(parcour) for parcour in self.parcours
        }
        e["parcours"] = []
        for parcour in self.parcours:
            p_dict = parcour.to_dict(with_annees=False)
            ects = self.get_ects(parcour, only_parcours=True)
            if ects is not None:
                p_dict["ects"] = ects
            e["parcours"].append(p_dict)

        if with_module_ue_coefs:
            if convert_objects:
                e["module_ue_coefs"] = [
                    c.to_dict(convert_objects=True) for c in self.module_ue_coefs
                ]
        else:
            e.pop("module_ue_coefs", None)
        _cache[key] = e
        return e

    def annee(self) -> int:
        """L'année dans la formation (commence à 1).
        En APC seulement, en classic renvoie toujours 1.
        """
        return 1 if self.semestre_idx is None else (self.semestre_idx - 1) // 2 + 1

    def is_locked(self):
        """True if UE should not be modified
        (contains modules used in a locked formsemestre)
        """
        # XXX todo : à ré-écrire avec SQLAlchemy
        from app.scodoc import sco_edit_ue

        return sco_edit_ue.ue_is_locked(self.id)

    def can_be_deleted(self) -> bool:
        """True si l'UE n'a pas de moduleimpl rattachés
        (pas un seul module de cette UE n'a de modimpl)
        """
        return (self.modules.count() == 0) or not any(
            m.modimpls.all() for m in self.modules
        )

    def guess_semestre_idx(self) -> None:
        """Lorsqu'on prend une ancienne formation non APC,
        les UE n'ont pas d'indication de semestre.
        Cette méthode fixe le semestre en prenant celui du premier module,
        ou à défaut le met à 1.
        """
        if self.semestre_idx is None:
            module = self.modules.first()
            if module is None:
                self.semestre_idx = 1
            else:
                self.semestre_idx = module.semestre_id
            db.session.add(self)
            db.session.commit()

    def get_ects(self, parcour: ApcParcours = None, only_parcours=False) -> float:
        """Crédits ECTS associés à cette UE.
        En BUT, cela peut quelquefois dépendre du parcours.
        Si only_parcours, renvoie None si pas de valeur spéciquement définie dans
        le parcours indiqué.
        """
        if parcour is not None:
            key = (parcour.id, self.id, only_parcours)
            ue_ects_cache = getattr(g, "_ue_ects_cache", None)
            if ue_ects_cache:
                ects = g._ue_ects_cache.get(key, False)
                if ects is not False:
                    return ects
            else:
                g._ue_ects_cache = {}
                ue_ects_cache = g._ue_ects_cache
            ue_parcour = UEParcours.query.filter_by(
                ue_id=self.id, parcours_id=parcour.id
            ).first()
            if ue_parcour is not None and ue_parcour.ects is not None:
                ue_ects_cache[key] = ue_parcour.ects
                return ue_parcour.ects
            if only_parcours:
                ue_ects_cache[key] = None
                return None
        return self.ects

    def set_ects(self, ects: float, parcour: ApcParcours = None):
        """Fixe les crédits. Do not commit.
        Si le parcours n'est pas spécifié, affecte les ECTS par défaut de l'UE.
        Si ects est None et parcours indiqué, efface l'association.
        """
        if parcour is not None:
            ue_parcour = UEParcours.query.filter_by(
                ue_id=self.id, parcours_id=parcour.id
            ).first()
            if ects is None:
                if ue_parcour:
                    db.session.delete(ue_parcour)
            else:
                if ue_parcour is None:
                    ue_parcour = UEParcours(parcours_id=parcour.id, ue_id=self.id)
                ue_parcour.ects = float(ects)
                db.session.add(ue_parcour)
        else:
            self.ects = ects
        log(f"ue.set_ects( ue_id={self.id}, acronyme={self.acronyme}, ects={ects} )")
        db.session.add(self)

    @classmethod
    def get_ue(cls, ue_id: int, accept_none=False) -> "UniteEns":
        """UE ou 404 (ou None si accept_none),
        cherche uniquement dans le département courant.
        Si accept_none, return None si l'id est invalide ou inexistant.
        """
        if not isinstance(ue_id, int):
            try:
                ue_id = int(ue_id)
            except (TypeError, ValueError):
                if accept_none:
                    return None
                abort(404, "ue_id invalide")

        query = cls.query.filter_by(id=ue_id)
        if g.scodoc_dept:
            from app.models import Formation

            query = query.join(Formation).filter_by(dept_id=g.scodoc_dept_id)

        if accept_none:
            return query.first()
        return query.first_or_404()

    def get_ressources(self):
        "Liste des modules ressources rattachés à cette UE"
        return self.modules.filter_by(module_type=scu.ModuleType.RESSOURCE).all()

    def get_saes(self):
        "Liste des modules SAE rattachés à cette UE"
        return self.modules.filter_by(module_type=scu.ModuleType.SAE).all()

    def get_modules_not_apc(self):
        "Listes des modules non SAE et non ressource (standards, mais aussi bonus...)"
        return self.modules.filter(
            (Module.module_type != scu.ModuleType.SAE),
            (Module.module_type != scu.ModuleType.RESSOURCE),
        ).all()

    def get_codes_apogee(self) -> set[str]:
        """Les codes Apogée (codés en base comme "VRT1,VRT2")"""
        if self.code_apogee:
            return {x.strip() for x in self.code_apogee.split(",") if x}
        return set()

    def get_codes_apogee_rcue(self) -> set[str]:
        """Les codes Apogée RCUE (codés en base comme "VRT1,VRT2")"""
        if self.code_apogee_rcue:
            return {x.strip() for x in self.code_apogee_rcue.split(",") if x}
        return set()

    def _parcours_niveaux_ids(self, parcours=list[ApcParcours]) -> set[int]:
        """set des ids de niveaux communs à tous les parcours listés"""
        return set.intersection(
            *[
                {
                    n.id
                    for n in self.niveau_competence.niveaux_annee_de_parcours(
                        parcour, self.annee(), self.formation.referentiel_competence
                    )
                }
                for parcour in parcours
            ]
        )

    def check_niveau_unique_dans_parcours(
        self, niveau: ApcNiveau, parcours=list[ApcParcours]
    ) -> tuple[bool, str]:
        """Vérifie que
        - le niveau est dans au moins l'un des parcours listés;
        - et que l'un des parcours associé à cette UE ne contient pas
        déjà une UE associée au niveau donné dans une autre année.
        Renvoie: (True, "") si ok, sinon (False, message).
        """
        # Le niveau est-il dans l'un des parcours listés ?
        if parcours:
            if niveau.id not in self._parcours_niveaux_ids(parcours):
                log(
                    f"Le niveau {niveau} ne fait pas partie des parcours de l'UE {self}."
                )
                return (
                    False,
                    f"""Le niveau {
                    niveau.libelle} ne fait pas partie des parcours de l'UE {self.acronyme}.""",
                )

        for parcour in parcours or [None]:
            if parcour is None:
                code_parcour = "TC"
                ues_meme_niveau = [
                    ue
                    for ue in self.formation.query_ues_parcour(None).filter(
                        UniteEns.niveau_competence == niveau
                    )
                ]
            else:
                code_parcour = parcour.code
                ues_meme_niveau = [
                    ue
                    for ue in parcour.ues
                    if ue.id != self.id
                    and ue.formation_id == self.formation_id
                    and ue.niveau_competence_id == niveau.id
                ]
            if ues_meme_niveau:
                msg_parc = f"parcours {code_parcour}" if parcour else "tronc commun"
                if len(ues_meme_niveau) > 1:  # deja 2 UE sur ce niveau
                    msg = f"""Niveau "{
                            niveau.libelle}" déjà associé à deux UE du {msg_parc}"""
                    log(
                        f"check_niveau_unique_dans_parcours(niveau_id={niveau.id}): "
                        + msg
                    )
                    return False, msg
                # s'il y a déjà une UE associée à ce niveau, elle doit être dans l'autre semestre
                # de la même année scolaire
                other_semestre_idx = self.semestre_idx + (
                    2 * (self.semestre_idx % 2) - 1
                )
                if ues_meme_niveau[0].semestre_idx != other_semestre_idx:
                    msg = f"""Erreur: niveau "{
                            niveau.libelle}" déjà associé à une autre UE du semestre S{
                                ues_meme_niveau[0].semestre_idx} du {msg_parc}"""
                    log(
                        f"check_niveau_unique_dans_parcours(niveau_id={niveau.id}): "
                        + msg
                    )
                    return False, msg

        return True, ""

    def is_used_in_validation_rcue(self) -> bool:
        """Vrai si cette UE est utilisée dans une validation enregistrée d'RCUE."""
        from app.models.but_validations import ApcValidationRCUE

        return (
            ApcValidationRCUE.query.filter(
                db.or_(
                    ApcValidationRCUE.ue1_id == self.id,
                    ApcValidationRCUE.ue2_id == self.id,
                )
            ).count()
            > 0
        )

    def set_niveau_competence(self, niveau: ApcNiveau | None) -> tuple[bool, str]:
        """Associe cette UE au niveau de compétence indiqué.
        Le niveau doit être dans l'un des parcours de l'UE (si elle n'est pas
        de tronc commun).
        Assure que ce soit la seule dans son parcours.
        Sinon, raises ScoFormationConflict.

        Si niveau est None, désassocie.

        Si l'UE est utilisée dans un validation de RCUE, on ne peut plus la changer de niveau.

        Returns
        - True if (de)association done, False on error.
        - Error message (string)
        """
        # Sanity checks
        if not self.formation.referentiel_competence:
            return (
                False,
                "La formation n'est pas associée à un référentiel de compétences",
            )
        # UE utilisée dans des validations RCUE ?
        if self.is_used_in_validation_rcue():
            return (
                False,
                "UE utilisée dans un RCUE validé: son niveau ne peut plus être modifié",
            )
        if niveau is not None:
            if self.niveau_competence_id is not None:
                return (
                    False,
                    f"""{self.acronyme} déjà associée à un niveau de compétences ({
                        self.id}, {self.niveau_competence_id})""",
                )
            if (
                niveau.competence.referentiel.id
                != self.formation.referentiel_competence.id
            ):
                return (
                    False,
                    "Le niveau n'appartient pas au référentiel de la formation",
                )
            if niveau.id == self.niveau_competence_id:
                return True, ""  # nothing to do
            if self.niveau_competence_id is not None:
                ok, error_message = self.check_niveau_unique_dans_parcours(
                    niveau, self.parcours
                )
                if not ok:
                    return ok, error_message
        elif self.niveau_competence_id is None:
            return True, ""  # nothing to do
        self.niveau_competence = niveau
        db.session.add(self)
        db.session.commit()
        # Invalidation du cache
        self.formation.invalidate_cached_sems()
        log(f"ue.set_niveau_competence( {self}, {niveau} )")
        return True, ""

    def set_parcours(self, parcours: list[ApcParcours]) -> tuple[bool, str]:
        """Associe cette UE aux parcours indiqués.
        Si un niveau est déjà associé, vérifie sa cohérence.
        Renvoie (True, "") si ok, sinon (False, error_message)
        """
        msg = ""
        # Safety check
        if self.formation.referentiel_competence is None:
            return False, "pas de référentiel de compétence"
        # Si tous les parcours, aucun (tronc commun)
        if {p.id for p in parcours} == {
            p.id for p in self.formation.referentiel_competence.parcours
        }:
            parcours = []
        # Le niveau est-il dans tous ces parcours ? Sinon, l'enlève
        prev_niveau = self.niveau_competence
        if (
            parcours
            and self.niveau_competence
            and self.niveau_competence.id not in self._parcours_niveaux_ids(parcours)
        ):
            self.niveau_competence = None
            msg = " (niveau compétence désassocié !)"

        if parcours and self.niveau_competence:
            ok, error_message = self.check_niveau_unique_dans_parcours(
                self.niveau_competence, parcours
            )
            if not ok:
                self.formation.invalidate_cached_sems()
                self.niveau_competence = prev_niveau  # restore
                return False, error_message

        self.parcours = parcours
        db.session.add(self)
        db.session.commit()
        # Invalidation du cache
        self.formation.invalidate_cached_sems()
        log(f"ue.set_parcours( {self}, {parcours} )")
        return True, "parcours enregistrés" + msg

    def add_parcour(self, parcour: ApcParcours) -> tuple[bool, str]:
        """Ajoute ce parcours à ceux de l'UE"""
        if parcour.id in {p.id for p in self.parcours}:
            return True, ""  # déjà présent
        if parcour.referentiel.id != self.formation.referentiel_competence.id:
            return False, "Le parcours n'appartient pas au référentiel de la formation"

        return self.set_parcours(self.parcours + [parcour])


class UEParcours(db.Model):
    """Association ue <-> parcours, indiquant les ECTS"""

    __tablename__ = "ue_parcours"
    ue_id = db.Column(
        db.Integer,
        db.ForeignKey("notes_ue.id", ondelete="CASCADE"),
        primary_key=True,
    )
    parcours_id = db.Column(
        db.Integer,
        db.ForeignKey("apc_parcours.id", ondelete="CASCADE"),
        primary_key=True,
    )
    ects = db.Column(db.Float, nullable=True)  # si NULL, on prendra les ECTS de l'UE

    def __repr__(self):
        return f"<UEParcours( ue_id={self.ue_id}, parcours_id={self.parcours_id}, ects={self.ects})>"


class DispenseUE(db.Model):
    """Dispense d'UE
    Utilisé en APC (BUT) pour indiquer
    - les étudiants redoublants avec une UE capitalisée qu'ils ne refont pas.
    - les étudiants "non inscrit" à une UE car elle ne fait pas partie de leur Parcours.

    La dispense d'UE n'est PAS une validation:
    - elle n'est pas affectée par les décisions de jury (pas effacée)
    - elle est associée à un formsemestre
    - elle ne permet pas la délivrance d'ECTS ou du diplôme.

    On utilise cette dispense et non une "inscription" par souci d'efficacité:
    en général, la grande majorité des étudiants suivront toutes les UEs de leur parcours,
    la dispense étant une exception.
    """

    __tablename__ = "dispenseUE"
    __table_args__ = (db.UniqueConstraint("formsemestre_id", "ue_id", "etudid"),)
    id = db.Column(db.Integer, primary_key=True)
    formsemestre_id = formsemestre_id = db.Column(
        db.Integer, db.ForeignKey("notes_formsemestre.id"), index=True, nullable=True
    )
    ue_id = db.Column(
        db.Integer,
        db.ForeignKey(UniteEns.id, ondelete="CASCADE"),
        index=True,
        nullable=False,
    )
    ue = db.relationship("UniteEns", back_populates="dispense_ues")
    etudid = db.Column(
        db.Integer,
        db.ForeignKey("identite.id", ondelete="CASCADE"),
        index=True,
        nullable=False,
    )
    etud = db.relationship("Identite", back_populates="dispense_ues")

    def __repr__(self) -> str:
        return f"""<{self.__class__.__name__} {self.id} etud={
                repr(self.etud)} ue={repr(self.ue)}>"""

    @classmethod
    def load_formsemestre_dispense_ues_set(
        cls, formsemestre: "FormSemestre", etudids: pd.Index, ues: list[UniteEns]
    ) -> set[tuple[int, int]]:
        """Construit l'ensemble des
        etudids = modimpl_inscr_df.index,  # les etudids
        ue_ids : modimpl_coefs_df.index,  # les UE du formsemestre sans les UE bonus sport

        Résultat: set de (etudid, ue_id).
        """
        # Prend toutes les dispenses obtenues par des étudiants de ce formsemestre,
        # puis filtre sur inscrits et ues
        ue_ids = {ue.id for ue in ues}
        dispense_ues = {
            (dispense_ue.etudid, dispense_ue.ue_id)
            for dispense_ue in DispenseUE.query.filter_by(
                formsemestre_id=formsemestre.id
            )
            if dispense_ue.etudid in etudids and dispense_ue.ue_id in ue_ids
        }
        return dispense_ues