Transcription des fonctions osgeo en fonctions pyqgis (schéma MN2)

schemas/mn1_rec.py

@@ -310,7 +310,7 @@ class Validator(BaseValidator):
         
         for tranchee in tranchees:
             if not arteres_full_buffer.contains(tranchee.geom):
-                self.log_error(MissingItem("Tranchée sans artère ('{}')".format(tranchee), layername=Tranchee.layername, field="*"))
+                self.log_error(MissingItem("Tranchée ou portion de tranchée sans artère ('{}')".format(tranchee), layername=Tranchee.layername, field="*"))
         
         # Verifie que chaque cable a au moins une artère (sauf si commentaire contient 'baguette')
         for cable in cables:
@@ -371,7 +371,7 @@ class Validator(BaseValidator):
                 self.log_error(MissingItem("Le nom du PBO ne coincide avec le nom d'aucune des ZAPBO qui le contient: {}".format(equipement), layername=Equipement.layername, field="EQ_NOM"))
                 break
              
-            if not hasattr(equipement, "nb_prises") or equipement.zapbo.nb_prises is None:
+            if not hasattr(equipement.zapbo, "nb_prises") or equipement.zapbo.nb_prises is None:
                 equipement.zapbo.nb_prises = 0
              
             # Controle du dimensionnement des PBO

schemas/mn2_rec.py

@@ -2,10 +2,9 @@
 
 @author: olivier.massot, 2018
 '''
-from osgeo.ogr import Feature
+from qgis.core import QgsGeometry
 
-from core import constants
-from core.cerberus_extend import is_float, is_multi_int, is_int, \
+from core.cerberus_ import is_float, is_multi_int, is_int, \
     is_modern_french_date
 from core.validator import QgsModel, BaseValidator
 from core.validation_errors import UniqueError, InvalidGeometry, RelationError, \
@@ -241,26 +240,6 @@ class Validator(BaseValidator):
             else:
                 self.log_error(UniqueError("Doublons dans le champs: {}".format(zapbo), layername=Zapbo.layername, field="ID_ZAPBO"))
         
-        # contrôle de la validité des géométries
-        for artere in arteres:
-            if not artere.valid:
-                self.log_error(InvalidGeometry("Géométrie invalide: {}".format(artere), layername=Artere.layername, field="geom"))
-        for tranchee in tranchees:
-            if not tranchee.valid:
-                self.log_error(InvalidGeometry("Géométrie invalide: {}".format(tranchee), layername=Tranchee.layername, field="geom"))
-        for cable in cables:
-            if not "baguette" in cable.CA_COMMENT.lower() and not cable.valid:
-                self.log_error(InvalidGeometry("Géométrie invalide: {}".format(cable), layername=Cable.layername, field="geom"))
-        for noeud in noeuds.values():
-            if not noeud.valid:
-                self.log_error(InvalidGeometry("Géométrie invalide: {}".format(noeud), layername=Noeud.layername, field="geom"))
-        for equipement in equipements.values():
-            if not equipement.valid:
-                self.log_error(InvalidGeometry("Géométrie invalide: {}".format(equipement), layername=Equipement.layername, field="geom"))
-        for zapbo in zapbos.values():
-            if not zapbo.valid:
-                self.log_error(InvalidGeometry("Géométrie invalide: {}".format(zapbo), layername=Zapbo.layername, field="geom"))
-        
         # rattachement les noeuds aux artères     
         for artere in arteres:
             try:
@@ -340,55 +319,53 @@ class Validator(BaseValidator):
             if not artere.noeud_a or not artere.noeud_b:
                 continue
             
-            buffer_a, buffer_b = artere.points[0].Buffer(TOLERANCE), artere.points[-1].Buffer(TOLERANCE)
-            
-            if not (buffer_a.Contains(artere.noeud_a.points[0]) and buffer_b.Contains(artere.noeud_b.points[0])) \
-               and not (buffer_a.Contains(artere.noeud_b.points[0]) and buffer_b.Contains(artere.noeud_a.points[0])):
+            if not any(((artere.points[0].distanceSquared(artere.noeud_a.points[0]) <= TOLERANCE and \
+                         artere.points[-1].distanceSquared(artere.noeud_b.points[0]) <= TOLERANCE), 
+                        (artere.points[0].distanceSquared(artere.noeud_b.points[0]) <= TOLERANCE and \
+                         artere.points[-1].distanceSquared(artere.noeud_a.points[0]) <= TOLERANCE))):
 
                 self.log_error(MissingItem("Pas de noeud aux coordonnées attendues ('{}')".format(artere), layername=Artere.layername, field="geom"))
         
         
         # Cables: comparer la géométrie à celle des equipements (on utilise en fait la geom du noeud correspondant à l'équipement)
         for cable in cables:
-            if not cable.equipement_a or not cable.equipement_b or not cable.valid or not cable.equipement_a.noeud or not cable.equipement_b.noeud:
+            if not cable.equipement_a or not cable.equipement_b or not cable.is_geometry_valid() or not cable.equipement_a.noeud or not cable.equipement_b.noeud:
                 continue
             
-            buffer_a, buffer_b = cable.points[0].Buffer(TOLERANCE), cable.points[-1].Buffer(TOLERANCE)
-            
-            if not (buffer_a.Contains(cable.equipement_a.noeud.points[0]) and buffer_b.Contains(cable.equipement_b.noeud.points[0])) \
-               and not (buffer_a.Contains(cable.equipement_b.noeud.points[0]) and buffer_b.Contains(cable.equipement_a.noeud.points[0])):
+            if not any(((cable.points[0].distanceSquared(cable.equipement_a.noeud.points[0]) <= TOLERANCE and \
+                         cable.points[-1].distanceSquared(cable.equipement_b.noeud.points[0]) <= TOLERANCE), 
+                        (cable.points[0].distanceSquared(cable.equipement_b.noeud.points[0]) <= TOLERANCE and \
+                         cable.points[-1].distanceSquared(cable.equipement_a.noeud.points[0]) <= TOLERANCE))):
             
                 self.log_error(MissingItem("Pas d'equipement aux coordonnées attendues ('{}')".format(cable), layername=Cable.layername, field="geom"))
             
-        del buffer_a, buffer_b 
-        
         # Verifie que chaque tranchée a au moins une artère
-        arteres_emprise = Feature.buffered_union(arteres, TOLERANCE)
+        arteres_full_buffer = QgsGeometry().unaryUnion([a.geom for a in arteres]).buffer(TOLERANCE, 100)
         
         for tranchee in tranchees:
-            if not arteres_emprise.Contains(tranchee.geom):
-                self.log_error(MissingItem("Tranchée sans artère ('{}')".format(tranchee), layername=Tranchee.layername, field="-"))
+            if not arteres_full_buffer.contains(tranchee.geom):
+                self.log_error(MissingItem("Tranchée ou portion de tranchée sans artère ('{}')".format(tranchee), layername=Tranchee.layername, field="*"))
         
         
         # Verifie que chaque cable a au moins une artère (sauf si commentaire contient 'baguette')
         for cable in cables:
-            if "baguette" in cable.CA_COMMENT.lower() or not cable.valid:
+            if "baguette" in cable.CA_COMMENT.lower() or not cable.is_geometry_valid():
                 continue
-            if not arteres_emprise.Contains(cable.geom):
-                self.log_error(MissingItem("Cable sans artère ('{}')".format(cable), layername=Cable.layername, field="-"))
+            if not arteres_full_buffer.contains(cable.geom):
+                self.log_error(MissingItem("Cable ou portion de cable sans artère ('{}')".format(cable), layername=Cable.layername, field="*"))
         
-        del arteres_emprise
+        del arteres_full_buffer
         
         # 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_emprise = Feature.buffered_union(cables, TOLERANCE)
+        cables_full_buffer = QgsGeometry().unaryUnion([c.geom for c in cables]).buffer(TOLERANCE, 100)
         
         for artere in arteres:
             if any(x in artere.AR_COMMENT.lower() for x in ['racco','client','adductio','attente','bus','sans cable']):
                 continue
-            if not cables_emprise.Contains(artere.geom):
-                self.log_error(MissingItem("Artère sans cable ('{}')".format(artere), layername=Artere.layername, field="-"))
+            if not cables_full_buffer.contains(artere.geom):
+                self.log_error(MissingItem("Artère sans cable ('{}')".format(artere), layername=Artere.layername, field="*"))
                 
-        del cables_emprise
+        del cables_full_buffer
                 
         # Contrôle des dimensions logiques
         for artere in arteres:
@@ -416,7 +393,7 @@ class Validator(BaseValidator):
             #zapbos englobant l'equipement
             candidates = []
             for zapbo in zapbos.values():
-                if zapbo.geom.Contains(equipement.geom):
+                if zapbo.geom.contains(equipement.geom):
                     candidates.append(zapbo)
                     
             # le pbo doit être contenu dans une zapbo
@@ -431,20 +408,20 @@ class Validator(BaseValidator):
                 self.log_error(MissingItem("Le nom du PBO ne coincide avec le nom d'aucune des ZAPBO qui le contient: {}".format(equipement), layername=Equipement.layername, field="EQ_NOM"))
                 break
             
-            if equipement.zapbo.nb_prises is None:
+            if not hasattr(equipement.zapbo, "nb_prises") or equipement.zapbo.nb_prises is None:
                 equipement.zapbo.nb_prises = 0
             
             # Controle du dimensionnement des PBO
             if equipement.zapbo.nb_prises is not None:
                 if equipement.EQ_TYPE_PH == 'PBO 6' and not equipement.zapbo.nb_prises < 6:
-                    self.log_error(DimensionError("Le PBO 6 contient plus de 5 prises: {}".format(equipement), layername=Equipement.layername, field="-"))
+                    self.log_error(DimensionError("Le PBO 6 contient plus de 5 prises: {}".format(equipement), layername=Equipement.layername, field="*"))
             
                 if equipement.EQ_TYPE_PH == 'PBO 12' and not equipement.zapbo.nb_prises >= 6 and equipement.zapbo.nb_prises <= 8:
-                    self.log_error(DimensionError("Le PBO 12 contient mois de 6 prises ou plus de 8 prises: {}".format(equipement), layername=Equipement.layername, field="-"))
+                    self.log_error(DimensionError("Le PBO 12 contient mois de 6 prises ou plus de 8 prises: {}".format(equipement), layername=Equipement.layername, field="*"))
         
             if equipement.zapbo.STATUT == "REC" and not equipement.EQ_STATUT == "REC":
-                self.log_error(TechnicalValidationError("Le statut du PBO n'est pas cohérent avec le statut de sa ZAPBO: {}".format(equipement), layername=Equipement.layername, field="-"))
+                self.log_error(TechnicalValidationError("Le statut du PBO n'est pas cohérent avec le statut de sa ZAPBO: {}".format(equipement), layername=Equipement.layername, field="*"))
         
             if equipement.EQ_STATUT == "REC" and not equipement.zapbo.STATUT == "REC" and not equipement.zapbo.ID_ZAPBO[:4].lower() == "att_":
-                self.log_error(TechnicalValidationError("Le statut du PBO n'est pas cohérent avec le statut de sa ZAPBO: {}".format(equipement), layername=Equipement.layername, field="-"))
+                self.log_error(TechnicalValidationError("Le statut du PBO n'est pas cohérent avec le statut de sa ZAPBO: {}".format(equipement), layername=Equipement.layername, field="*"))