489 lines
18 KiB
Python
489 lines
18 KiB
Python
"""
|
|
Planning Service - GMAO Collège
|
|
Service de calcul de planification des interventions
|
|
"""
|
|
from datetime import datetime, timedelta, date, time
|
|
from typing import List, Dict, Optional, Tuple
|
|
from app_new.extensions import db
|
|
from app_new.core.models.planning import (
|
|
ScheduledTask, AdminTask, WorkSchedule, CollegeClosure,
|
|
ClosureSchedule, ClosureWorkDay, TechnicianAvailability, PersonalLeave,
|
|
Training, TrainingParticipant, PlanningDay, PlanningItem
|
|
)
|
|
from app_new.core.models.maintenance import Intervention
|
|
from app_new.core.models.equipment import Equipment
|
|
from app_new.core.models.college import Room
|
|
from sqlalchemy import and_, or_
|
|
|
|
|
|
def get_french_public_holidays(year: int) -> List[Tuple[date, str]]:
|
|
"""
|
|
Calcule les jours fériés français pour une année donnée.
|
|
Inclut les jours fériés fixes et variables (Pâques, Ascension, Pentecôte).
|
|
Utilise l'algorithme de Gauss pour calculer la date de Pâques.
|
|
"""
|
|
holidays = []
|
|
|
|
# Jours fériés fixes
|
|
fixed_holidays = [
|
|
(1, 1, "Jour de l'an"),
|
|
(5, 1, "Fête du travail"),
|
|
(5, 8, "Victoire 1945"),
|
|
(7, 14, "Fête nationale"),
|
|
(8, 15, "Assomption"),
|
|
(11, 1, "Toussaint"),
|
|
(11, 11, "Armistice 1918"),
|
|
(12, 25, "Noël"),
|
|
]
|
|
|
|
for month, day, name in fixed_holidays:
|
|
holidays.append((date(year, month, day), name))
|
|
|
|
# Calcul de Pâques avec l'algorithme de Butcher-Meeus
|
|
# Formule simplifiée pour l'Église catholique (grégorien)
|
|
a = year % 19
|
|
b = year // 100
|
|
c = year % 100
|
|
d = b // 4
|
|
e = b % 4
|
|
f = (b + 8) // 25
|
|
g = (b - f + 1) // 3
|
|
h = (19 * a + b - d - g + 15) % 30
|
|
i = c // 4
|
|
k = c % 4
|
|
l = (32 + 2 * e + 2 * i - h - k) % 7
|
|
m = (a + 11 * h + 22 * l) // 451
|
|
month_easter = (h + l - 7 * m + 114) // 31
|
|
day_easter = ((h + l - 7 * m + 114) % 31) + 1
|
|
|
|
easter_sunday = date(year, month_easter, day_easter)
|
|
|
|
# Lundi de Pâques (lendemain de Pâques)
|
|
holidays.append((easter_sunday + timedelta(days=1), "Lundi de Pâques"))
|
|
|
|
# Ascension (39 jours après Pâques)
|
|
holidays.append((easter_sunday + timedelta(days=39), "Ascension"))
|
|
|
|
# Lundi de Pentecôte (50 jours après Pâques)
|
|
holidays.append((easter_sunday + timedelta(days=50), "Lundi de Pentecôte"))
|
|
|
|
return holidays
|
|
|
|
|
|
def is_public_holiday(d: date) -> Tuple[bool, str]:
|
|
"""
|
|
Vérifie si une date est un jour férié français.
|
|
Retourne (True, nom du jour férié) ou (False, '').
|
|
"""
|
|
holidays = get_french_public_holidays(d.year)
|
|
for h_date, h_name in holidays:
|
|
if h_date == d:
|
|
return True, h_name
|
|
return False, ''
|
|
|
|
|
|
class DaySchedule:
|
|
"""Structure pour représenter le planning d'une journée."""
|
|
def __init__(self, date_obj: date):
|
|
self.date = date_obj
|
|
self.is_working = False
|
|
self.day_name = ''
|
|
self.working_hours = None
|
|
self.closure = None
|
|
self.leave = None
|
|
self.holiday = None # Nom du jour férié si applicable
|
|
self.items = []
|
|
self.grouped_items = []
|
|
|
|
|
|
class PlanningService:
|
|
"""Service de planification des interventions."""
|
|
|
|
# Priorités
|
|
PRIORITY_ADMIN = 1
|
|
PRIORITY_CURATIVE_HIGH = 2
|
|
PRIORITY_CURATIVE_NORMAL = 3
|
|
PRIORITY_PREVENTIVE_PRIORITARY = 2
|
|
PRIORITY_PREVENTIVE_NORMAL = 4
|
|
PRIORITY_FORMATION = 1
|
|
|
|
@staticmethod
|
|
def maintenance_unavailability(d: date, user_id=None):
|
|
"""Retourne le motif qui interdit toute maintenance ce jour-là."""
|
|
leave_query = PersonalLeave.query.filter(
|
|
PersonalLeave.start_date <= d,
|
|
PersonalLeave.end_date >= d,
|
|
)
|
|
if user_id is not None:
|
|
leave_query = leave_query.filter(PersonalLeave.user_id == user_id)
|
|
leave = leave_query.first()
|
|
if leave:
|
|
return f"Absence : {leave.leave_type or 'indisponible'}"
|
|
|
|
training_query = Training.query.join(TrainingParticipant).filter(
|
|
Training.start_date <= d,
|
|
Training.end_date >= d,
|
|
TrainingParticipant.is_confirmed.is_(True),
|
|
)
|
|
if user_id is not None:
|
|
training_query = training_query.filter(TrainingParticipant.user_id == user_id)
|
|
training = training_query.first()
|
|
if training:
|
|
return f"Formation : {training.name}"
|
|
return None
|
|
|
|
@staticmethod
|
|
def get_working_hours(d: date, user_id=None) -> Optional[Tuple[time, time, Optional[time], Optional[time]]]:
|
|
"""
|
|
Retourne les horaires de travail pour une date donnée.
|
|
Retourne (start, end, lunch_start, lunch_end) ou None si non travaillé.
|
|
"""
|
|
# Vérifier si c'est un jour férié français
|
|
is_holiday, holiday_name = is_public_holiday(d)
|
|
if is_holiday:
|
|
return None
|
|
|
|
if PlanningService.maintenance_unavailability(d, user_id):
|
|
return None
|
|
|
|
# Vérifier si c'est un jour de vacances
|
|
closure = CollegeClosure.query.filter(
|
|
CollegeClosure.start_date <= d,
|
|
CollegeClosure.end_date >= d
|
|
).first()
|
|
|
|
day_of_week = d.weekday() # 0=lundi, 6=dimanche
|
|
|
|
if closure:
|
|
# Une fermeture scolaire est non travaillée par défaut. Une date
|
|
# exacte peut toutefois être rouverte avec ses propres horaires.
|
|
exceptional_day = ClosureWorkDay.query.filter_by(
|
|
closure_id=closure.id,
|
|
work_date=d,
|
|
).first()
|
|
if exceptional_day:
|
|
return (
|
|
exceptional_day.start_time,
|
|
exceptional_day.end_time,
|
|
exceptional_day.lunch_start,
|
|
exceptional_day.lunch_end,
|
|
)
|
|
|
|
if closure.work_hours_type == 'none':
|
|
return None
|
|
|
|
if closure.work_hours_type == 'reduced':
|
|
return (time(9), time(12), None, None)
|
|
|
|
if closure.work_hours_type == 'custom':
|
|
schedule = ClosureSchedule.query.filter_by(
|
|
closure_id=closure.id,
|
|
day_of_week=day_of_week
|
|
).first()
|
|
if schedule:
|
|
return (schedule.start_time, schedule.end_time,
|
|
schedule.lunch_start, schedule.lunch_end)
|
|
return None
|
|
|
|
# Vérifier les congés personnels
|
|
leave = TechnicianAvailability.query.filter_by(date=d).first()
|
|
if leave and leave.availability_type in ('conge', 'maladie'):
|
|
return None
|
|
|
|
# Horaires normaux - chercher le premier schedule avec des horaires valides
|
|
schedules = WorkSchedule.query.filter_by(day_of_week=day_of_week).all()
|
|
for schedule in schedules:
|
|
if schedule.start_time and schedule.end_time:
|
|
return (schedule.start_time, schedule.end_time,
|
|
schedule.lunch_start, schedule.lunch_end)
|
|
|
|
# Si un schedule existe mais sans horaires, considérer comme non travaillé
|
|
if schedules:
|
|
return None
|
|
|
|
# Week-end non travaillé par défaut
|
|
if day_of_week >= 5:
|
|
return None
|
|
|
|
# Horaires par défaut (8h-17h avec pause 12h-13h)
|
|
return (time(8), time(17), time(12), time(13))
|
|
|
|
@staticmethod
|
|
def get_day_schedule(d: date) -> DaySchedule:
|
|
"""Retourne le planning complet pour une journée."""
|
|
days_fr = ['Lundi', 'Mardi', 'Mercredi', 'Jeudi', 'Vendredi', 'Samedi', 'Dimanche']
|
|
|
|
schedule = DaySchedule(d)
|
|
schedule.day_name = days_fr[d.weekday()]
|
|
|
|
# Vérifier les vacances
|
|
closure = CollegeClosure.query.filter(
|
|
CollegeClosure.start_date <= d,
|
|
CollegeClosure.end_date >= d
|
|
).first()
|
|
|
|
# Vérifier les congés personnels
|
|
leave = TechnicianAvailability.query.filter_by(date=d).first()
|
|
|
|
# Horaires de travail
|
|
hours = PlanningService.get_working_hours(d)
|
|
|
|
if hours:
|
|
schedule.is_working = True
|
|
schedule.working_hours = {
|
|
'start': hours[0],
|
|
'end': hours[1],
|
|
'lunch_start': hours[2],
|
|
'lunch_end': hours[3]
|
|
}
|
|
else:
|
|
schedule.is_working = False
|
|
if closure:
|
|
schedule.closure = closure.name
|
|
elif leave:
|
|
schedule.leave = leave.availability_type
|
|
else:
|
|
# Vérifier si c'est un jour férié
|
|
is_holiday, holiday_name = is_public_holiday(d)
|
|
if is_holiday:
|
|
schedule.holiday = holiday_name
|
|
# Ne pas ajouter de tâches si ce n'est pas un jour de travail
|
|
return schedule
|
|
|
|
# Récupérer les interventions planifiées pour ce jour
|
|
scheduled_tasks = ScheduledTask.query.filter(
|
|
ScheduledTask.scheduled_date == d,
|
|
ScheduledTask.status.in_(('planned', 'in_progress', 'suspended', 'postponed')),
|
|
ScheduledTask.intervention_id.is_(None),
|
|
).all()
|
|
|
|
for task in scheduled_tasks:
|
|
# Récupérer le titre depuis la tâche préventive associée
|
|
task_title = 'Tâche planifiée'
|
|
if task.preventive_task:
|
|
task_title = task.preventive_task.name
|
|
elif task.equipment:
|
|
task_title = f"Intervention sur {task.equipment.name}"
|
|
|
|
item = {
|
|
'id': task.id,
|
|
'type': 'preventive',
|
|
'title': task_title,
|
|
'start_time': task.scheduled_start,
|
|
'end_time': task.scheduled_end,
|
|
'duration': task.estimated_duration,
|
|
'room_id': task.room_id,
|
|
'room_name': task.room.name if task.room else (task.equipment.effective_room.name if task.equipment and task.equipment.effective_room else None),
|
|
'status': task.status
|
|
}
|
|
schedule.items.append(item)
|
|
|
|
# Les interventions préventives et curatives appartiennent au même
|
|
# planning. Les échéances déjà converties sont représentées ici par
|
|
# leur intervention, sans doublon avec ScheduledTask.
|
|
interventions = Intervention.query.filter(
|
|
Intervention.scheduled_date == d,
|
|
Intervention.is_deleted.is_(False),
|
|
~Intervention.status.in_(('terminee', 'cloturee', 'annulee', 'ignoree')),
|
|
).all()
|
|
|
|
for interv in interventions:
|
|
item = {
|
|
'id': interv.id,
|
|
'type': 'curative' if interv.type == 'curatif' else 'preventive',
|
|
'title': interv.title,
|
|
'start_time': None,
|
|
'end_time': None,
|
|
'duration': None,
|
|
'room_id': None,
|
|
'room_name': interv.room.name if interv.room else None,
|
|
'status': interv.status,
|
|
}
|
|
schedule.items.append(item)
|
|
|
|
# Récupérer les tâches administratives
|
|
# AdminTask est récurrent (frequency, day_of_week) - on filtre par jour de la semaine
|
|
admin_tasks = AdminTask.query.filter_by(
|
|
is_active=True
|
|
).all()
|
|
|
|
for admin in admin_tasks:
|
|
# Vérifier si la tâche correspond au jour actuel
|
|
if admin.frequency == 'daily':
|
|
pass # Tous les jours
|
|
elif admin.frequency == 'weekly' and admin.day_of_week != d.weekday():
|
|
continue
|
|
elif admin.frequency == 'monthly' and admin.day_of_month != d.day:
|
|
continue
|
|
|
|
item = {
|
|
'id': admin.id,
|
|
'type': 'admin',
|
|
'title': admin.name,
|
|
'start_time': admin.start_time,
|
|
'end_time': None, # Calculé à partir de start_time + duration
|
|
'duration': admin.duration_minutes,
|
|
'room_id': None,
|
|
'room_name': None,
|
|
'status': 'planned'
|
|
}
|
|
schedule.items.append(item)
|
|
|
|
# Grouper par salle
|
|
rooms = {}
|
|
for item in schedule.items:
|
|
room_name = item.get('room_name') or 'Sans salle'
|
|
if room_name not in rooms:
|
|
rooms[room_name] = []
|
|
rooms[room_name].append(item)
|
|
|
|
schedule.grouped_items = [
|
|
{'room_name': room_name, 'task_items': items}
|
|
for room_name, items in sorted(rooms.items())
|
|
]
|
|
|
|
return schedule
|
|
|
|
@staticmethod
|
|
def is_room_available(room_id: int, start_time: datetime, end_time: datetime) -> bool:
|
|
"""
|
|
Vérifie si une salle est disponible pendant une période donnée.
|
|
Utilise le planning PRONOTE importé.
|
|
"""
|
|
from app_new.core.models.college import PronoteSchedule
|
|
|
|
courses = PronoteSchedule.query.filter(
|
|
PronoteSchedule.room_id == room_id,
|
|
PronoteSchedule.start_time < end_time,
|
|
PronoteSchedule.end_time > start_time
|
|
).all()
|
|
|
|
return len(courses) == 0
|
|
|
|
@staticmethod
|
|
def get_room_occupancy(room_id: int, d: date) -> List[Tuple[time, time]]:
|
|
"""Retourne les périodes occupées d'une salle pour une date."""
|
|
from app_new.core.models.college import PronoteSchedule
|
|
|
|
start_dt = datetime.combine(d, time(0))
|
|
end_dt = datetime.combine(d, time(23, 59))
|
|
|
|
courses = PronoteSchedule.query.filter(
|
|
PronoteSchedule.room_id == room_id,
|
|
PronoteSchedule.start_time >= start_dt,
|
|
PronoteSchedule.start_time < end_dt
|
|
).order_by(PronoteSchedule.start_time).all()
|
|
|
|
return [(c.start_time.time(), c.end_time.time()) for c in courses]
|
|
|
|
@staticmethod
|
|
def calculate_planning(start_date: date, end_date: date) -> List[PlanningDay]:
|
|
"""
|
|
Calcule le planning sur une période donnée.
|
|
Prend en compte les priorités, les disponibilités et les regroupements par salle.
|
|
"""
|
|
from app_new.core.models.equipment import EquipmentRestriction
|
|
|
|
planning_days = []
|
|
current_date = start_date
|
|
|
|
while current_date <= end_date:
|
|
# Vérifier si le jour est travaillé
|
|
if PlanningService.get_working_hours(current_date):
|
|
day_schedule = PlanningService.get_day_schedule(current_date)
|
|
planning_days.append(day_schedule)
|
|
|
|
current_date += timedelta(days=1)
|
|
|
|
return planning_days
|
|
|
|
@staticmethod
|
|
def schedule_task(task: ScheduledTask, preferred_date: Optional[date] = None) -> bool:
|
|
"""
|
|
Planifie une tâche à une date optimale.
|
|
Tient compte des disponibilités, restrictions et regroupements.
|
|
"""
|
|
from app_new.core.models.equipment import EquipmentRestriction
|
|
|
|
if preferred_date:
|
|
task.scheduled_date = preferred_date
|
|
task.status = 'planned'
|
|
db.session.commit()
|
|
return True
|
|
|
|
# Trouver la prochaine date disponible
|
|
current_date = date.today()
|
|
max_days = 365 # Maximum 1 an
|
|
|
|
for _ in range(max_days):
|
|
current_date += timedelta(days=1)
|
|
|
|
# Vérifier si c'est un jour travaillé
|
|
hours = PlanningService.get_working_hours(current_date)
|
|
if not hours:
|
|
continue
|
|
|
|
# Vérifier les restrictions de l'équipement
|
|
if task.equipment:
|
|
restrictions = EquipmentRestriction.query.filter_by(
|
|
equipment_id=task.equipment.id
|
|
).all()
|
|
|
|
valid = True
|
|
for restriction in restrictions:
|
|
if restriction.restriction_type == 'time':
|
|
# Vérifier si le jour de la semaine correspond
|
|
day_names = ['monday', 'tuesday', 'wednesday', 'thursday', 'friday', 'saturday', 'sunday']
|
|
if restriction.days_of_week:
|
|
allowed_days = restriction.days_of_week.split(',')
|
|
if day_names[current_date.weekday()] not in allowed_days:
|
|
valid = False
|
|
break
|
|
|
|
elif restriction.restriction_type == 'room_occupancy':
|
|
# Vérifier si la salle est libre
|
|
room_id = restriction.room_id
|
|
if room_id and not PlanningService.is_room_available(room_id,
|
|
datetime.combine(current_date, time(8, 0)),
|
|
datetime.combine(current_date, time(17, 0))):
|
|
valid = False
|
|
break
|
|
|
|
if not valid:
|
|
continue
|
|
|
|
# Vérifier la disponibilité de la salle
|
|
if task.room_id:
|
|
# TODO: Intégrer avec PRONOTE
|
|
pass
|
|
|
|
# Planifier la tâche
|
|
task.scheduled_date = current_date
|
|
task.status = 'planned'
|
|
db.session.commit()
|
|
return True
|
|
|
|
return False
|
|
|
|
@staticmethod
|
|
def postpone_task(task_id: int, new_date: Optional[date] = None, auto: bool = False) -> bool:
|
|
"""
|
|
Reporte une tâche à une date ultérieure.
|
|
Si auto=True, reporte automatiquement au prochain jour disponible.
|
|
"""
|
|
task = ScheduledTask.query.get(task_id)
|
|
if not task:
|
|
return False
|
|
|
|
if auto:
|
|
# Trouver le prochain jour disponible
|
|
current_date = task.scheduled_date or date.today()
|
|
current_date += timedelta(days=1)
|
|
return PlanningService.schedule_task(task, preferred_date=None)
|
|
else:
|
|
if new_date:
|
|
task.scheduled_date = new_date
|
|
db.session.commit()
|
|
return True
|
|
|
|
return False
|