210 lines
8.3 KiB
Python
210 lines
8.3 KiB
Python
"""Règles métier du cycle de vie des équipements et de leurs échéances."""
|
||
from datetime import date, datetime, timedelta, timezone
|
||
|
||
from app_new.extensions import db
|
||
from app_new.core.models.equipment import Equipment, EquipmentLifecycleEvent
|
||
from app_new.core.models.maintenance import Intervention
|
||
from app_new.core.models.planning import ScheduledTask
|
||
|
||
|
||
OPEN_TASK_STATUSES = ("planned", "in_progress", "suspended")
|
||
|
||
|
||
def record_event(equipment, event_type, user_id=None, previous_status=None,
|
||
new_status=None, quantity=1, reason=None, intervention_id=None):
|
||
"""Enregistre un événement avec la localisation et le lot tels qu'ils sont aujourd'hui."""
|
||
room = equipment.effective_room
|
||
lot = equipment.effective_lot
|
||
event = EquipmentLifecycleEvent(
|
||
equipment=equipment,
|
||
event_type=event_type,
|
||
previous_status=previous_status,
|
||
new_status=new_status,
|
||
quantity=max(quantity or 1, 1),
|
||
reason=reason,
|
||
room_id_snapshot=room.id if room else None,
|
||
room_name_snapshot=room.name if room else None,
|
||
lot_id_snapshot=lot.id if lot else None,
|
||
lot_name_snapshot=lot.name if lot else None,
|
||
intervention_id=intervention_id,
|
||
created_by_id=user_id,
|
||
)
|
||
db.session.add(event)
|
||
return event
|
||
|
||
|
||
def extract_units(group, quantity, user_id=None, reason=None):
|
||
"""Individualise des unités d'un groupe sans modifier son total physique."""
|
||
if not group.is_group:
|
||
if quantity != 1:
|
||
raise ValueError("Un équipement individuel représente une seule unité.")
|
||
return [group]
|
||
quantity = max(int(quantity or 0), 0)
|
||
if quantity < 1 or quantity > group.unindividualized_quantity:
|
||
raise ValueError(
|
||
f"Quantité invalide : {group.unindividualized_quantity} unité(s) seulement sont disponibles."
|
||
)
|
||
root = group
|
||
while root.parent_id:
|
||
root = root.parent
|
||
next_number = max([item.individual_number or 0 for item in root.all_children] or [0]) + 1
|
||
created = []
|
||
room = group.effective_room
|
||
for offset in range(quantity):
|
||
unit = Equipment(
|
||
name=f"{root.name} #{next_number + offset}",
|
||
description=root.description,
|
||
status="en_service",
|
||
lifecycle_status="en_service",
|
||
parent_id=group.id,
|
||
room_id=room.id if room else None,
|
||
is_group=False,
|
||
quantity=1,
|
||
tracked_individually=True,
|
||
individual_number=next_number + offset,
|
||
mobility=root.mobility,
|
||
)
|
||
db.session.add(unit)
|
||
db.session.flush()
|
||
record_event(unit, "individualisation", user_id, new_status="en_service", reason=reason)
|
||
created.append(unit)
|
||
return created
|
||
|
||
|
||
def _open_tasks(equipment):
|
||
return ScheduledTask.query.filter(
|
||
ScheduledTask.equipment_id == equipment.id,
|
||
ScheduledTask.status.in_(OPEN_TASK_STATUSES),
|
||
).all()
|
||
|
||
|
||
def _suspend_preventive(equipment, reason):
|
||
now = datetime.now(timezone.utc)
|
||
for task in _open_tasks(equipment):
|
||
rule = task.lot_task
|
||
if rule and (rule.is_safety_critical or rule.unavailable_policy == "keep"):
|
||
continue
|
||
task.status = "suspended"
|
||
task.suspension_reason = reason
|
||
task.suspended_at = now
|
||
|
||
|
||
def _cancel_preventive(equipment, reason, keep_safety=False):
|
||
for task in _open_tasks(equipment):
|
||
if keep_safety and task.lot_task and task.lot_task.is_safety_critical:
|
||
continue
|
||
task.status = "cancelled"
|
||
task.postpone_reason = reason
|
||
task.suspension_reason = None
|
||
|
||
|
||
def _resume_preventive(equipment):
|
||
"""Réactive les échéances suspendues en appliquant la règle du lot."""
|
||
from app_new.core.services.maintenance_engine import next_working_day
|
||
|
||
today = date.today()
|
||
for task in _open_tasks(equipment):
|
||
if task.status != "suspended":
|
||
continue
|
||
rule = task.lot_task
|
||
if rule and rule.after_repair_policy == "reset" and rule.jours_entre_interventions:
|
||
target = today + timedelta(days=rule.jours_entre_interventions)
|
||
else:
|
||
target = max(task.scheduled_date or today, today)
|
||
task.scheduled_date = next_working_day(target)
|
||
task.status = "planned"
|
||
task.suspension_reason = None
|
||
task.suspended_at = None
|
||
task.lifecycle_status_snapshot = "en_service"
|
||
|
||
|
||
def create_corrective_intervention(source, units, reason, user_id=None):
|
||
room = source.effective_room
|
||
lot = source.effective_lot
|
||
quantity = len(units)
|
||
intervention = Intervention(
|
||
title=f"Réparation — {quantity} × {source.name}"[:300],
|
||
description=reason or "Équipement retiré provisoirement pour diagnostic et réparation.",
|
||
type="curatif",
|
||
status="en_attente",
|
||
priority="normale",
|
||
equipment_id=source.id,
|
||
room_id=room.id if room else None,
|
||
lot_id=lot.id if lot else None,
|
||
quantity_affected=quantity,
|
||
individual_numbers=",".join(str(unit.individual_number) for unit in units if unit.individual_number),
|
||
# L'intervention reste non affectée jusqu'à sa prise en charge. L'auteur
|
||
# de l'action est conservé dans le journal de cycle de vie.
|
||
)
|
||
db.session.add(intervention)
|
||
db.session.flush()
|
||
return intervention
|
||
|
||
|
||
def start_repair(source, quantity=1, reason=None, user_id=None, create_intervention=True):
|
||
units = extract_units(source, quantity, user_id=user_id, reason=reason) if source.is_group else [source]
|
||
intervention = create_corrective_intervention(source, units, reason, user_id) if create_intervention else None
|
||
now = datetime.now(timezone.utc)
|
||
for unit in units:
|
||
previous = unit.lifecycle_status
|
||
unit.lifecycle_status = "en_reparation"
|
||
unit.status = "en_maintenance"
|
||
unit.unavailable_since = now
|
||
_suspend_preventive(unit, "Équipement indisponible pendant sa réparation")
|
||
record_event(unit, "retrait_reparation", user_id, previous, unit.lifecycle_status,
|
||
reason=reason, intervention_id=intervention.id if intervention else None)
|
||
return units, intervention
|
||
|
||
|
||
def complete_repair(equipment, reason=None, user_id=None):
|
||
previous = equipment.lifecycle_status
|
||
equipment.lifecycle_status = "en_service"
|
||
equipment.status = "en_service"
|
||
equipment.unavailable_since = None
|
||
equipment.create_scheduled_tasks_from_lot(commit=False)
|
||
_resume_preventive(equipment)
|
||
record_event(equipment, "reparation_terminee", user_id, previous, "en_service", reason=reason)
|
||
|
||
|
||
def propose_disposal(source, quantity=1, reason=None, planned_date=None, user_id=None):
|
||
units = extract_units(source, quantity, user_id=user_id, reason=reason) if source.is_group else [source]
|
||
for unit in units:
|
||
unit.create_scheduled_tasks_from_lot(commit=False)
|
||
previous = unit.lifecycle_status
|
||
unit.status = "a_jeter"
|
||
unit.lifecycle_status = "mise_au_rebut_planifiee"
|
||
unit.planned_disposal_date = planned_date
|
||
unit.disposal_reason = reason
|
||
_cancel_preventive(unit, "Mise au rebut proposée", keep_safety=True)
|
||
record_event(unit, "rebut_propose", user_id, previous, unit.lifecycle_status, reason=reason)
|
||
return units
|
||
|
||
|
||
def cancel_disposal(equipment, reason=None, user_id=None):
|
||
previous = equipment.lifecycle_status
|
||
equipment.status = "en_service"
|
||
equipment.lifecycle_status = "en_service"
|
||
equipment.planned_disposal_date = None
|
||
equipment.disposal_reason = None
|
||
_resume_preventive(equipment)
|
||
record_event(equipment, "rebut_annule", user_id, previous, "en_service", reason=reason)
|
||
|
||
|
||
def confirm_disposal(equipment, reason=None, user_id=None):
|
||
previous = equipment.lifecycle_status
|
||
equipment.status = "jete"
|
||
equipment.lifecycle_status = "jete"
|
||
equipment.disposal_date = date.today()
|
||
equipment.disposal_reason = reason or equipment.disposal_reason
|
||
_cancel_preventive(equipment, "Équipement sorti définitivement du patrimoine")
|
||
record_event(equipment, "rebut_confirme", user_id, previous, "jete", reason=equipment.disposal_reason)
|
||
equipment.room_id = None
|
||
|
||
|
||
def update_future_task_location(equipment):
|
||
room = equipment.effective_room
|
||
for task in _open_tasks(equipment):
|
||
if task.intervention_id:
|
||
continue
|
||
task.room_id = room.id if room else None
|
||
task.room_name_snapshot = room.name if room else None
|