remove schema apd from public schemas

schemas/mn3_apd.py

@@ -1,309 +0,0 @@
-'''
-
-    Schéma de validation des données MN2
-
-@author: olivier.massot, 2018
-'''
-
-import logging
-from qgis.core import QgsProject
-
-from core.cerberus_ import is_positive_int, is_positive_float, CerberusValidator, \
-    CerberusErrorHandler, _translate_messages
-from core.checking import BaseChecker
-from core.mncheck import QgsModel
-
-# TODO: verifier que les zapbos intersectent au moins une zasro et même nom que cette zasro
-# TODO: verifier que toute les prises comprises dans une zapbo soient contenues dans une seule et même zasro
-
-
-
-
-logger = logging.getLogger("mncheck")
-
-SCHEMA_NAME = "Schéma MN v3 APD"
-
-XMIN, XMAX, YMIN, YMAX = 1341999, 1429750, 8147750, 8294000
-CRS = 'EPSG:3949' # Coordinate Reference System
-
-TOLERANCE = 0.5
-
-STATUTS = ['ETUDE PRELIMINAIRE', 'ETUDE DE DIAGNOSTIC', 'AVANT-PROJET PROJET', 'PROJET', 'PASSATION DES MARCHES DE TRAVAUX', 
-           'ETUDE D EXECUTION', 'TRAVAUX' , 'RECOLEMENT', 'MAINTIEN EN CONDITIONS OPERATIONNELLES']
-
-##### Modeles
-    
-class SiteTelecom(QgsModel):
-    layername = "SITE_TELECOM"
-    geom_type = QgsModel.GEOM_POINT
-    crs = CRS
-    bounding_box = (XMIN,YMIN,XMAX,YMAX)
-    pk = "ST_CODE"
-    schema = {'ST_CODE': {'type': 'string', 'empty': False}, 
-            'ST_NOM': {'type': 'string', 'empty': False}, 
-            'ST_TYPFCT': {'type': 'string', 'empty': False, 'allowed': ['SITE HEBERGEMENT', 
-                                                                        'NOEUD RACCORDEMENT OPTIQUE', 
-                                                                        'SOUS-REPARTITEUR OPTIQUE', 
-                                                                        'SOUS-REPARTITEUR OPTIQUE COLOCALISE', 
-                                                                        'NOEUD RACCORDEMENT D ABONNES', 
-                                                                        'NOEUD RACCORDEMENT D ABONNES - HAUT DEBIT', 
-                                                                        'NOEUD RACCORDEMENT D ABONNES - MONTEE EN DEBIT']}, 
-            'ST_STATUT': {'type': 'string', 'empty': False, 'allowed': STATUTS}, 
-            'ST_NBPRISE': {'empty': False, 'validator': is_positive_int}
-            }
-
-class SiteClient(QgsModel):
-    layername = "SITE_CLIENT"
-    geom_type = QgsModel.GEOM_POINT
-    crs = CRS
-    bounding_box = (XMIN,YMIN,XMAX,YMAX)
-    schema = {
-              'SC_TYPFON': {'type': 'string', 'empty': False, 'allowed': ['RESIDENTIEL', 'PROFESSIONNEL', 'OPERATEUR', 'TECHNIQUE']},
-              'SC_STATUT': {'type': 'string', 'empty': False, 'allowed': STATUTS}, 
-              'SC_NBPRISE': {'empty': False, 'validator': is_positive_int}, 
-              'SC_NBPRHAB': {'empty': False, 'validator': is_positive_int}, 
-              'SC_NBPRPRO': {'empty': False, 'validator': is_positive_int}, 
-              'SC_ID_SITE': {'type': 'string', 'empty': False}, 
-              'SC_NUM_RUE': {'type': 'string', 'empty': False}, 
-              'SC_REPET': {'type': 'string', 'empty': True}, 
-              'SC_DNUBAT': {'type': 'string', 'empty': False}, 
-              'SC_DESC': {'type': 'string', 'empty': False}, 
-              'SC_NOM_RUE': {'type': 'string', 'empty': False}, 
-              'SC_NOM_COM': {'type': 'string', 'empty': False}, 
-              'SC_ETAT': {'type': 'string', 'empty': False, 'allowed': ['VALIDE', 'OBSOLETE', 'NOUVEAU', 'FUTUR']},
-              'SC_ATT_PTO': {'empty': False, 'validator': is_positive_float}
-              }
-
-class Cable(QgsModel):
-    layername = "CABLE"
-    geom_type = QgsModel.GEOM_LINE
-    crs = CRS
-    bounding_box = (XMIN,YMIN,XMAX,YMAX)
-    schema = {'CA_TYPFON': {'type': 'string', 'empty': False, 'allowed': ['COLLECTE TRANSPORT DISTRIBUTION', 'COLLECTE', 'COLLECTE TRANSPORT', 'COLLECTE DISTRIBUTION', 'TRANSPORT DISTRIBUTION', 'TRANSPORT', 'DISTRIBUTION', 'RACCORDEMENT FINAL', 'BOUCLE METROPOLITAINE', 'LONGUE DISTANCE (LONG HAUL)', 'NON COMMUNIQUE']},
-              'CA_TYPSTR': {'type': 'string', 'empty': False, 'allowed': ['CONDUITE', 'AERIEN', 'COLONNE MONTANTE', 'IMMERGE', 'FACADE']},
-              'CA_STATUT': {'type': 'string', 'empty': False, 'allowed': STATUTS},
-              'CA_CAPFO': {'empty': False, 'allowed': [720,576,288,144,96,72,48,24,12]},
-              'CA_MODULO': {'empty': False, 'allowed': [6, 12]},
-              'CA_SUPPORT': {'type': 'string', 'empty': False, 'allowed': ['0', '1']}
-              }
-    
-class Zapbo(QgsModel):
-    layername = "ZAPBO"
-    geom_type = QgsModel.GEOM_POLYGON
-    crs = CRS
-    bounding_box = (XMIN,YMIN,XMAX,YMAX)
-    schema = {'ZP_NBPRISE': {'empty': False, 'validator': is_positive_int},
-              'ZP_ISOLE': {'type': 'string', 'empty': False, 'allowed': ['0', '1']},
-              'ZP_STATUT': {'type': 'string', 'empty': False, 'allowed': STATUTS}
-              }
-    
-class Zasro(QgsModel):
-    layername = "ZASRO"
-    geom_type = QgsModel.GEOM_POLYGON
-    crs = CRS
-    bounding_box = (XMIN,YMIN,XMAX,YMAX)
-    schema = {'ZS_CODE': {'type': 'string', 'empty': False},
-              'ZS_NBPRISE': {'empty': False, 'validator': is_positive_int}
-              }
-    
-class Adduction(QgsModel):
-    layername = "ADDUCTION"
-    geom_type = QgsModel.GEOM_LINE
-    crs = CRS
-    bounding_box = (XMIN,YMIN,XMAX,YMAX)
-    schema = {'AD_ISOLE': {'type': 'string', 'empty': False, 'allowed': ['0', '1']},
-              'AD_LONG': {'empty': False, 'validator': is_positive_float}
-              }
-    
-models = [SiteTelecom, SiteClient, Cable, Zapbo, Zasro, Adduction]
-    
-####### Validateur
-
-
-class Mn3ApdChecker(BaseChecker):
-            
-    def setUp(self):
-        
-        self.dataset = {}
-        for model in models:
-            model.layer = next((l for l in QgsProject.instance().mapLayers().values() \
-                          if l.name().lower() == model.layername.lower()), None)
-            self.dataset[model] = [model(f) for f in model.layer.getFeatures()] if model.layer else []
-        
-        self.sites_telecom = self.dataset.get(SiteTelecom, [])
-        self.sites_client = self.dataset.get(SiteClient, [])
-        self.cables = self.dataset.get(Cable, [])
-        self.zapbos = self.dataset.get(Zapbo, [])
-        self.zasros = self.dataset.get(Zasro, [])
-        self.adductions = self.dataset.get(Adduction, [])
-        
-    def test_load_layers(self):
-        """ Chargement des données
-            Contrôle la présence des couches attendues
-        """
-        for model in models:
-            if model.layer is None:
-                self.log_error("Couche manquante", model=model)
-                continue
-            
-            if model.pk:
-                if not model.pk.lower() in [f.name().lower() for f in model.layer.fields()]:
-                    self.log_error(f"Clef primaire manquante ({model.pk})", model=model)
-                    continue
-        
-    def test_scr(self):
-        """ Contrôle des projections 
-        Vérifie que les couches ont le bon sytème de projection
-        """
-        for model in models:
-            if model.layer and model.layer.crs().authid() != model.crs:
-                self.log_error(f"Mauvaise projection (attendu: {model.crs})", model=model)
-
-    def _validate_structure(self, model, items):
-        v = CerberusValidator(model.schema, purge_unknown=True, error_handler=CerberusErrorHandler, require_all=True)
-        
-        for item in items:
-            v.validate(item.__dict__)
-            
-            for field, verrors in v.errors.items():
-                for err in verrors:
-                    self.log_error(f"{field} : {_translate_messages(err)}", item=item)
-
-    def test_structure_sites_telecom(self):
-        """ Structure des données: SiteTelecom 
-        Contrôle les données attributaires de la couche SITE_TELECOM: 
-        présence, format, valeurs autorisées...
-        """       
-        self._validate_structure(SiteTelecom, self.sites_telecom)
-
-    def test_structure_sites_client(self):
-        """ Structure des données: SiteClient 
-        Contrôle les données attributaires de la couche SITE_CLIENT: 
-        présence, format, valeurs autorisées...
-        """
-        self._validate_structure(SiteClient, self.sites_client)
-
-    def test_structure_cables(self):
-        """ Structure des données: Cables 
-        Contrôle les données attributaires de la couche CABLE: 
-        présence, format, valeurs autorisées...
-        """
-        self._validate_structure(Cable, self.cables)
-        
-    def test_structure_zapbos(self):
-        """ Structure des données: Zapbo 
-        Contrôle les données attributaires de la couche ZAPBO: 
-        présence, format, valeurs autorisées...
-        """
-        self._validate_structure(Zapbo, self.zapbos)
-
-    def test_structure_zasros(self):
-        """ Structure des données: Zasro 
-        Contrôle les données attributaires de la couche ZASRO: 
-        présence, format, valeurs autorisées...
-        """
-        self._validate_structure(Zasro, self.zasros)
-
-    def test_structure_adductions(self):
-        """ Structure des données: Adduction 
-        Contrôle les données attributaires de la couche ADDUCTION: 
-        présence, format, valeurs autorisées...
-        """
-        self._validate_structure(Adduction, self.adductions)
-
-    def test_geometry_validity(self):
-        """ Contrôle de la validité des géométries 
-        """
-        for model in models:
-            for item in self.dataset[model]:
-                if not item.is_geometry_valid():
-                    self.log_error("La géométrie de l'objet est invalide", item=item)
-
-    def test_geometry_type(self):
-        """ Contrôle des types de géométries 
-        """
-        for model in models:
-            for item in self.dataset[model]:
-                item_geom_type = item.get_geom_type()
-                if item_geom_type != model.geom_type:
-                    expected, found = model.get_geom_name(model.geom_type), model.get_geom_name(item_geom_type)
-                    self.log_error(f"Type de géométrie invalide (attendu: {expected}, trouvé: {found})", item=item)
-                    
-    def test_bounding_box(self):
-        """ Contrôle des emprises 
-        Vérifie que les objets sont dans le périmètre attendu
-        """
-        
-        for model in models:
-            xmin, ymin, xmax, ymax = model.bounding_box
-            
-            for item in self.dataset[model]:
-                if not item.is_geometry_valid():
-                    continue
-                
-                x1, y1, x2, y2 = item.get_bounding_box()
-                
-                if any(x < xmin or x > xmax for x in (x1, x2)) or \
-                   any(y < ymin or y > ymax for y in (y1, y2)):
-                    self.log_error("Hors de l'emprise autorisée", item=item)
-        
-    def test_duplicates(self):
-        """ Recherche de doublons 
-        Recherche d'éventuels doublons dans des champs qui supposent l'unicité
-        """
-        # doublons dans ST_CODE
-        tmp = []
-        for site_telecom in self.sites_telecom:
-            if not site_telecom.ST_CODE:
-                continue
-            if not site_telecom.ST_CODE in tmp:
-                tmp.append(site_telecom.ST_CODE)
-            else:
-                self.log_error("Doublons dans le champs ST_CODE", item=site_telecom)
-        
-    def test_dimension_zasro(self):
-        """ Contrôle le dimensionnement des ZASRO """
-        for zasro in self.zasros:
-            if zasro.ZS_NBPRISE > 850:
-                self.log_error("Le nombre de prises est supérieur à 850", item=zasro)
-        
-    def test_affaiblissement(self):
-        """ Contrôle l'affaiblissement """
-        for site_client in self.sites_client:
-            if site_client.SC_ATT_PTO > 28:
-                self.log_error("L'affaiblissement est supérieur à 28", item=site_client)
-
-    def test_capacite_modulo(self):
-        """ Cohérence capacité du cable / modulo """
-        for cable in self.cables:
-            if cable.CA_CAPFO in [720,576,288,144,96] and cable.CA_MODULO != 12:
-                self.log_error(f"Modulo invalide (capacite: {cable.CA_CAPFO}, modulo: {cable.CA_MODULO})", item=cable)
-            elif cable.CA_CAPFO in [48,24,12] and cable.CA_MODULO != 6:
-                self.log_error(f"Modulo invalide (capacite: {cable.CA_CAPFO}, modulo: {cable.CA_MODULO})", item=cable)
-        
-    def test_longueur_racco(self):
-        """ Contrôle la longueur des raccordements """
-        for adduction in self.adductions:
-            if adduction.AD_ISOLE == "1" and adduction.AD_LONG <= 100:
-                self.log_error("L'adduction ne devrait pas être consiudérée comme isolée (long.: {adduction.AD_LONG})", item=adduction)
-        
-            elif adduction.AD_ISOLE == "0" and adduction.AD_LONG > 100:
-                self.log_error("L'adduction devrait être considérée comme isolée (long.: {adduction.AD_LONG})", item=adduction)
-        
-    def test_zapbos_prises(self):
-        """ Topologie: Zapbos / Prises 
-        Toutes les zapbo contiennent au moins une prise"""
-        
-        for zapbo in self.zapbos:
-            nb_prises = sum([site_client.SC_NBPRISE for site_client in self.sites_client if zapbo.geom.contains(site_client.geom)])
-            if not nb_prises:
-                self.log_error("La Zapbo ne contient aucune prise", item=zapbo)
-        
-        for site_client in self.sites_client:
-            nb_zapbo = len([zapbo for zapbo in self.zapbos if zapbo.geom.contains(site_client.geom)])
-            if nb_zapbo == 0:
-                self.log_error("La prise n'est contenue dans aucune ZAPBO", item=site_client)
-            elif nb_zapbo > 1:
-                self.log_error("La prise est contenue dans plus d'une ZAPBO", item=site_client)
-                
-checkers = [Mn3ApdChecker]