add tests to find prises with no zapbo

MnCheck/schemas/mn1_rec.py

@@ -188,6 +188,7 @@ models = [Artere, Cable, Equipement, Noeud, Tranchee, Zapbo]
 class Prise(QgsModel):
     layername = "FTTH_SITES_GEO_VALIDE"
     layername_bis = "SITE_GEO"
+    pk = "ID_PRISE"
 
 
 ####### Validateur
@@ -224,12 +225,34 @@ class Mn1Checker(BaseChecker):
         Prise.layer = next((l for l in QgsProject.instance().mapLayers().values() \
                           if l.name().lower() in [Prise.layername.lower(), Prise.layername_bis.lower()]), None)
         
+        # Zapm partielles concernées, déduites des prises
+        self.zapm_pa = []
+        
     def _load_prises(self):
-        """ on charge la couche des prises si elle existe """
-        self.prises = []
-        if Prise.layer:
-            prises = [Prise(f) for f in Prise.layer.getFeatures()]
-            self.prises = [p for p in prises if p.T_ETAT.lower() != "obsolete"]
+        """ on charge la couche des prises si elle existe et qu'elle n'est pas déjà chargée """
+        if not self.prises:
+            self.prises = []
+            self.zapmp = []
+            if Prise.layer:
+                prises = [Prise(f) for f in Prise.layer.getFeatures()]
+                self.prises = [p for p in prises if p.T_ETAT.lower() != "obsolete"]
+            
+                for prise in self.prises:
+                    prise.zapbos = []
+                
+                if self.zapbos:
+                    
+                    for zapbo in self.zapbos:
+                        zapbo.prises = []
+                        
+                        for prise in self.prises:
+                            if zapbo.geom.contains(prise.geom):
+                                zapbo.prises.append(prise)
+                                prise.zapbos.append(zapbo)
+                    
+                                if not prise.ID_ZAPM_PA in self.zapm_pa:
+                                    # la prise est dans une zapbo: on ajoute son id de zapm partielles à la liste
+                                    self.zapm_pa.append(prise.ID_ZAPM_PA)
             
     def test_load_layers(self):
         """ Chargement des données
@@ -719,9 +742,14 @@ class Mn1Checker(BaseChecker):
             raise CheckingException("La couche des prises n'est pas chargée")
         
         for zapbo in self.zapbos:
-            if not next((prise for prise in self.prises if zapbo.geom.contains(prise.geom)), 0):
+            if not zapbo.prises:
                 self.log_error("La Zapbo ne contient aucune prise", item=zapbo)
         
+        for prise in self.prises:
+            if prise.ID_ZAPM_PA in self.zapm_pa and not prise.zapbos:
+                self.log_error("La prise n'est contenue dans aucune ZAPBO", item=prise)
+        
+        
     def test_pbo_dimension(self):
         """ Dimensionnement des PBO 
         Compare la couche des prises (si chargée) au dimensionnement des PBO"""
@@ -739,7 +767,7 @@ class Mn1Checker(BaseChecker):
             if not zapbo:
                 continue
                 
-            zapbo.nb_prises = sum([int(prise.NB_PRISE) for prise in self.prises if zapbo.geom.contains(prise.geom) and prise.NB_PRISE])
+            zapbo.nb_prises = sum([int(prise.NB_PRISE) for prise in zapbo.prises if prise.NB_PRISE])
             
             # Controle du dimensionnement des PBO
             if equipement.EQ_TYPE_PH == 'PBO 6' and not zapbo.nb_prises < 6:

MnCheck/schemas/mn2_rec.py

@@ -210,6 +210,7 @@ models = [Artere, Cable, Equipement, Noeud, Tranchee, Zapbo]
 class Prise(QgsModel):
     layername = "FTTH_SITES_GEO_VALIDE"
     layername_bis = "SITE_GEO"
+    pk = "ID_PRISE"
     
 ####### Validateur
 
@@ -246,13 +247,35 @@ class Mn2Checker(BaseChecker):
         Prise.layer = next((l for l in QgsProject.instance().mapLayers().values() \
                           if l.name().lower() in [Prise.layername.lower(), Prise.layername_bis.lower()]), None)
         
+        # Zapm partielles concernées, déduites des prises
+        self.zapm_pa = []
+        
     def _load_prises(self):
         """ on charge la couche des prises si elle existe """
-        self.prises = []
-        if Prise.layer:
-            prises = [Prise(f) for f in Prise.layer.getFeatures()]
-            self.prises = [p for p in prises if p.T_ETAT.lower() != "obsolete"]
-        
+        if not self.prises:
+            self.prises = []
+            
+            if Prise.layer:
+                prises = [Prise(f) for f in Prise.layer.getFeatures()]
+                self.prises = [p for p in prises if p.T_ETAT.lower() != "obsolete"]
+                
+                for prise in self.prises:
+                    prise.zapbos = []
+                
+                if self.zapbos:
+                    
+                    for zapbo in self.zapbos:
+                        zapbo.prises = []
+                        
+                        for prise in self.prises:
+                            if zapbo.geom.contains(prise.geom):
+                                zapbo.prises.append(prise)
+                                prise.zapbos.append(zapbo)
+                    
+                                if not prise.ID_ZAPM_PA in self.zapm_pa:
+                                    # la prise est dans une zapbo: on ajoute son id de zapm partielles à la liste
+                                    self.zapm_pa.append(prise.ID_ZAPM_PA)
+                                
     def test_load_layers(self):
         """ Chargement des données
             Contrôle la présence des couches attendues
@@ -848,9 +871,13 @@ class Mn2Checker(BaseChecker):
             raise CheckingException("La couche des prises n'est pas chargée")
         
         for zapbo in self.zapbos:
-            if not next((prise for prise in self.prises if zapbo.geom.contains(prise.geom)), 0):
+            if not zapbo.prises:
                 self.log_error("La Zapbo ne contient aucune prise", item=zapbo)
         
+        for prise in self.prises:
+            if prise.ID_ZAPM_PA in self.zapm_pa and not prise.zapbos:
+                self.log_error("La prise n'est contenue dans aucune ZAPBO", item=prise)
+        
     def test_pbo_dimension(self):
         """ Dimensionnement des PBO 
         Compare la couche des prises (si chargée) au dimensionnement des PBO"""
@@ -868,7 +895,7 @@ class Mn2Checker(BaseChecker):
             if not zapbo:
                 continue
                 
-            zapbo.nb_prises = sum([int(prise.NB_PRISE) for prise in self.prises if zapbo.geom.contains(prise.geom) and prise.NB_PRISE])
+            zapbo.nb_prises = sum([int(prise.NB_PRISE) for prise in zapbo.prises if prise.NB_PRISE])
             
             # Controle du dimensionnement des PBO
             if equipement.EQ_TYPE_PH == 'PBO 6' and not zapbo.nb_prises < 6:

MnCheck/test/test_mn1_rec.py

@@ -550,6 +550,7 @@ class Test(SchemaTest):
         self.assertEqual(len(r.errors), 0)
         
         # pas de prise dans la zapbo
+        self.checker.prises = []
         new_geom = self.checker.zapbos[0]._feature.geometry()
         new_geom.translate(1000,1000)
         self.checker.zapbos[0]._feature.setGeometry(new_geom)
@@ -559,6 +560,7 @@ class Test(SchemaTest):
         self.assertErrorLogged(r, "La Zapbo ne contient aucune prise")
         
         # couche non chargée
+        self.checker.prises = []
         self.schema.Prise.layer = None
 
         r = self.run_test("test_zapbos_prises", True)
@@ -576,6 +578,7 @@ class Test(SchemaTest):
         self.assertEqual(len(r.errors), 0)
         
         # couche non chargée
+        self.checker.prises = []
         self.schema.Prise.layer = None
         
         r = self.run_test("test_pbo_dimension", True)

MnCheck/test/test_mn2_rec.py

@@ -576,6 +576,7 @@ class Test(SchemaTest):
         self.assertEqual(len(r.errors), 0)
 
         # pas de prise dans la zapbo
+        self.checker.prises = []
         new_geom = self.checker.zapbos[0]._feature.geometry()
         new_geom.translate(1000,1000)
         self.checker.zapbos[0]._feature.setGeometry(new_geom)
@@ -585,6 +586,7 @@ class Test(SchemaTest):
         self.assertErrorLogged(r, "La Zapbo ne contient aucune prise")
         
         # couche non chargée
+        self.checker.prises = []
         self.schema.Prise.layer = None
 
         r = self.run_test("test_zapbos_prises", True)
@@ -602,6 +604,7 @@ class Test(SchemaTest):
         self.assertEqual(len(r.errors), 0)
         
         # couche non chargée
+        self.checker.prises = []
         self.schema.Prise.layer = None
         
         r = self.run_test("test_pbo_dimension", True)