357 lines
14 KiB
Python
357 lines
14 KiB
Python
from django.contrib.gis.db import models
|
|
from django.contrib.gis.geos import Point
|
|
from django.contrib.contenttypes.models import ContentType
|
|
from django.contrib.contenttypes.fields import GenericRelation
|
|
|
|
from django.utils.translation import get_language, gettext_lazy as _
|
|
|
|
from common.models import Thematic
|
|
|
|
|
|
class GeoSyncMixin:
|
|
"""
|
|
Mixin providing spatial field synchronization (geom, lon, lat, geojson).
|
|
Automatically updates geojson and lon/lat when geom or lon/lat change.
|
|
"""
|
|
|
|
def sync_geom_and_geojson(self):
|
|
modified_fields = set()
|
|
|
|
old_lon = None
|
|
old_lat = None
|
|
old_geom = None
|
|
|
|
if self.pk:
|
|
try:
|
|
old_instance = type(self).objects.get(pk=self.pk)
|
|
old_lon = getattr(old_instance, 'lon', None)
|
|
old_lat = getattr(old_instance, 'lat', None)
|
|
old_geom = getattr(old_instance, 'geom', None)
|
|
except Exception:
|
|
pass
|
|
|
|
cur_geom = getattr(self, 'geom', None)
|
|
cur_lon = getattr(self, 'lon', None)
|
|
cur_lat = getattr(self, 'lat', None)
|
|
|
|
# Check if lon/lat were modified while geom remained unchanged
|
|
lon_lat_modified = (
|
|
cur_lon is not None and cur_lat is not None and
|
|
(cur_lon != old_lon or cur_lat != old_lat) and
|
|
(cur_geom == old_geom)
|
|
)
|
|
|
|
if lon_lat_modified:
|
|
# lon/lat were explicitly updated: rebuild geom from lon/lat
|
|
try:
|
|
pt_4326 = Point(cur_lon, cur_lat, srid=4326)
|
|
new_geojson = pt_4326.geojson
|
|
if getattr(self, 'geojson', None) != new_geojson:
|
|
self.geojson = new_geojson
|
|
modified_fields.add('geojson')
|
|
|
|
if hasattr(self, 'geom'):
|
|
field = type(self)._meta.get_field('geom')
|
|
geom_type = getattr(field, 'geom_type', '').upper()
|
|
if geom_type in ('POINT', 'GEOMETRY'):
|
|
srid = getattr(field, 'srid', 3812)
|
|
pt_target = pt_4326 if srid == 4326 else pt_4326.transform(srid, clone=True)
|
|
if getattr(self, 'geom', None) != pt_target:
|
|
self.geom = pt_target
|
|
modified_fields.add('geom')
|
|
except Exception:
|
|
pass
|
|
|
|
elif cur_geom is not None:
|
|
# geom is present and is the source of truth
|
|
try:
|
|
geom_4326 = cur_geom.transform(4326, clone=True)
|
|
new_geojson = geom_4326.geojson
|
|
if getattr(self, 'geojson', None) != new_geojson:
|
|
self.geojson = new_geojson
|
|
modified_fields.add('geojson')
|
|
|
|
if geom_4326.geom_type == 'Point':
|
|
target_point = geom_4326
|
|
else:
|
|
centroid = geom_4326.centroid
|
|
if geom_4326.intersects(centroid) or geom_4326.contains(centroid):
|
|
target_point = centroid
|
|
else:
|
|
target_point = geom_4326.point_on_surface
|
|
|
|
new_lon = target_point.x
|
|
new_lat = target_point.y
|
|
|
|
if getattr(self, 'lon', None) != new_lon:
|
|
self.lon = new_lon
|
|
modified_fields.add('lon')
|
|
if getattr(self, 'lat', None) != new_lat:
|
|
self.lat = new_lat
|
|
modified_fields.add('lat')
|
|
except Exception:
|
|
pass
|
|
|
|
elif cur_lon is not None and cur_lat is not None:
|
|
# geom is None, but lon and lat are provided
|
|
try:
|
|
pt_4326 = Point(cur_lon, cur_lat, srid=4326)
|
|
new_geojson = pt_4326.geojson
|
|
if getattr(self, 'geojson', None) != new_geojson:
|
|
self.geojson = new_geojson
|
|
modified_fields.add('geojson')
|
|
|
|
if hasattr(self, 'geom'):
|
|
field = type(self)._meta.get_field('geom')
|
|
geom_type = getattr(field, 'geom_type', '').upper()
|
|
if geom_type in ('POINT', 'GEOMETRY'):
|
|
srid = getattr(field, 'srid', 3812)
|
|
pt_target = pt_4326 if srid == 4326 else pt_4326.transform(srid, clone=True)
|
|
if getattr(self, 'geom', None) != pt_target:
|
|
self.geom = pt_target
|
|
modified_fields.add('geom')
|
|
except Exception:
|
|
pass
|
|
|
|
elif cur_geom is None and cur_lon is None and cur_lat is None:
|
|
if getattr(self, 'geojson', None) is not None:
|
|
self.geojson = None
|
|
modified_fields.add('geojson')
|
|
|
|
return modified_fields
|
|
|
|
def save(self, *args, **kwargs):
|
|
modified = self.sync_geom_and_geojson()
|
|
if 'update_fields' in kwargs and kwargs['update_fields'] is not None:
|
|
update_fields = set(kwargs['update_fields'])
|
|
update_fields.update(modified)
|
|
kwargs['update_fields'] = list(update_fields)
|
|
super().save(*args, **kwargs)
|
|
|
|
|
|
# Abstract base model for locations
|
|
class AbstractLocation(GeoSyncMixin, models.Model):
|
|
code = models.CharField(max_length=50)
|
|
name_fr = models.CharField(max_length=200,null=True,blank=True)
|
|
name_nl = models.CharField(max_length=200,null=True,blank=True)
|
|
lon = models.FloatField(null=True,blank=True)
|
|
lat = models.FloatField(null=True,blank=True)
|
|
geom = models.PolygonField(srid=3812, null=True,blank=True)
|
|
geojson = models.TextField(null=True, blank=True)
|
|
updated_at = models.DateTimeField(auto_now=True)
|
|
updated_by = models.ForeignKey('auth.User', on_delete=models.SET_NULL, null=True, blank=True, related_name='%(app_label)s_%(class)s_location_updated_by', verbose_name="Mis à jour par")
|
|
|
|
def get_name(self):
|
|
lang = get_language()
|
|
name = getattr(self, f"name_{lang}", self.name_fr) or "" # Fallback to name_fr
|
|
return name
|
|
|
|
def __str__(self):
|
|
return self.code + " - " + self.get_name()
|
|
|
|
class Meta:
|
|
abstract = True
|
|
ordering = ["code"]
|
|
|
|
|
|
|
|
# Category of assets
|
|
class AssetCategory(models.Model):
|
|
thematic = models.ForeignKey(Thematic, on_delete=models.SET_NULL, null=True, blank=True, related_name="asset_categories")
|
|
parent = models.ForeignKey(
|
|
'self',
|
|
on_delete=models.SET_NULL,
|
|
null=True,
|
|
blank=True,
|
|
related_name='children',
|
|
verbose_name=_("Catégorie parente"),
|
|
help_text=_("Laisser vide pour une catégorie de premier niveau. Une sous-catégorie ne peut pas avoir d'enfants."),
|
|
)
|
|
code = models.CharField(max_length=200)
|
|
name_fr = models.CharField(max_length=255)
|
|
name_nl = models.CharField(max_length=255)
|
|
allowed_models = models.ManyToManyField(
|
|
ContentType,
|
|
blank=True,
|
|
related_name="asset_categories",
|
|
verbose_name=_("Modèles Django autorisés"),
|
|
help_text=_("Les types de modèles d'assets (Django) autorisés pour cette catégorie.")
|
|
)
|
|
|
|
def get_name(self):
|
|
lang = get_language()
|
|
return getattr(self, f"name_{lang}", self.name_fr) # Fallback to name_fr
|
|
|
|
@property
|
|
def name(self):
|
|
return self.get_name()
|
|
|
|
def __str__(self):
|
|
return f"{self.get_name()}"
|
|
|
|
@property
|
|
def is_leaf(self):
|
|
"""Retourne True si cette catégorie n'a pas d'enfants (feuille de l'arbre)."""
|
|
return not self.children.exists()
|
|
|
|
def get_children(self):
|
|
"""Retourne le queryset des sous-catégories directes."""
|
|
return self.children.all()
|
|
|
|
def get_category_ids_with_children(self):
|
|
"""Retourne la liste des ids : cette catégorie + toutes ses sous-catégories."""
|
|
ids = [self.pk]
|
|
ids.extend(self.children.values_list('pk', flat=True))
|
|
return ids
|
|
|
|
class Meta:
|
|
ordering = ["thematic", "name_fr"]
|
|
|
|
|
|
# Abstract base model for "models of assets"
|
|
class AbstractAssetModel(models.Model):
|
|
code = models.CharField(max_length=50)
|
|
name_fr = models.CharField(max_length=200,null=True,blank=True)
|
|
name_nl = models.CharField(max_length=200,null=True,blank=True)
|
|
description = models.TextField(null=True,blank=True)
|
|
brand = models.CharField(max_length=500, null=True,blank=True)
|
|
category = models.ForeignKey(AssetCategory, on_delete=models.SET_NULL, null=True, blank=True)
|
|
|
|
def get_name(self):
|
|
lang = get_language()
|
|
name = getattr(self, f"name_{lang}", self.name_fr) or "" # Fallback to name_fr
|
|
return name
|
|
|
|
def __str__(self):
|
|
return self.code + " - " + self.get_name()
|
|
|
|
class Meta:
|
|
abstract = True
|
|
|
|
|
|
ASSET_STATUS_CHOICES = {
|
|
'to_be_approved': _('À approuver'),
|
|
'planned': _('Planifié'),
|
|
'active': _('Actif'),
|
|
'damaged': _('Endommagé'),
|
|
'to_replace': _('A remplacer'),
|
|
'voluntarily_out_of_service': _('Mis hors service volontairement'),
|
|
'removed': _('Enlevé'),
|
|
'archived': _('Archivé'),
|
|
}
|
|
|
|
ASSET_OBSOLESCENCE_CHOICES = {
|
|
'good': _('Bon état'),
|
|
'medium': _('Moyen'),
|
|
'bad': _('Dégradé'),
|
|
'critical': _('Critique'),
|
|
'unknown': _('Inconnu'),
|
|
}
|
|
|
|
ASSET_VALIDATION_STATUS_CHOICES = {
|
|
'to_be_validated': _('À valider'),
|
|
'validated': _('Validé'),
|
|
'to_be_completed': _('À compléter'),
|
|
'rejected': _('Rejeté'),
|
|
}
|
|
|
|
ASSET_INSPECTION_STATUS_CHOICES = {
|
|
'to_inspect': _('À inspecter'),
|
|
'compliant': _('Conforme'),
|
|
'non_compliant': _('Non conforme'),
|
|
'updated': _('Mis à jour'),
|
|
'unknown': _('Non inspecté'),
|
|
}
|
|
|
|
# Abstract base model for assets
|
|
class AbstractAsset(models.Model):
|
|
code = models.CharField(max_length=50)
|
|
name_fr = models.CharField(max_length=200,null=True,blank=True)
|
|
name_nl = models.CharField(max_length=200,null=True,blank=True)
|
|
status = models.CharField(choices=ASSET_STATUS_CHOICES, default='active', max_length=50)
|
|
obsolescence = models.CharField(choices=ASSET_OBSOLESCENCE_CHOICES, default='unknown', null=True, blank=True)
|
|
installation_date = models.DateTimeField(null=True,blank=True)
|
|
uninstallation_date = models.DateTimeField(null=True,blank=True)
|
|
serial_number = models.CharField(max_length=200,null=True,blank=True)
|
|
warranty_duration = models.IntegerField(null=True,blank=True, verbose_name="Duration of the warranty (months)")
|
|
last_inspection_date = models.DateTimeField(null=True,blank=True)
|
|
inspection_status = models.CharField(max_length=50, choices=ASSET_INSPECTION_STATUS_CHOICES, default='unknown', null=True, blank=True, verbose_name="Statut d'inspection")
|
|
last_preventive_date = models.DateTimeField(null=True,blank=True)
|
|
last_corrective_date = models.DateTimeField(null=True,blank=True)
|
|
last_survey_date = models.DateTimeField(null=True,blank=True)
|
|
category = models.ForeignKey(AssetCategory, on_delete=models.SET_NULL, null=True, blank=True)
|
|
brand = models.CharField(max_length=500, null=True,blank=True)
|
|
replaced_by = models.ForeignKey('self', on_delete=models.SET_NULL, null=True, blank=True, related_name='replaces', verbose_name="Remplacé par")
|
|
marked_as_removed_date = models.DateTimeField(null=True, blank=True, verbose_name="Date de marquage comme retiré")
|
|
marked_as_removed_by = models.ForeignKey('auth.User', on_delete=models.SET_NULL, null=True, blank=True, related_name='%(app_label)s_%(class)s_marked_as_removed', verbose_name="Marqué comme retiré par")
|
|
note = models.TextField(null=True, blank=True, verbose_name="Note")
|
|
validation_status = models.CharField(max_length=50, choices=ASSET_VALIDATION_STATUS_CHOICES, null=True, blank=True, verbose_name="Statut de validation")
|
|
funding_program = models.CharField(max_length=200, null=True, blank=True, verbose_name=_("Programme de financement / Projet"))
|
|
updated_at = models.DateTimeField(auto_now=True)
|
|
updated_by = models.ForeignKey('auth.User', on_delete=models.SET_NULL, null=True, blank=True, related_name='%(app_label)s_%(class)s_updated_by', verbose_name="Mis à jour par")
|
|
|
|
|
|
asset_inspections = GenericRelation(
|
|
'assets.AssetInspection',
|
|
content_type_field='asset_content_type',
|
|
object_id_field='asset_object_id',
|
|
related_query_name='%(app_label)s_%(class)s'
|
|
)
|
|
|
|
|
|
|
|
def get_replacement_chain(self):
|
|
"""Get the complete replacement chain starting from this asset."""
|
|
chain = [self]
|
|
current = self
|
|
seen_ids = {self.id}
|
|
|
|
while current.replaced_by:
|
|
if current.replaced_by.id in seen_ids: # Avoid infinite loops
|
|
break
|
|
chain.append(current.replaced_by)
|
|
seen_ids.add(current.replaced_by.id)
|
|
current = current.replaced_by
|
|
|
|
return chain
|
|
|
|
def get_original_asset(self):
|
|
"""Get the first asset in the replacement chain."""
|
|
original = self
|
|
seen_ids = {self.id}
|
|
|
|
# Try to find an asset that replaces this one
|
|
replacer = self.replaces.first()
|
|
while replacer:
|
|
if replacer.id in seen_ids: # Avoid infinite loops
|
|
break
|
|
original = replacer
|
|
seen_ids.add(replacer.id)
|
|
replacer = replacer.replaces.first()
|
|
|
|
return original
|
|
|
|
def get_name(self):
|
|
lang = get_language()
|
|
name = getattr(self, f"name_{lang}", self.name_fr) or "" # Fallback to name_fr
|
|
return name
|
|
|
|
def __str__(self):
|
|
return self.code + " - " + self.get_name()
|
|
|
|
class Meta:
|
|
abstract = True
|
|
|
|
|
|
# Abstract base model for geo assets
|
|
class AbstractGeoAsset(GeoSyncMixin, AbstractAsset):
|
|
lon = models.FloatField(null=True,blank=True)
|
|
lat = models.FloatField(null=True,blank=True)
|
|
geom = models.PointField(srid=3812, null=True,blank=True)
|
|
geojson = models.TextField(null=True, blank=True)
|
|
|
|
class Meta:
|
|
abstract = True
|
|
|
|
|
|
|