fix: set_detection_control_status implemented for DetectionData

Author: stephenprn

core/management/commands/import_detections.py

@@ -341,7 +341,7 @@ def queue_detection(self, detection_row: Dict[str, Any]):
 
             if linked_detection_same_tileset:
                 log_event(f"Detection already exists in tileset {self.tile_set.name}, id: {
-                      linked_detection_same_tileset.id}. Skipping...")
+                    linked_detection_same_tileset.id}. Skipping...")
                 return
         else:
             linked_detections = DetectionService.get_linked_detections(
@@ -404,15 +404,15 @@ def queue_detection(self, detection_row: Dict[str, Any]):
                 detection_object.address = serialized_detection["address"]
                 detection_object.save()
 
-            if not detection_data.detection_control_status:
-                detection_data.detection_control_status = (
-                    linked_detection.detection_data.detection_control_status
-                )
-
             if not detection_data.detection_validation_status:
                 detection_data.detection_validation_status = (
                     linked_detection.detection_data.detection_validation_status
                 )
+
+            if not detection_data.detection_control_status:
+                detection_data.set_detection_control_status(
+                    linked_detection.detection_data.detection_control_status
+                )
         else:
             parcel = (
                 Parcel.objects.filter(geometry__contains=centroid)
@@ -449,7 +449,7 @@ def queue_detection(self, detection_row: Dict[str, Any]):
             self.detection_objects_to_insert.append(detection_object)
 
             if not detection_data.detection_control_status:
-                detection_data.detection_control_status = (
+                detection_data.set_detection_control_status(
                     DetectionControlStatus.NOT_CONTROLLED
                 )
 
@@ -510,7 +510,7 @@ def insert_detections(self, force=False):
             )
         else:
             log_event(f"Inserted {
-                  self.total_inserted_detections} detections in total")
+                self.total_inserted_detections} detections in total")
 
         log_event(f"Elapsed time: {datetime.now() - self.start_time}")
 

core/management/commands/import_pvs.py

@@ -140,7 +140,7 @@ def handle(self, *args, **options):
                         detection.detection_data.official_report_date = (
                             official_report_date
                         )
-                    detection.detection_data.detection_control_status = (
+                    detection.detection_data.set_detection_control_status(
                         detection_control_status
                     )
 

core/migrations/0115_add_calculate_surface_function.py

@@ -0,0 +1,28 @@
+# Generated by Django 5.0.6 on 2025-12-19 14:54
+
+from django.db import migrations
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("core", "0114_email_alter_historicaluser_date_joined_and_more"),
+    ]
+
+    operations = [
+        migrations.RunSQL(
+            sql="""
+                CREATE OR REPLACE FUNCTION calculate_surface_m2(geom geometry)
+                RETURNS double precision
+                LANGUAGE plpgsql
+                IMMUTABLE
+                AS $$
+                BEGIN
+                    -- For metropolitan France, use Lambert-93 (EPSG:2154)
+                    -- For other territories, you might need different projections
+                    RETURN ST_Area(ST_Transform(geom, 2154));
+                END;
+$$;
+            """,
+            reverse_sql="DROP FUNCTION IF EXISTS calculate_surface_m2(geometry);",
+        ),
+    ]

core/models/detection_data.py

@@ -75,6 +75,16 @@ class DetectionData(TimestampedModelMixin, UuidModelMixin, DeletableModelMixin):
         bases=[HistoriedModelMixin], cascade_delete_history=True
     )
 
+    def set_detection_control_status(self, value: DetectionControlStatus):
+        self.detection_control_status = value
+
+        if (
+            value != DetectionControlStatus.NOT_CONTROLLED
+            and self.detection_validation_status
+            == DetectionValidationStatus.DETECTED_NOT_VERIFIED
+        ):
+            self.detection_validation_status = DetectionValidationStatus.SUSPECT
+
     class Meta:
         indexes = UuidModelMixin.Meta.indexes + [
             models.Index(fields=["detection_validation_status"]),

core/serializers/detection_data.py

@@ -192,14 +192,13 @@ def update(self, instance: DetectionData, validated_data):
             )
             detection_authorization.save()
 
+        detection_control_status = validated_data.pop("detection_control_status", None)
+
         for key, value in validated_data.items():
             setattr(instance, key, value)
 
-        if (
-            instance.detection_validation_status
-            == DetectionValidationStatus.DETECTED_NOT_VERIFIED
-        ):
-            instance.detection_validation_status = DetectionValidationStatus.SUSPECT
+        if detection_control_status:
+            instance.set_detection_control_status(detection_control_status)
 
         instance.user_last_update = user
         instance.save()

core/services/detection_bulk_update.py

@@ -2,7 +2,6 @@
 from django.contrib.gis.geos import MultiPolygon
 
 from core.models.detection import Detection
-from core.models.detection_data import DetectionValidationStatus
 from core.models.object_type import ObjectType
 from core.permissions.user import UserPermission
 from django.core.exceptions import BadRequest
@@ -40,9 +39,6 @@ def update_multiple_detections(
                     detection, detection_data_fields_to_update, update_data
                 )
 
-                # Apply business rules for validation status
-                self._apply_validation_status_rules(detection)
-
                 detection.detection_data.user_last_update = self.user
                 detection_datas_to_update.append(detection.detection_data)
 
@@ -99,22 +95,18 @@ def _update_detection_data_fields(
     ) -> None:
         """Update detection data fields with new values."""
         for field in fields_to_update:
+            if field == "detection_control_status":
+                detection.detection_data.set_detection_control_status(
+                    update_data[field]
+                )
+                continue
+
             setattr(
                 detection.detection_data,
                 field,
                 update_data[field],
             )
 
-    def _apply_validation_status_rules(self, detection: Detection) -> None:
-        """Apply business rules for validation status changes."""
-        if (
-            detection.detection_data.detection_validation_status
-            == DetectionValidationStatus.DETECTED_NOT_VERIFIED
-        ):
-            detection.detection_data.detection_validation_status = (
-                DetectionValidationStatus.SUSPECT
-            )
-
     def _perform_bulk_updates(
         self,
         detection_datas_to_update: List,

core/services/detection_process.py

@@ -62,7 +62,7 @@ def merge_double_detections(tile_set_id: int):
             detection_to_keep.detection_source = extract_higest_priority_value(
                 detections, "detection_source"
             )
-            detection_to_keep.detection_data.detection_control_status = (
+            detection_to_keep.detection_data.set_detection_control_status(
                 extract_higest_priority_value(
                     detections_data, "detection_control_status"
                 )

core/services/external_api.py

@@ -69,19 +69,11 @@ def update_control_status(
         for detection_object in parcel.detection_objects.all():
             for detection in detection_object.detections.all():
                 detection_data = detection.detection_data
-                detection_data.detection_control_status = control_status
+                detection_data.set_detection_control_status(control_status)
                 detection_data.detection_validation_status_change_reason = (
                     DetectionValidationStatusChangeReason.EXTERNAL_API
                 )
 
-                if (
-                    detection_data.detection_validation_status
-                    == DetectionValidationStatus.DETECTED_NOT_VERIFIED
-                ):
-                    detection_data.detection_validation_status = (
-                        DetectionValidationStatus.SUSPECT
-                    )
-
                 detections_data_to_update.append(detection_data)
 
         bulk_update_with_history(

core/services/prior_letter.py

@@ -73,7 +73,7 @@ def _update_control_status(self, detection_object: DetectionObject) -> None:
         detections_to_update = []
         for detection in detection_object.detections.all():
             if detection.detection_data:
-                detection.detection_data.detection_control_status = (
+                detection.detection_data.set_detection_control_status(
                     DetectionControlStatus.PRIOR_LETTER_SENT
                 )
                 detection.detection_data.user_last_update = self.user

core/views/utils/generate_prior_letter.py

@@ -83,7 +83,7 @@ def update_control_status(detection_object: DetectionObject, user: User):
     ]:
         return
 
-    detection.detection_data.detection_control_status = (
+    detection.detection_data.set_detection_control_status(
         DetectionControlStatus.PRIOR_LETTER_SENT
     )
     detection.detection_data.save()