""" Planificateur automatique - GMAO College Genere les taches planifiees en croisant : - Taches preventives dues (PreventiveTask + LotTask) - Horaires de travail (WorkSchedule) - Vacances (CollegeClosure) + absences (PersonalLeave) + formations (Training) - Contraintes de salle (RoomConstraint) - Reserve 40% de creneaux pour le curatif - Regroupe par batiment/zone pour optimiser les deplacements """ from datetime import datetime, timezone, date, time, timedelta from ..extensions import db from ..core.models.planning import ( WorkSchedule, CollegeClosure, ClosureWorkDay, PersonalLeave, Training, PreventiveTask, PlanningDay, PlanningItem ) from ..core.models.maintenance import Intervention, LotTask, Lot from ..core.models.equipment import Equipment, RoomConstraint, is_room_available from ..core.models.college import Room, Building from sqlalchemy import func def get_working_hours(check_date, user_id=1): """Retourne les horaires de travail pour une date donnee. Prend en compte les vacances (CollegeClosure) et les horaires normaux (WorkSchedule). Retourne (start_time, end_time, lunch_start, lunch_end) ou None si non travaille. """ from ..core.services.planning_service import PlanningService return PlanningService.get_working_hours(check_date, user_id=user_id) def is_user_available(check_date, user_id=1): """Verifie si l'utilisateur est disponible (pas en conge, pas en formation).""" from ..core.services.planning_service import PlanningService reason = PlanningService.maintenance_unavailability(check_date, user_id=user_id) return (reason is None, reason) def get_due_tasks(weeks_ahead=4): """Recupere toutes les taches preventives dues dans les N prochaines semaines. Retourne une liste de dicts avec : task, equipment, room, due_date, duration, lot_task """ today = date.today() horizon = today + timedelta(weeks=weeks_ahead) due_tasks = [] # 1. PreventiveTask avec period_days for pt in PreventiveTask.query.filter_by(is_active=True).all(): if not pt.period_days: continue # Trouver les equipements correspondants if pt.lot_id: equipments = Equipment.query.filter_by(lot_id=pt.lot_id).all() elif pt.category_id: equipments = Equipment.query.filter_by(category_id=pt.category_id).all() else: continue for eq in equipments: # Verifier si deja planifie existing = Intervention.query.filter_by( preventive_task_id=pt.id, equipment_id=eq.id, status='planifiee', is_deleted=False ).first() if existing: continue # Calculer la date due last_done = Intervention.query.filter_by( preventive_task_id=pt.id, equipment_id=eq.id, status='terminee', is_deleted=False ).order_by(Intervention.scheduled_date.desc()).first() if last_done: due_date = last_done.scheduled_date + timedelta(days=pt.period_days) else: due_date = today # Jamais fait -> dues maintenant if due_date <= horizon: due_tasks.append({ 'preventive_task': pt, 'equipment': eq, 'room': eq.room, 'due_date': due_date, 'duration': pt.duration_minutes or 60, 'lot_task': None, }) # 2. LotTask avec jours_entre_interventions for lt in LotTask.query.filter_by(is_active=True, trigger_type='calendar').all(): if not lt.jours_entre_interventions: continue lot = Lot.query.get(lt.lot_id) if not lot: continue equipments = Equipment.query.filter_by(lot_id=lot.id).all() for eq in equipments: existing = Intervention.query.filter_by( lot_task_id=lt.id, equipment_id=eq.id, status='planifiee', is_deleted=False ).first() if existing: continue last_done = Intervention.query.filter_by( lot_task_id=lt.id, equipment_id=eq.id, status='terminee', is_deleted=False ).order_by(Intervention.scheduled_date.desc()).first() if last_done: due_date = last_done.scheduled_date + timedelta(days=lt.jours_entre_interventions) else: due_date = today if due_date <= horizon: due_tasks.append({ 'lot_task': lt, 'equipment': eq, 'room': eq.room, 'due_date': due_date, 'duration': lt.effective_duration(), 'preventive_task': None, }) # Trier par date due (les plus en retard d'abord) due_tasks.sort(key=lambda x: x['due_date']) return due_tasks def group_by_zone(tasks): """Regroupe les taches par batiment/zone pour optimiser les deplacements. Retourne un dict {building_name: [tasks]} """ groups = {} for t in tasks: room = t.get('room') building_name = 'Autre' if room and hasattr(room, 'building') and room.building: building_name = room.building.name elif room: building_name = f"Salle {room.name}" else: building_name = 'Non localise' if building_name not in groups: groups[building_name] = [] groups[building_name].append(t) return groups def find_slot(check_date, duration_min, room_id, user_id=1, reserved_ratio=0.4): """Trouve un creneau disponible pour une tache. Verifie : horaires de travail, conges, contraintes de salle. Reserve reserved_ratio (40%) du temps pour le curatif. Retourne (start_time, end_time) ou None. """ wh = get_working_hours(check_date, user_id) if not wh: return None work_start, work_end, lunch_start, lunch_end = wh available_minutes = 0 if lunch_start and lunch_end: # Matin + apres-midi morning = (lunch_start.hour * 60 + lunch_start.minute) - (work_start.hour * 60 + work_start.minute) afternoon = (work_end.hour * 60 + work_end.minute) - (lunch_end.hour * 60 + lunch_end.minute) available_minutes = morning + afternoon else: available_minutes = (work_end.hour * 60 + work_end.minute) - (work_start.hour * 60 + work_start.minute) # Reserver 40% pour le curatif bookable_minutes = int(available_minutes * (1 - reserved_ratio)) # Verifier combien de temps est deja pris ce jour-la booked = Intervention.query.filter_by( scheduled_date=check_date, status='planifiee', is_deleted=False ).all() booked_minutes = sum(t.estimated_duration or 60 for t in booked) remaining = bookable_minutes - booked_minutes if remaining < duration_min: return None # Pas assez de place # Trouver un creneau libre # Commencer apres le dernier creneau booke if booked: last_end = max( (t.scheduled_end for t in booked if t.scheduled_end), default=work_start ) # Si le dernier se termine apres le debut, commencer apres slot_start = last_end else: slot_start = work_start # Ajuster pour la pause dejeuner if lunch_start and lunch_end and slot_start < lunch_end: if slot_start < lunch_start: # On peut commencer avant la pause, verifier la duree morning_end = lunch_start morning_available = (morning_end.hour * 60 + morning_end.minute) - (slot_start.hour * 60 + slot_start.minute) if morning_available >= duration_min: slot_end_dt = datetime.combine(check_date, slot_start) + timedelta(minutes=duration_min) slot_end = slot_end_dt.time() # Verifier contrainte de salle if room_id: available, reason = is_room_available(room_id, check_date, slot_start, slot_end) if not available: return None return (slot_start, slot_end) # Commencer apres la pause slot_start = lunch_end slot_end_dt = datetime.combine(check_date, slot_start) + timedelta(minutes=duration_min) slot_end = slot_end_dt.time() if slot_end > work_end: return None # Depasse la fin du travail # Verifier contrainte de salle if room_id: available, reason = is_room_available(room_id, check_date, slot_start, slot_end) if not available: return None return (slot_start, slot_end) def run_scheduler(weeks_ahead=4, user_id=1): """Lance le planificateur. - Recupere les taches dues - Les regroupe par zone - Pour chaque jour ouvrable, essaie de placer les taches - Reserve 40% pour le curatif - Retourne un resume """ due_tasks = get_due_tasks(weeks_ahead) if not due_tasks: return {'success': True, 'planned': 0, 'skipped': 0, 'errors': [], 'message': 'Aucune tache due'} # Regrouper par zone groups = group_by_zone(due_tasks) planned = 0 skipped = 0 errors = [] # Parcourir les jours a partir d'aujourd'hui today = date.today() remaining_tasks = list(due_tasks) for day_offset in range(weeks_ahead * 7): check_date = today + timedelta(days=day_offset) # Verifier disponibilite utilisateur avail, reason = is_user_available(check_date, user_id) if not avail: continue # Verifier horaires wh = get_working_hours(check_date, user_id) if not wh: continue # Trier les taches restantes par date due (urgentes d'abord) remaining_tasks.sort(key=lambda x: x['due_date']) # Pour chaque tache, essayer de trouver un creneau tasks_to_remove = [] for t in remaining_tasks: slot = find_slot(check_date, t['duration'], t['room'].id if t['room'] else None, user_id) if slot: start_time, end_time = slot # Creer l'Intervention au statut planifiee title = '' if t['preventive_task']: title = f"[{check_date}] {t['preventive_task'].name}"[:290] elif t['lot_task']: title = f"[{check_date}] {t['lot_task'].tache or 'Tache lot'}"[:290] else: title = f"Tache {t['equipment'].name} du {check_date}"[:290] interv = Intervention( title=title, type='preventif', status='planifiee', priority='normale', equipment_id=t['equipment'].id, room_id=t['room'].id if t['room'] else None, scheduled_date=check_date, scheduled_start=start_time, scheduled_end=end_time, estimated_duration=float(t['duration']), preventive_task_id=t['preventive_task'].id if t['preventive_task'] else None, lot_task_id=t['lot_task'].id if t['lot_task'] else None, ) db.session.add(interv) planned += 1 tasks_to_remove.append(t) for t in tasks_to_remove: remaining_tasks.remove(t) if not remaining_tasks: break skipped = len(remaining_tasks) if remaining_tasks: for t in remaining_tasks[:5]: errors.append(f"Non planifie: {t['equipment'].name} (du {t['due_date'].strftime('%d/%m')})") db.session.commit() return { 'success': True, 'planned': planned, 'skipped': skipped, 'total_due': len(due_tasks), 'errors': errors, 'message': f"{planned} tache(s) planifiee(s), {skipped} non planifiee(s) sur {len(due_tasks)} dues" }