utilise la lib processing pour les dissolved buffers

core/checking.py

@@ -9,9 +9,6 @@ import re
 import sys
 import traceback
 
-from PyQt5.QtCore import QObject
-
-
 UNKNOWN = 0
 SUCCESS = 1
 FAILURE = 2

core/mncheck.py

@@ -6,11 +6,12 @@ import importlib
 import inspect
 import logging
 import pkgutil
-from qgis.core import QgsWkbTypes, QgsGeometry, QgsPoint
+from qgis.core import QgsProject, QgsWkbTypes, QgsGeometry, QgsPoint
 
-from PyQt5.QtCore import QVariant, pyqtSignal, QObject
+from PyQt5.QtCore import QVariant
 
-from core.checking import BaseChecker, Comlink
+from core.checking import BaseChecker
+from plugins import processing
 
 
 logger = logging.getLogger("mncheck")
@@ -107,6 +108,23 @@ class QgsModel():
                 
         return [QgsPoint(p) for p in temp_geom]
 
+    @classmethod
+    def full_buffer(cls, distance):
+        layer = next((l for l in QgsProject.instance().mapLayers().values() if l.name().lower() == cls.layername.lower()))
+
+        result = processing.run("native:buffer", {'INPUT':layer.dataProvider().dataSourceUri(),  #@UndefinedVariable
+                                                 'DISTANCE':distance,
+                                                 'SEGMENTS':5,
+                                                 'END_CAP_STYLE':0,
+                                                 'JOIN_STYLE':0,
+                                                 'MITER_LIMIT':2,
+                                                 'DISSOLVE':True,
+                                                 'OUTPUT':'memory:mncheck_temp'})
+
+        buffer = next((f for f in result['OUTPUT'].getFeatures()))
+        
+        return buffer.geometry()
+
 
 def validate(schema_name):
     try:

schemas/mn1_rec.py

@@ -470,12 +470,15 @@ class Mn1Checker(BaseChecker):
                          cable_points[-1].distanceSquared(equip_a_point) <= TOLERANCE))):
             
                 self.log_error("Pas d'équipement aux coordonnées attendues", item=cable)
-            
+    
     def test_tranchee_artere(self):
         """ Topologie: Tranchées / Artères
         Compare la géométrie des tranchées à celle des artères """
 
-        arteres_full_buffer = QgsGeometry().unaryUnion([a.geom for a in self.arteres]).buffer(TOLERANCE, 100)
+        arteres_full_buffer = Artere.full_buffer(TOLERANCE)
+        
+        if not arteres_full_buffer.isGeosValid():
+            raise ValueError("Buffer: géométrie invalide")
         
         for tranchee in self.tranchees:
             if not arteres_full_buffer.contains(tranchee.geom):
@@ -485,17 +488,27 @@ class Mn1Checker(BaseChecker):
         """ Topologie: Cables / Artères
         Compare la géométrie des cables à celle des artères """
         
-        arteres_full_buffer = QgsGeometry().unaryUnion([a.geom for a in self.arteres]).buffer(TOLERANCE, 100)
+        # Vérifie que chaque cable a au moins une artère (sauf si commentaire contient 'baguette')
+        arteres_full_buffer = Artere.full_buffer(TOLERANCE)
+        
+        if not arteres_full_buffer.isGeosValid():
+            raise ValueError("Buffer: géométrie invalide")
         
-        # Verifie que chaque cable a au moins une artère (sauf si commentaire contient 'baguette')
         for cable in self.cables:
             if "baguette" in cable.CA_COMMENT.lower() or not cable.is_geometry_valid():
                 continue
             if not arteres_full_buffer.contains(cable.geom):
                 self.log_error("Cable ou portion de cable sans artère", item=cable)
         
-        # Verifie que chaque artère a au moins un cable (sauf si commentaire contient un de ces mots 'racco client adductio attente bus 'sans cable'')
-        cables_full_buffer = QgsGeometry().unaryUnion([c.geom for c in self.cables]).buffer(TOLERANCE, 100)
+    def test_artere_cable(self):
+        """ Topologie: Artères / Cables
+        Compare la géométrie des artères à celle des cables """
+        
+        # Vérifie que chaque artère a au moins un cable (sauf si commentaire contient un de ces mots 'racco client adductio attente bus 'sans cable'')
+        cables_full_buffer = Cable.full_buffer(TOLERANCE)
+        
+        if not cables_full_buffer.isGeosValid():
+            raise ValueError("Buffer: géométrie invalide")
         
         for artere in self.arteres:
             if any(x in artere.AR_COMMENT.lower() for x in ['racco','client','adductio','attente','bus','sans cable']):

schemas/mn2_rec.py

@@ -512,7 +512,10 @@ class Mn2Checker(BaseChecker):
         """ Topologie: Tranchées / Artères
         Compare la géométrie des tranchées à celle des artères """
 
-        arteres_full_buffer = QgsGeometry().unaryUnion([a.geom for a in self.arteres]).buffer(TOLERANCE, 100)
+        arteres_full_buffer = Artere.full_buffer(TOLERANCE)
+        
+        if not arteres_full_buffer.isGeosValid():
+            raise ValueError("Buffer: géométrie invalide")
         
         for tranchee in self.tranchees:
             if not arteres_full_buffer.contains(tranchee.geom):
@@ -522,17 +525,27 @@ class Mn2Checker(BaseChecker):
         """ Topologie: Cables / Artères
         Compare la géométrie des cables à celle des artères """
         
-        arteres_full_buffer = QgsGeometry().unaryUnion([a.geom for a in self.arteres]).buffer(TOLERANCE, 100)
+        # Vérifie que chaque cable a au moins une artère (sauf si commentaire contient 'baguette')
+        arteres_full_buffer = Artere.full_buffer(TOLERANCE)
+        
+        if not arteres_full_buffer.isGeosValid():
+            raise ValueError("Buffer: géométrie invalide")
         
-        # Verifie que chaque cable a au moins une artère (sauf si commentaire contient 'baguette')
         for cable in self.cables:
             if "baguette" in cable.CA_COMMENT.lower() or not cable.is_geometry_valid():
                 continue
             if not arteres_full_buffer.contains(cable.geom):
                 self.log_error("Cable ou portion de cable sans artère", item=cable)
         
-        # Verifie que chaque artère a au moins un cable (sauf si commentaire contient un de ces mots 'racco client adductio attente bus 'sans cable'')
-        cables_full_buffer = QgsGeometry().unaryUnion([c.geom for c in self.cables]).buffer(TOLERANCE, 100)
+    def test_artere_cable(self):
+        """ Topologie: Artères / Cables
+        Compare la géométrie des artères à celle des cables """
+        
+        # Vérifie que chaque artère a au moins un cable (sauf si commentaire contient un de ces mots 'racco client adductio attente bus 'sans cable'')
+        cables_full_buffer = Cable.full_buffer(TOLERANCE)
+        
+        if not cables_full_buffer.isGeosValid():
+            raise ValueError("Buffer: géométrie invalide")
         
         for artere in self.arteres:
             if any(x in artere.AR_COMMENT.lower() for x in ['racco','client','adductio','attente','bus','sans cable']):