iziram/ScoDoc
1
0
Fork 0
forked from ScoDoc/ScoDoc
Code Issues 12 Pull Requests Projects 1 Releases Wiki Activity
6955c36ff5
ScoDoc/app/scodoc/sco_assiduites.py

985 lines
33 KiB
Python
Raw Blame History

"""
Ecrit par Matthias Hartmann.
"""
from datetime import date, datetime, time, timedelta
from pytz import UTC
from flask import g
from flask_sqlalchemy.query import Query
from app import log, db, set_sco_dept
from app.models import (
Evaluation,
Identite,
FormSemestre,
FormSemestreInscription,
ModuleImpl,
ModuleImplInscription,
ScoDocSiteConfig,
)
from app.models.assiduites import Assiduite, Justificatif
from app.scodoc import sco_formsemestre_inscriptions
from app.scodoc import sco_preferences
from app.scodoc import sco_cache
from app.scodoc import sco_etud
import app.scodoc.sco_utils as scu
class CountCalculator:
"""
La classe CountCalculator est conçue pour gérer le comptage des assiduités,
en calculant le nombre total de jours complets,
de demi-journées, et d'heures passées sur une période donnée.
Elle prend en compte les jours non travaillés,
les horaires de travail standard et les assiduités s'étendant sur plusieurs jours.
Utilisation :
------------
1. Initialisation : La classe peut être initialisée avec des horaires personnalisés
pour le matin, le midi et le soir, ainsi qu'une durée de pause déjeuner.
Si non spécifiés, les valeurs par défaut seront
chargées depuis la configuration `ScoDocSiteConfig`.
Exemple d'initialisation :
calculator = CountCalculator(
morning="08:00",
noon="13:00",
evening="18:00",
nb_heures_par_jour=8
)
2. Ajout d'assiduités :
Exemple d'ajout d'assiduité :
- calculator.compute_assiduites(etudiant.assiduites)
- calculator.compute_assiduites([
<Assiduite>,
<Assiduite>,
<Assiduite>,
<Assiduite>
])
3. Accès aux métriques : Après l'ajout des assiduités,
on peut accéder aux métriques telles que :
le nombre total de jours, de demi-journées et d'heures calculées.
Exemple d'accès aux métriques :
metrics = calculator.to_dict()
4.Réinitialisation du comptage: Si besoin on peut réinitialiser
le compteur sans perdre la configuration
(horaires personnalisés)
Exemple de réinitialisation :
calculator.reset()
Méthodes Principales :
---------------------
- reset() : Réinitialise les compteurs de la classe.
- add_half_day(day: date, is_morning: bool) : Ajoute une demi-journée au comptage.
- add_day(day: date) : Ajoute un jour complet au comptage.
- compute_long_assiduite(assi: Assiduite) : Traite les assiduités
s'étendant sur plus d'un jour.
- compute_assiduites(assiduites: Query | list) : Calcule les métriques pour
une collection d'assiduités.
- to_dict() : Retourne les métriques sous forme de dictionnaire.
Notes :
------
Détails des calculs des heures:
Pour les assiduités courtes (<= 1 jour):
heures = assi.deb - assi.fin
Pour les assiduités longues (> 1 jour):
heures =
heures(assi.deb => fin_journee)
nb_heure_par_jour * (nb_jours-2) +
heures(assi.fin => fin_journee)
"""
def __init__(
self,
morning: str = None,
noon: str = None,
evening: str = None,
nb_heures_par_jour: int = None,
) -> None:
self.morning: time = str_to_time(
morning if morning else ScoDocSiteConfig.get("assi_morning_time", "08:00")
)
# Date pivot pour déterminer les demi-journées
self.noon: time = str_to_time(
noon if noon else ScoDocSiteConfig.get("assi_lunch_time", "13:00")
)
self.evening: time = str_to_time(
evening if evening else ScoDocSiteConfig.get("assi_afternoon_time", "18:00")
)
self.non_work_days: list[scu.NonWorkDays] = (
scu.NonWorkDays.get_all_non_work_days(dept_id=g.scodoc_dept_id)
)
# Sera utilisé pour les assiduités longues (> 1 journée)
self.nb_heures_par_jour = (
nb_heures_par_jour
if nb_heures_par_jour
else sco_preferences.get_preference(
"nb_heures_par_jour", dept_id=g.scodoc_dept_id
)
)
self.data = {}
self.reset()
def reset(self):
"""Remet à zero les compteurs"""
self.data = {
"total": {
"journee": [],
"demi": [],
"heure": 0,
"compte": 0,
},
"absent": {
"journee": [],
"demi": [],
"heure": 0,
"compte": 0,
},
"absent_just": {
"journee": [],
"demi": [],
"heure": 0,
"compte": 0,
},
"absent_non_just": {
"journee": [],
"demi": [],
"heure": 0,
"compte": 0,
},
"retard": {
"journee": [],
"demi": [],
"heure": 0,
"compte": 0,
},
"retard_just": {
"journee": [],
"demi": [],
"heure": 0,
"compte": 0,
},
"retard_non_just": {
"journee": [],
"demi": [],
"heure": 0,
"compte": 0,
},
"present": {
"journee": [],
"demi": [],
"heure": 0,
"compte": 0,
},
}
def get_count_key(self, etat: scu.EtatAssiduite, justi: bool = False) -> str:
"""Récupère une clé de dictionnaire en fonction de l'état de l'assiduité
et si elle est justifié
"""
keys: dict[scu.EtatAssiduite, str] = {
scu.EtatAssiduite.ABSENT: "absent",
scu.EtatAssiduite.RETARD: "retard",
scu.EtatAssiduite.PRESENT: "present",
}
count_key: str = keys.get(etat)
if etat != scu.EtatAssiduite.PRESENT:
count_key += "_just" if justi else "_non_just"
return count_key
def add_half_day(self, day: date, assi: Assiduite, is_morning: bool = True):
"""Ajoute une demi-journée dans le comptage"""
key: tuple[date, bool] = (day, is_morning)
count_key: str = self.get_count_key(assi.etat, assi.est_just)
if assi.etat != scu.EtatAssiduite.PRESENT:
_key: str = scu.EtatAssiduite.inverse().get(assi.etat).name.lower()
if key not in self.data[_key]["demi"]:
self.data[_key]["demi"].append(key)
if key not in self.data["total"]["demi"]:
self.data["total"]["demi"].append(key)
if key not in self.data[count_key]["demi"]:
self.data[count_key]["demi"].append(key)
def add_day(self, day: date, assi: Assiduite):
"""Ajoute un jour dans le comptage"""
count_key: str = self.get_count_key(assi.etat, assi.est_just)
if assi.etat != scu.EtatAssiduite.PRESENT:
key: str = scu.EtatAssiduite.inverse().get(assi.etat).name.lower()
if day not in self.data[key]["journee"]:
self.data[key]["journee"].append(day)
if day not in self.data["total"]["journee"]:
self.data["total"]["journee"].append(day)
if day not in self.data[count_key]["journee"]:
self.data[count_key]["journee"].append(day)
def add_hours(self, hours: float, assi: Assiduite):
"""Ajoute des heures dans le comptage"""
count_key: str = self.get_count_key(assi.etat, assi.est_just)
if assi.etat != scu.EtatAssiduite.PRESENT:
self.data[scu.EtatAssiduite.inverse().get(assi.etat).name.lower()][
"heure"
] += hours
self.data[count_key]["heure"] += hours
self.data["total"]["heure"] += hours
def add_count(self, assi: Assiduite):
"""Ajoute 1 count dans le comptage"""
count_key: str = self.get_count_key(assi.etat, assi.est_just)
if assi.etat != scu.EtatAssiduite.PRESENT:
self.data[scu.EtatAssiduite.inverse().get(assi.etat).name.lower()][
"compte"
] += 1
self.data[count_key]["compte"] += 1
self.data["total"]["compte"] += 1
def is_in_morning(self, period: tuple[datetime, datetime]) -> bool:
"""Vérifiée si la période donnée fait partie du matin
(Test sur la date de début)
"""
interval_morning: tuple[datetime, datetime] = (
scu.localize_datetime(datetime.combine(period[0].date(), self.morning)),
scu.localize_datetime(datetime.combine(period[0].date(), self.noon)),
)
in_morning: bool = scu.is_period_overlapping(
period, interval_morning, bornes=False
)
return in_morning
def is_in_evening(self, period: tuple[datetime, datetime]) -> bool:
"""Vérifie si la période fait partie de l'aprèm
(test sur la date de début)
"""
interval_evening: tuple[datetime, datetime] = (
scu.localize_datetime(
datetime.combine(period[0].date(), self.noon) + timedelta(seconds=1)
),
scu.localize_datetime(datetime.combine(period[0].date(), self.evening)),
)
in_evening: bool = scu.is_period_overlapping(period, interval_evening)
return in_evening
def compute_long_assiduite(self, assi: Assiduite):
"""Calcule les métriques sur une assiduité longue (plus d'un jour)"""
pointer_date: date = assi.date_debut.date() + timedelta(days=1)
self.add_day(assi.date_debut.date(), assi)
self.add_day(assi.date_fin.date(), assi)
start_period: tuple[datetime, datetime] = (
assi.date_debut,
scu.localize_datetime(
datetime.combine(assi.date_debut.date(), self.evening)
),
)
finish_period: tuple[datetime, datetime] = (
scu.localize_datetime(datetime.combine(assi.date_fin.date(), self.morning)),
assi.date_fin,
)
for period in (start_period, finish_period):
if self.is_in_evening(period):
self.add_half_day(period[0].date(), assi, False)
if self.is_in_morning(period):
self.add_half_day(period[0].date(), assi)
while pointer_date < assi.date_fin.date():
if pointer_date.weekday() not in self.non_work_days:
self.add_day(pointer_date, assi)
self.add_half_day(pointer_date, assi)
self.add_half_day(pointer_date, assi, False)
self.add_hours(self.nb_heures_par_jour, assi)
pointer_date += timedelta(days=1)
# Gestion des heures des dates de début et des dates de fin
deb_hours = (start_period[1] - start_period[0]).total_seconds() / 3600
fin_hours = (finish_period[1] - finish_period[0]).total_seconds() / 3600
self.add_hours(deb_hours + fin_hours, assi)
def compute_assiduites(self, assiduites: Query | list):
"""Calcule les métriques pour la collection d'assiduité donnée"""
assi: Assiduite
for assi in assiduites:
# Ajout vérification workday
# (Si préférence mise après avoir déjà noté des assiduités)
if assi.date_debut.weekday() in self.non_work_days:
continue
self.add_count(assi)
delta: timedelta = assi.date_fin - assi.date_debut
if delta.days > 0:
self.compute_long_assiduite(assi)
continue
period: tuple[datetime, datetime] = (assi.date_debut, assi.date_fin)
deb_date: date = assi.date_debut.date()
if self.is_in_morning(period):
self.add_half_day(deb_date, assi)
if self.is_in_evening(period):
self.add_half_day(deb_date, assi, False)
self.add_day(deb_date, assi)
self.add_hours(delta.total_seconds() / 3600, assi)
self.setup_data()
def setup_data(self):
"""Met en forme les données
pour les journées et les demi-journées :
au lieu d'avoir list[str] on a le nombre (len(list[str]))
"""
for value in self.data.values():
value["journee"] = len(value["journee"])
value["demi"] = len(value["demi"])
def to_dict(self, only_total: bool = True) -> dict[str, int | float]:
"""Retourne les métriques sous la forme d'un dictionnaire"""
return self.data["total"] if only_total else self.data
def str_to_time(time_str: str) -> time:
"""Convertit une chaîne de caractères représentant une heure en objet time
exemples :
- "08:00" -> time(8, 0)
- "18:00:00" -> time(18, 0, 0)
"""
return time(*list(map(int, time_str.split(":"))))
def get_assiduites_stats(
assiduites: Query, metric: str = "all", filtered: dict[str, object] = None
) -> dict[str, int | float]:
"""
Calcule les statistiques sur les assiduités
(nombre de jours, demi-journées et heures passées,
non justifiées, justifiées et total)
Les filtres :
- etat : filtre les assiduités par leur état
valeur : (absent, present, retard)
- date_debut/date_fin : prend les assiduités qui se trouvent entre les dates
valeur : datetime.datetime
- moduleimpl_id : filtre les assiduités en fonction du moduleimpl_id
valeur : int | None
- formsemestre : prend les assiduités du formsemestre donné
valeur : FormSemestre
- formsemestre_modimpls : prend les assiduités avec un moduleimpl du formsemestre
valeur : FormSemestre
- est_just : filtre les assiduités en fonction de si elles sont justifiées ou non
valeur : bool
- user_id : filtre les assiduités en fonction de l'utilisateur qui les a créées
valeur : int
- split : effectue un comptage par état d'assiduité
valeur : str (du moment que la clé est présente dans filtered)
Les métriques :
- journee : comptage en nombre de journée
- demi : comptage en nombre de demi journée
- heure : comptage en heure
- compte : nombre d'objets
- all : renvoi toute les métriques
"""
if filtered is not None:
deb, fin = None, None
for key in filtered:
match key:
case "etat":
assiduites = filter_assiduites_by_etat(assiduites, filtered[key])
case "date_fin":
fin = filtered[key]
case "date_debut":
deb = filtered[key]
case "moduleimpl_id":
assiduites = filter_by_module_impl(assiduites, filtered[key])
case "formsemestre":
assiduites = filter_by_formsemestre(
assiduites, Assiduite, filtered[key]
)
case "formsemestre_modimpls":
assiduites = filter_by_modimpls(
assiduites, Assiduite, filtered[key]
)
case "est_just":
assiduites = filter_assiduites_by_est_just(
assiduites, filtered[key]
)
case "user_id":
assiduites = filter_by_user_id(assiduites, filtered[key])
if (deb, fin) != (None, None):
assiduites = filter_by_date(assiduites, Assiduite, deb, fin)
metrics: list[str] = metric.split(",")
output: dict = {}
calculator: CountCalculator = CountCalculator()
calculator.compute_assiduites(assiduites)
# S'il n'y a pas de filtre ou que le filtre split n'est pas dans les filtres
if filtered is None or "split" not in filtered:
# On récupère le comptage total
# only_total permet de ne récupérer que le total
count: dict = calculator.to_dict(only_total=True)
# On ne garde que les métriques demandées
for key, val in count.items():
if key in metrics:
output[key] = val
# On renvoie le total si on a rien demandé (ou que metrics == ["all"])
return output if output else count
# Préparation du dictionnaire de retour avec les valeurs du calcul
count: dict = calculator.to_dict(only_total=False)
# Récupération des états depuis la saisie utilisateur
etats: list[str] = (
filtered["etat"].split(",") if "etat" in filtered else scu.EtatAssiduite.all()
)
for etat in etats:
# On vérifie que l'état est bien un état d'assiduité
# sinon on passe à l'état suivant
if not scu.EtatAssiduite.contains(etat):
continue
# On récupère le comptage pour chaque état
if etat != "present":
output[etat] = count[etat]
output[etat]["justifie"] = count[etat + "_just"]
output[etat]["non_justifie"] = count[etat + "_non_just"]
else:
output[etat] = count[etat]
output["total"] = count["total"]
# le dictionnaire devrait ressembler à :
# {
# "absent": {
# "journee": 1,
# "demi": 2,
# "heure": 3,
# "compte": 4,
# "justifie": {
# "journee": 1,
# "demi": 2,
# "heure": 3,
# "compte": 4
# },
# "non_justifie": {
# "journee": 1,
# "demi": 2,
# "heure": 3,
# "compte": 4
# }
# },
# ...
# "total": {
# "journee": 1,
# "demi": 2,
# "heure": 3,
# "compte": 4
# }
# }
return output
def filter_assiduites_by_etat(assiduites: Assiduite, etat: str) -> Query:
"""
Filtrage d'une collection d'assiduites en fonction de leur état
"""
etats: list[str] = list(etat.split(","))
etats = [scu.EtatAssiduite.get(e, -1) for e in etats]
return assiduites.filter(Assiduite.etat.in_(etats))
def filter_assiduites_by_est_just(assiduites: Assiduite, est_just: bool) -> Query:
"""
Filtrage d'une collection d'assiduites en fonction de s'ils sont justifiés
"""
return assiduites.filter(Assiduite.est_just == est_just)
def filter_by_user_id(
collection: Assiduite | Justificatif,
user_id: int,
) -> Query:
"""
Filtrage d'une collection en fonction de l'user_id
"""
return collection.filter_by(user_id=user_id)
def filter_by_date(
collection: Assiduite | Justificatif,
collection_cls: Assiduite | Justificatif,
date_deb: datetime = None,
date_fin: datetime = None,
strict: bool = False,
) -> Query:
"""
Filtrage d'une collection d'assiduites en fonction d'une date
"""
if date_deb is None:
date_deb = datetime.min
if date_fin is None:
date_fin = datetime.max
if not strict:
return collection.filter(
collection_cls.date_debut <= date_fin, collection_cls.date_fin >= date_deb
)
return collection.filter(
collection_cls.date_debut < date_fin, collection_cls.date_fin > date_deb
)
def filter_justificatifs_by_etat(justificatifs: Query, etat: str) -> Query:
"""
Filtrage d'une collection de justificatifs en fonction de leur état
"""
etats: list[str] = list(etat.split(","))
etats = [scu.EtatJustificatif.get(e, -1) for e in etats]
return justificatifs.filter(Justificatif.etat.in_(etats))
def filter_by_module_impl(assiduites: Assiduite, module_impl_id: int | None) -> Query:
"""
Filtrage d'une collection d'assiduites en fonction de l'ID du module_impl
"""
return assiduites.filter(Assiduite.moduleimpl_id == module_impl_id)
def filter_by_formsemestre(
collection_query: Assiduite | Justificatif,
collection_class: Assiduite | Justificatif,
formsemestre: FormSemestre,
) -> Query:
"""
Filtrage d'une collection : conserve les élements tels que
- l'étudiant est inscrit au formsemestre
- et la plage de dates intersecte celle du formsemestre
"""
if formsemestre is None:
return collection_query.filter(False)
collection_result = (
collection_query.join(Identite, collection_class.etudid == Identite.id)
.join(
FormSemestreInscription,
Identite.id == FormSemestreInscription.etudid,
)
.filter(FormSemestreInscription.formsemestre_id == formsemestre.id)
)
collection_result = collection_result.filter(
collection_class.date_debut <= formsemestre.date_fin
).filter(collection_class.date_fin >= formsemestre.date_debut)
return collection_result
def filter_by_modimpls(
collection_query: Assiduite | Justificatif,
collection_class: Assiduite | Justificatif,
formsemestre: FormSemestre,
) -> Query:
"""
Filtrage d'une collection d'assiduités: conserve les élements
- si l'étudiant est inscrit au formsemestre
- Et que l'assiduité concerne un moduleimpl de ce formsemestre
Ne fait rien sur les justificatifs.
Ne fait rien si formsemestre is None
"""
if (collection_class != Assiduite) or (formsemestre is None):
return collection_query
# restreint aux inscrits:
collection_result = (
collection_query.join(Identite, collection_class.etudid == Identite.id)
.join(
FormSemestreInscription,
Identite.id == FormSemestreInscription.etudid,
)
.filter(FormSemestreInscription.formsemestre_id == formsemestre.id)
)
collection_result = (
collection_result.join(ModuleImpl)
.join(ModuleImplInscription)
.filter(ModuleImplInscription.etudid == collection_class.etudid)
)
return collection_result
def justifies(justi: Justificatif, obj: bool = False) -> list[int] | Query:
"""
Retourne la liste des assiduite_id qui sont justifié par la justification
Une assiduité est justifiée si elle est COMPLETEMENT ou PARTIELLEMENT
comprise dans la plage du justificatif
et que l'état du justificatif est "valide".
Renvoie des id si obj == False, sinon les Assiduités
"""
if justi.etat != scu.EtatJustificatif.VALIDE:
return []
assiduites_query: Assiduite = Assiduite.query.filter_by(etudid=justi.etudid)
assiduites_query = assiduites_query.filter(
Assiduite.date_debut >= justi.date_debut, Assiduite.date_fin <= justi.date_fin
)
if not obj:
return [assi.id for assi in assiduites_query.all()]
return assiduites_query
def get_all_justified(
etudid: int,
date_deb: datetime = None,
date_fin: datetime = None,
moduleimpl_id: int = None,
) -> Query:
"""Retourne toutes les assiduités justifiées sur une période"""
if date_deb is None:
date_deb = datetime.min
if date_fin is None:
date_fin = datetime.max
date_deb = scu.localize_datetime(date_deb)
date_fin = scu.localize_datetime(date_fin)
justified: Query = Assiduite.query.filter_by(est_just=True, etudid=etudid)
if moduleimpl_id is not None:
justified = justified.filter_by(moduleimpl_id=moduleimpl_id)
after = filter_by_date(
justified,
Assiduite,
date_deb,
date_fin,
)
return after
def create_absence_billet(
date_debut: datetime,
date_fin: datetime,
etudid: int,
description: str = None,
est_just: bool = False,
) -> int:
"""
Permet de rapidement créer une absence.
**UTILISÉ UNIQUEMENT POUR LES BILLETS**
Ne pas utiliser autre par.
TALK: Vérifier si nécessaire
"""
etud: Identite = Identite.query.filter_by(etudid=etudid).first_or_404()
assiduite_unique: Assiduite = Assiduite.create_assiduite(
etud=etud,
date_debut=date_debut,
date_fin=date_fin,
etat=scu.EtatAssiduite.ABSENT,
description=description,
)
db.session.add(assiduite_unique)
db.session.commit()
if est_just:
justi = Justificatif.create_justificatif(
etudiant=etud,
date_debut=date_debut,
date_fin=date_fin,
etat=scu.EtatJustificatif.VALIDE,
raison=description,
)
db.session.add(justi)
db.session.commit()
justi.justifier_assiduites()
calculator: CountCalculator = CountCalculator()
calculator.compute_assiduites([assiduite_unique])
return calculator.to_dict()["demi"]
def get_evaluation_assiduites(evaluation: Evaluation) -> Query:
"""
Renvoie une query d'assiduité en fonction des étudiants inscrits à l'évaluation
et de la date de l'évaluation.
Attention : Si l'évaluation n'a pas de date, renvoie None
"""
# Evaluation sans date
if evaluation.date_debut is None:
return None
# Récupération des étudiants inscrits à l'évaluation
etuds: Query = Identite.query.join(
ModuleImplInscription, Identite.id == ModuleImplInscription.etudid
).filter(ModuleImplInscription.moduleimpl_id == evaluation.moduleimpl_id)
etudids: list[int] = [etud.id for etud in etuds]
# Récupération des assiduités des étudiants inscrits à l'évaluation
date_debut: datetime = evaluation.date_debut
date_fin: datetime
if evaluation.date_fin is not None:
date_fin = evaluation.date_fin
else:
# On met à la fin de la journée de date_debut
date_fin = datetime.combine(date_debut.date(), time.max)
# Filtrage par rapport à la plage de l'évaluation
assiduites: Query = Assiduite.query.filter(
Assiduite.date_debut >= date_debut,
Assiduite.date_fin <= date_fin,
Assiduite.etudid.in_(etudids),
)
return assiduites
def get_etud_evaluations_assiduites(etud: Identite) -> list[dict]:
"""
Retourne la liste des évaluations d'un étudiant. Pour chaque évaluation,
retourne la liste des assiduités concernant la plage de l'évaluation.
"""
etud_evaluations_assiduites: list[dict] = []
# On récupère les moduleimpls puis les évaluations liés aux moduleimpls
modsimpl_ids: list[int] = [
modimp_inscr.moduleimpl_id
for modimp_inscr in ModuleImplInscription.query.filter_by(etudid=etud.id)
]
evaluations: Query = Evaluation.query.filter(
Evaluation.moduleimpl_id.in_(modsimpl_ids)
)
# Pour chaque évaluation, on récupère l'assiduité de l'étudiant sur la plage
# de l'évaluation
for evaluation in evaluations:
eval_assis: dict = {"evaluation_id": evaluation.id, "assiduites": []}
# Pas d'assiduités si pas de date
if evaluation.date_debut is None:
continue
date_debut: datetime = evaluation.date_debut
date_fin: datetime
if evaluation.date_fin is not None:
date_fin = evaluation.date_fin
else:
# On met à la fin de la journée de date_debut
date_fin = datetime.combine(date_debut.date(), time.max)
# Filtrage par rapport à la plage de l'évaluation
assiduites: Query = etud.assiduites.filter(
Assiduite.date_debut >= date_debut,
Assiduite.date_fin <= date_fin,
)
# On récupère les assiduités et on met à jour le dictionnaire
eval_assis["assiduites"] = [
assi.to_dict(format_api=True) for assi in assiduites
]
# On ajoute le dictionnaire à la liste des évaluations
etud_evaluations_assiduites.append(eval_assis)
return etud_evaluations_assiduites
# Gestion du cache
def get_assiduites_count(etudid: int, sem: dict) -> tuple[int, int, int]:
"""Les comptes d'absences de cet étudiant dans ce semestre:
tuple (nb abs non justifiées, nb abs justifiées, nb abs total)
Utilise un cache.
"""
metrique = sco_preferences.get_preference("assi_metrique", sem["formsemestre_id"])
return get_assiduites_count_in_interval(
etudid,
sem["date_debut_iso"],
sem["date_fin_iso"],
scu.translate_assiduites_metric(metrique),
)
def formsemestre_get_assiduites_count(
etudid: int, formsemestre: FormSemestre, moduleimpl_id: int = None
) -> tuple[int, int, int]:
"""Les comptes d'absences de cet étudiant dans ce semestre:
tuple (nb abs non justifiées, nb abs justifiées, nb abs total)
Utilise un cache (si moduleimpl_id n'est pas spécifié).
"""
metrique = sco_preferences.get_preference("assi_metrique", formsemestre.id)
return get_assiduites_count_in_interval(
etudid,
date_debut=scu.localize_datetime(
datetime.combine(formsemestre.date_debut, time(0, 0))
),
date_fin=scu.localize_datetime(
datetime.combine(formsemestre.date_fin, time(23, 0))
),
metrique=scu.translate_assiduites_metric(metrique),
moduleimpl_id=moduleimpl_id,
)
def get_assiduites_count_in_interval(
etudid,
date_debut_iso: str = "",
date_fin_iso: str = "",
metrique="demi",
date_debut: datetime = None,
date_fin: datetime = None,
moduleimpl_id: int = None,
) -> tuple[int, int, int]:
"""Les comptes d'absences de cet étudiant entre ces deux dates, incluses:
tuple (nb abs non justifiées, nb abs justifiées, nb abs total)
On peut spécifier les dates comme datetime ou iso.
Utilise un cache (si moduleimpl_id n'est pas spécifié).
"""
date_debut_iso = date_debut_iso or date_debut.strftime("%Y-%m-%d")
date_fin_iso = date_fin_iso or date_fin.strftime("%Y-%m-%d")
key = f"{etudid}_{date_debut_iso}_{date_fin_iso}_assiduites"
r = sco_cache.AbsSemEtudCache.get(key)
if not r or moduleimpl_id is not None:
date_debut: datetime = date_debut or datetime.fromisoformat(date_debut_iso)
date_fin: datetime = date_fin or datetime.fromisoformat(date_fin_iso)
assiduites: Query = Assiduite.query.filter_by(etudid=etudid)
assiduites = assiduites.filter(Assiduite.etat == scu.EtatAssiduite.ABSENT)
assiduites = filter_by_date(assiduites, Assiduite, date_debut, date_fin)
if moduleimpl_id is not None:
assiduites = assiduites.filter_by(moduleimpl_id=moduleimpl_id)
calculator: CountCalculator = CountCalculator()
calculator.compute_assiduites(assiduites)
calcul: dict = calculator.to_dict(only_total=False)
r = calcul
if moduleimpl_id is None:
ans = sco_cache.AbsSemEtudCache.set(key, r)
if not ans:
log("warning: get_assiduites_count failed to cache")
nb_abs: int = r["absent"][metrique]
nb_abs_nj: int = r["absent_non_just"][metrique]
nb_abs_just: int = r["absent_just"][metrique]
return (nb_abs_nj, nb_abs_just, nb_abs)
def invalidate_assiduites_count(etudid: int, sem: dict):
"""Invalidate (clear) cached counts"""
date_debut = sem["date_debut_iso"]
date_fin = sem["date_fin_iso"]
key = str(etudid) + "_" + date_debut + "_" + date_fin + "_assiduites"
sco_cache.AbsSemEtudCache.delete(key)
# Non utilisé
def invalidate_assiduites_count_sem(sem: dict):
"""Invalidate (clear) cached abs counts for all the students of this semestre"""
inscriptions = (
sco_formsemestre_inscriptions.do_formsemestre_inscription_listinscrits(
sem["formsemestre_id"]
)
)
for ins in inscriptions:
invalidate_assiduites_count(ins["etudid"], sem)
def invalidate_assiduites_etud_date(etudid: int, the_date: datetime):
"""Doit etre appelé à chaque modification des assiduites
pour cet étudiant et cette date.
Invalide cache absence et caches semestre
"""
# Semestres a cette date:
etud = sco_etud.get_etud_info(etudid=etudid, filled=True)
if len(etud) == 0:
return
else:
etud = etud[0]
sems = [
sem
for sem in etud["sems"]
if scu.is_iso_formated(sem["date_debut_iso"], True).replace(tzinfo=UTC)
<= the_date.replace(tzinfo=UTC)
and scu.is_iso_formated(sem["date_fin_iso"], True).replace(tzinfo=UTC)
>= the_date.replace(tzinfo=UTC)
]
# Invalide les PDF et les absences:
for sem in sems:
# Inval cache bulletin et/ou note_table
# efface toujours le PDF car il affiche en général les absences
sco_cache.invalidate_formsemestre(
formsemestre_id=sem["formsemestre_id"], pdfonly=True
)
# Inval cache compteurs absences:
invalidate_assiduites_count(etudid, sem)
def simple_invalidate_cache(obj: dict, etudid: str | int = None):
"""Invalide le cache de l'étudiant et du / des semestres"""
date_debut = (
obj["date_debut"]
if isinstance(obj["date_debut"], datetime)
else scu.is_iso_formated(obj["date_debut"], True)
)
date_fin = (
obj["date_fin"]
if isinstance(obj["date_fin"], datetime)
else scu.is_iso_formated(obj["date_fin"], True)
)
etudid = etudid if etudid is not None else obj["etudid"]
invalidate_assiduites_etud_date(etudid, date_debut)
invalidate_assiduites_etud_date(etudid, date_fin)
# mettre à jour le scodoc_dept en fonction de l'étudiant
etud = Identite.query.filter_by(etudid=etudid).first_or_404()
set_sco_dept(etud.departement.acronym)
# Invalide les caches des tableaux de l'étudiant
sco_cache.RequeteTableauAssiduiteCache.delete_pattern(
pattern=f"tableau-etud-{etudid}*"
)
# Invalide les tableaux "bilan dept"
sco_cache.RequeteTableauAssiduiteCache.delete_pattern(pattern="tableau-dept*")
Reference in New Issue View Git Blame Copy Permalink