228 lines
11 KiB
Python
228 lines
11 KiB
Python
import logging
|
|
from datetime import timedelta
|
|
from django.utils import timezone
|
|
from django.contrib.contenttypes.models import ContentType
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def _resolve_inspection_target(intervention, sub_type):
|
|
"""
|
|
Détermine l'équipement (Asset), la localisation ou l'intervention cible pour l'inspection.
|
|
1. Si un équipement valide est déjà attaché à l'intervention (interventionasset_set), on le conserve
|
|
(en vérifiant s'il correspond au modèle/catégories configurés si applicables).
|
|
2. Sinon, on identifie les modèles d'assets candidats :
|
|
- sub_type.target_asset_model si configuré
|
|
- allowed_models des sub_type.concerned_categories
|
|
- Fallback : Roadway si thématique roads ou sous-type de type Hot Box / voirie
|
|
3. On sélectionne l'équipement du modèle candidat le plus proche spatialement
|
|
du centroïde de l'intervention (en priorité sur la rue associée).
|
|
4. Fallback sur la première localisation ou l'intervention elle-même.
|
|
"""
|
|
from django.contrib.gis.db import models as gis_models
|
|
from django.contrib.gis.geos import Point
|
|
from django.contrib.gis.db.models.functions import Distance
|
|
|
|
# 1. Vérifier si un équipement correspondant est déjà rattaché à l'intervention
|
|
if hasattr(intervention, 'interventionasset_set') and intervention.interventionasset_set.exists():
|
|
if sub_type and sub_type.target_asset_model_id:
|
|
matching = intervention.interventionasset_set.filter(content_type=sub_type.target_asset_model).first()
|
|
if matching and matching.content_type_id and matching.object_id:
|
|
return matching.content_type, matching.object_id
|
|
|
|
if sub_type and sub_type.concerned_categories.exists():
|
|
allowed_cts = ContentType.objects.filter(asset_categories__in=sub_type.concerned_categories.all()).distinct()
|
|
if allowed_cts.exists():
|
|
matching = intervention.interventionasset_set.filter(content_type__in=allowed_cts).first()
|
|
if matching and matching.content_type_id and matching.object_id:
|
|
return matching.content_type, matching.object_id
|
|
|
|
first_asset = intervention.interventionasset_set.first()
|
|
if first_asset and first_asset.content_type_id and first_asset.object_id:
|
|
return first_asset.content_type, first_asset.object_id
|
|
|
|
# 2. Identifier les modèles d'assets cibles candidats
|
|
candidate_cts = []
|
|
if sub_type:
|
|
if sub_type.target_asset_model_id:
|
|
candidate_cts.append(sub_type.target_asset_model)
|
|
elif sub_type.concerned_categories.exists():
|
|
allowed_cts = list(ContentType.objects.filter(asset_categories__in=sub_type.concerned_categories.all()).distinct())
|
|
candidate_cts.extend(allowed_cts)
|
|
|
|
# Fallback si aucun modèle configuré : pour les voiries ou Hot Box, cibler Roadway
|
|
if not candidate_cts:
|
|
thematic_code = getattr(intervention.thematic, 'code', None) if intervention.thematic else None
|
|
is_road_repair = (thematic_code == 'roads') or (
|
|
sub_type and any(k in (sub_type.code or '').lower() or k in (sub_type.name_fr or '').lower() for k in ('hotbox', 'hot_box', 'hot box', 'chaussee', 'chaussée'))
|
|
)
|
|
if is_road_repair:
|
|
roadway_ct = ContentType.objects.filter(app_label='assets', model='roadway').first()
|
|
if roadway_ct:
|
|
candidate_cts.append(roadway_ct)
|
|
|
|
# 3. Résolution spatiale du plus proche asset
|
|
centroid = None
|
|
if intervention.geom:
|
|
try:
|
|
centroid = intervention.geom.centroid
|
|
except Exception as e:
|
|
logger.debug(f"Error computing centroid from intervention.geom: {e}")
|
|
|
|
if not centroid and intervention.lon and intervention.lat:
|
|
try:
|
|
p = Point(float(intervention.lon), float(intervention.lat), srid=4326)
|
|
p.transform(3812)
|
|
centroid = p
|
|
except Exception as e:
|
|
logger.debug(f"Error transforming intervention coordinates to 3812: {e}")
|
|
|
|
# Récupérer les rues associées à l'intervention
|
|
street_ids = []
|
|
if hasattr(intervention, 'locations'):
|
|
for loc in intervention.locations.all():
|
|
if loc.content_type and loc.content_type.model == 'roadstreet':
|
|
street_ids.append(loc.object_id)
|
|
|
|
for ct in candidate_cts:
|
|
model_cls = ct.model_class()
|
|
if not model_cls:
|
|
continue
|
|
|
|
has_geom = any(
|
|
isinstance(f, gis_models.GeometryField)
|
|
for f in model_cls._meta.fields
|
|
if hasattr(f, 'geom_type')
|
|
) or hasattr(model_cls, 'geom')
|
|
|
|
if has_geom and centroid:
|
|
# Recherche prioritaire sur la même rue
|
|
if street_ids and hasattr(model_cls, 'street'):
|
|
closest = model_cls.objects.filter(
|
|
street_id__in=street_ids,
|
|
geom__isnull=False
|
|
).annotate(distance=Distance('geom', centroid)).order_by('distance').first()
|
|
if closest:
|
|
return ct, closest.pk
|
|
|
|
# Recherche globale du plus proche
|
|
closest = model_cls.objects.filter(
|
|
geom__isnull=False
|
|
).annotate(distance=Distance('geom', centroid)).order_by('distance').first()
|
|
if closest:
|
|
return ct, closest.pk
|
|
|
|
# Sans géométrie ou centroïde : filtrer par rue si possible
|
|
if street_ids and hasattr(model_cls, 'street'):
|
|
obj = model_cls.objects.filter(street_id__in=street_ids).first()
|
|
if obj:
|
|
return ct, obj.pk
|
|
|
|
first_obj = model_cls.objects.first()
|
|
if first_obj:
|
|
return ct, first_obj.pk
|
|
|
|
# 4. Fallback sur la première localisation ou l'intervention
|
|
first_loc = intervention.locations.first() if hasattr(intervention, 'locations') else None
|
|
if first_loc and first_loc.content_type_id and first_loc.object_id:
|
|
return first_loc.content_type, first_loc.object_id
|
|
|
|
return ContentType.objects.get_for_model(intervention), intervention.pk
|
|
|
|
|
|
def schedule_post_intervention_inspection(intervention):
|
|
"""
|
|
Planifie automatiquement une inspection de durabilité pour une intervention
|
|
lorsque son sous-type requiert une inspection post-travaux (ex: Hot Box).
|
|
"""
|
|
from .models import Inspection, InspectionZone
|
|
|
|
sub_type = getattr(intervention, 'sub_type', None)
|
|
if not sub_type or not getattr(sub_type, 'requires_post_inspection', False):
|
|
return None
|
|
|
|
# Éviter les doublons : vérifier si une inspection post-intervention non annulée existe déjà
|
|
existing = Inspection.objects.filter(
|
|
checked_intervention=intervention,
|
|
nature=sub_type.post_inspection_nature
|
|
).exclude(status='cancelled').first()
|
|
if existing:
|
|
# Si l'inspection existante pointe sur une localisation (ex: RoadStreet) et qu'un asset plus précis est trouvé
|
|
if existing.asset_content_type and existing.asset_content_type.model == 'roadstreet':
|
|
better_ct, better_id = _resolve_inspection_target(intervention, sub_type)
|
|
if better_ct and better_ct.model != 'roadstreet':
|
|
existing.asset_content_type = better_ct
|
|
existing.asset_object_id = better_id
|
|
existing.save(update_fields=['asset_content_type', 'asset_object_id'])
|
|
logger.info(f"Updated inspection {existing.pk} target from roadstreet to {better_ct.model} #{better_id}")
|
|
return existing
|
|
|
|
# Calcul des dates de la fenêtre d'inspection [fin + x, fin + x + y]
|
|
ref_date = intervention.end_time or timezone.now()
|
|
min_days = sub_type.post_inspection_delay_min_days or 30
|
|
max_days = sub_type.post_inspection_delay_max_days or 60
|
|
scheduled_start_date = ref_date.date() + timedelta(days=min_days)
|
|
scheduled_due_date = ref_date.date() + timedelta(days=max_days)
|
|
|
|
# Détermination de la zone d'inspection
|
|
thematic = intervention.thematic
|
|
zones = InspectionZone.objects.filter(thematic=thematic, is_active=True) if thematic else InspectionZone.objects.none()
|
|
zone = None
|
|
|
|
# 1. Résolution spatiale par intersection / inclusion
|
|
if zones.exists():
|
|
if intervention.geom:
|
|
zone = zones.filter(geom__intersects=intervention.geom).first()
|
|
|
|
if not zone and intervention.lon and intervention.lat:
|
|
try:
|
|
from django.contrib.gis.geos import Point
|
|
p = Point(float(intervention.lon), float(intervention.lat), srid=4326)
|
|
p.transform(3812)
|
|
zone = zones.filter(geom__contains=p).first()
|
|
except Exception as e:
|
|
logger.debug(f"Error resolving inspection zone via coordinates: {e}")
|
|
|
|
# 2. Résolution administrative par commune associée
|
|
if not zone:
|
|
from common.models import Municipality
|
|
for loc in intervention.locations.all():
|
|
loc_obj = loc.location
|
|
muni_to_check = loc_obj if isinstance(loc_obj, Municipality) else getattr(loc_obj, 'municipality', None)
|
|
if muni_to_check:
|
|
zone = zones.filter(municipalities=muni_to_check).first()
|
|
if zone:
|
|
break
|
|
|
|
# Détermination de la cible (Asset, Localisation ou Intervention elle-même)
|
|
target_ct, target_id = _resolve_inspection_target(intervention, sub_type)
|
|
|
|
# Pré-assignation si un seul inspecteur est assigné à la zone
|
|
assigned_inspector = None
|
|
if zone:
|
|
zone_inspectors = list(zone.inspectors.all()[:2])
|
|
if len(zone_inspectors) == 1:
|
|
assigned_inspector = zone_inspectors[0]
|
|
|
|
from .models import Inspection, InspectionZone, HotBoxInspection
|
|
|
|
is_hotbox = any(k in (sub_type.code or '').lower() or k in (sub_type.name_fr or '').lower() for k in ('hotbox', 'hot_box', 'hot box'))
|
|
inspection_cls = HotBoxInspection if is_hotbox else Inspection
|
|
|
|
# Création de l'inspection planifiée
|
|
notes = f"Inspection de durabilité planifiée pour {sub_type.get_name()} (Intervention {intervention.code})."
|
|
inspection = inspection_cls.objects.create(
|
|
asset_content_type=target_ct,
|
|
asset_object_id=target_id,
|
|
checked_intervention=intervention,
|
|
nature=sub_type.post_inspection_nature,
|
|
status='scheduled',
|
|
result_status='pending',
|
|
scheduled_start_date=scheduled_start_date,
|
|
scheduled_due_date=scheduled_due_date,
|
|
zone=zone,
|
|
inspector=assigned_inspector,
|
|
notes=notes,
|
|
)
|
|
logger.info(f"Scheduled post-intervention inspection {inspection.pk} created for intervention {intervention.code}")
|
|
return inspection
|