''' Schéma de validation des données MN2 @author: olivier.massot, 2018 ''' import logging from qgis.core import QgsProject from core.cerberus_ import is_float, is_multi_int, is_int, \ is_modern_french_date, CerberusValidator, CerberusErrorHandler, \ _translate_messages, is_positive_int, is_strictly_positive_float, \ is_positive_float, is_strictly_positive_int from core.checking import BaseChecker, CheckingException from core.mncheck import QgsModel logger = logging.getLogger("mncheck") SCHEMA_NAME = "Schéma MN v2 REC" STATUTS = ['EN ETUDE', 'EN REALISATION', 'EN SERVICE', 'HORS SERVICE', 'En étude', 'En réalisation', 'En service', 'Hors service'] XMIN, XMAX, YMIN, YMAX = 1341999, 1429750, 8147750, 8294000 CRS = 'EPSG:3949' # Coordinate Reference System TOLERANCE = 0.5 ##### Modeles class Artere(QgsModel): layername = "artere_geo" geom_type = QgsModel.GEOM_LINE crs = CRS bounding_box = (XMIN,YMIN,XMAX,YMAX) pk = "AR_CODE" schema = {'AR_CODE': {'type': 'string', 'empty': False, 'maxlength': 30}, 'AR_NOM': {'type': 'string', 'empty': False, 'maxlength': 30}, 'AR_ID_INSE': {'type': 'string', 'empty': False, 'regex': r'50\d{3}'}, 'AR_LONG': {'empty': False, 'validator': is_float}, 'AR_ETAT': {'type': 'string', 'empty': False, 'allowed': ['0', '1', '2', '3', '4']}, 'AR_OCCP': {'type': 'string', 'empty': False, 'allowed': ['0', '1.1', '1.2', '2', '3', '4']}, 'AR_NOEUD_A': {'type': 'string', 'empty': False, 'maxlength': 30}, 'AR_NOEUD_B': {'type': 'string', 'empty': False, 'maxlength': 30}, 'AR_NB_FOUR': {'empty': True, 'validator': is_multi_int}, 'AR_FOU_DIS': {'empty': True, 'validator': is_int}, 'AR_TYPE_FO': {'type': 'string', 'empty': False, 'multiallowed': ['PVC', 'PEHD', 'SOUS-TUBAGE PEHD', 'SOUS-TUBAGE SOUPLE', 'FACADE', 'AERIEN', 'ENCORBELLEMENT', 'AUTRE']}, 'AR_TYFO_AI': {'type': 'string', 'empty': False, 'multiallowed': ['PVC', 'PEH', 'TUB', 'FAC', 'ENC', 'APP']}, 'AR_DIAM_FO': {'type': 'string', 'multiallowed': ['10', '14', '18', '25', '28', '32', '40', '45', '60', '80', '150', 'NUL', '']}, 'AR_PRO_FOU': {'type': 'string', 'empty': False, 'multiallowed': ['MANCHE NUMERIQUE', 'COLLECTIVITE', 'ORANGE', 'PRIVE', 'ERDF', 'SDEM', 'AUTRE']}, 'AR_FAB': {'type': 'string', 'empty': True, 'maxlength': 100}, 'AR_REFFAB': {'type': 'string', 'empty': True, 'maxlength': 100}, 'AR_COULEUR': {'type': 'string', 'empty': True, 'maxlength': 20}, 'AR_AIGUIL': {'type': 'string', 'empty': True, 'maxlength': 3, 'allowed': ['OUI', 'NON']}, 'AR_NBCABL': {'empty': False, 'validator': is_positive_int}, 'AR_PRO_CAB': {'type': 'string', 'empty': False, 'allowed': ['MANCHE NUMERIQUE']}, 'AR_GEST_FO': {'type': 'string', 'multiallowed': ['MANCHE NUMERIQUE', 'MANCHE TELECOM', 'COLLECTIVITE', 'ORANGE', 'MANCHE FIBRE', 'PRIVE', 'ERDF', 'SDEM', 'AUTRE', 'NUL']}, 'AR_UTIL_FO': {'type': 'string', 'multiallowed': ['MANCHE NUMERIQUE', 'MANCHE TELECOM', 'COLLECTIVITE', 'ORANGE', 'MANCHE FIBRE', 'PRIVE', 'AUTRE', 'SDEM', 'NUL', '']}, 'AR_DATE_IN': {'empty': True, 'validator': is_modern_french_date}, 'AR_DATE_RE': {'empty': True, 'validator': is_modern_french_date}, 'AR_REF_PLA': {'type': 'string', 'maxlength': 100}, 'AR_SRC_GEO': {'type': 'string', 'empty': True, 'maxlength': 50}, 'AR_QLT_GEO': {'type': 'string', 'empty': False, 'allowed': ['A', 'B', 'C']}, 'AR_PRO_MD': {'type': 'string', 'empty': False, 'default': 'MANCHE NUMERIQUE', 'allowed': ['MANCHE NUMERIQUE']}, 'AR_COMMENT': {'type': 'string', 'maxlength': 300, 'empty': True}, 'AR_STATUT': {'type': 'string', 'empty': False, 'allowed': STATUTS}} class Cable(QgsModel): layername = "cable_geo" geom_type = QgsModel.GEOM_LINE crs = CRS bounding_box = (XMIN,YMIN,XMAX,YMAX) pk = "CA_CODE" schema = {'CA_CODE': {'type': 'string', 'empty': False, 'maxlength': 30}, 'CA_NOM': {'type': 'string', 'empty': False, 'maxlength': 30}, 'CA_NUMERO': {'type': 'string', 'empty': False, 'maxlength': 30}, 'CA_EMPRISE': {'type': 'string', 'empty': False, 'maxlength': 10}, 'CA_FAB': {'type': 'string', 'empty': False, 'maxlength': 100}, 'CA_REFFAB': {'type': 'string', 'empty': False, 'maxlength': 100}, 'CA_MODELE': {'type': 'string', 'empty': False, 'maxlength': 10}, 'CA_TYPE': {'type': 'string', 'maxlength': 10, 'empty': False, 'allowed': ['AERIEN', 'IMMEUBLE', 'FACADE', 'MIXTE', 'SOUTERRAIN']}, 'CA_TYPFCT': {'type': 'string', 'maxlength': 3, 'empty': False, 'allowed': ['CDI', 'CTR', 'CBM', 'RAC', 'CBO']}, 'CA_ETAT': {'type': 'string', 'maxlength': 1, 'empty': False, 'allowed': ['0', '1', '2', '3', '4']}, 'CA_LONG': {'empty': False, 'validator': is_strictly_positive_float}, 'CA_EQ_A': {'type': 'string', 'empty': False, 'maxlength': 30}, 'CA_EQ_B': {'type': 'string', 'empty': False, 'maxlength': 30}, 'CA_DIAMETR': {'empty': False, 'validator': is_positive_float}, 'CA_COULEUR': {'type': 'string', 'maxlength': 20, 'empty': False, 'allowed': ['NOIR', 'BLEU', 'BLANC']}, 'CA_TECHNOL': {'type': 'string', 'maxlength': 17, 'empty': False, 'allowed': ['G657A2_M6', 'G657A2_M12', 'G652D_M12']}, 'CA_NB_FO_U': {'empty': False, 'validator': is_positive_int}, 'CA_NB_FO_D': {'empty': False, 'validator': is_positive_int}, 'CA_PRO': {'type': 'string', 'maxlength': 20, 'empty': False, 'allowed': ['MANCHE NUMERIQUE']}, 'CA_GEST': {'type': 'string', 'maxlength': 20, 'empty': False, 'allowed': ['MANCHE FIBRE']}, 'CA_DATE_IN': {'empty': True, 'validator': is_modern_french_date}, 'CA_COMMENT': {'type': 'string', 'maxlength': 300, 'empty': True}, 'CA_STATUT': {'type': 'string', 'empty': False, 'allowed': STATUTS}} class Equipement(QgsModel): layername = "equipement_passif" geom_type = QgsModel.GEOM_POINT crs = CRS bounding_box = (XMIN,YMIN,XMAX,YMAX) pk = "EQ_CODE" schema = {'EQ_CODE': {'type': 'string', 'empty': False, 'maxlength': 30}, 'EQ_NOM': {'type': 'string', 'empty': False, 'maxlength': 30}, 'EQ_NOM_NOE': {'type': 'string', 'empty': False, 'maxlength': 30}, 'EQ_REF': {'type': 'string', 'empty': False, 'maxlength': 100}, 'EQ_EMPRISE': {'type': 'string', 'empty': False, 'maxlength': 10}, 'EQ_FABR': {'type': 'string', 'empty': False, 'maxlength': 100}, 'EQ_CAPFO': {'empty': True, 'validator': is_strictly_positive_int}, 'EQ_NBMXEQ': {'empty': True, 'validator': is_strictly_positive_int}, 'EQ_NBCAB': {'empty': True, 'validator': is_strictly_positive_int}, 'EQ_DIMENS': {'type': 'string', 'empty': True, 'maxlength': 50}, 'EQ_TYPEQ': {'type': 'string', 'empty': True, 'maxlength': 100}, 'EQ_ETAT': {'type': 'string', 'empty': False, 'allowed': ['0', '1', '2', '3', '4']}, 'EQ_OCCP': {'type': 'string', 'empty': False, 'allowed': ['0', '1.1', '1.2', '2', '3', '4']}, 'EQ_TYPE': {'type': 'string', 'empty': False, 'allowed': ['PBO', 'BPE', 'BAI', 'PEC']}, 'EQ_TYPSTRC': {'type': 'string', 'empty': True, 'allowed': ['CHAMBRE', 'AERIEN', 'FACADE', 'COLONNE MONTANTE', 'PIED IMMEUBLE', 'DTIO']}, 'EQ_TYPE_LQ': {'type': 'string', 'empty': False, 'allowed': ['PBO', 'BPE JB', 'BPE JD', 'BAIDC', 'BAIOP']}, 'EQ_TYPE_PH': {'type': 'string', 'empty': False, 'allowed': ['PBO 6', 'PBO 12', 'BPE 12EP', 'BPE 24EP', 'BPE 48EP', 'BPE 72EP', 'BPE 96EP', 'BPE 144EP', 'BPE 288EP', 'BPE 576EP', 'BPE 720EP', 'BAI']}, 'EQ_PRO': {'type': 'string', 'empty': False, 'allowed': ['MANCHE NUMERIQUE', 'COLLECTIVITE', 'ORANGE', 'PRIVE', 'AUTRE', 'NUL']}, 'EQ_GEST': {'type': 'string', 'empty': False, 'allowed': ['MANCHE NUMERIQUE', 'MANCHE TELECOM', 'COLLECTIVITE', 'ORANGE', 'MANCHE FIBRE', 'PRIVE', 'AUTRE', 'NUL']}, 'EQ_HAUT': {'empty': True, 'validator': is_float}, 'EQ_DATE_IN': {'empty': True, 'validator': is_modern_french_date}, 'EQ_COMMENT': {'type': 'string', 'maxlength': 300, 'empty': True}, 'EQ_STATUT': {'type': 'string', 'empty': False, 'allowed': STATUTS}} class Noeud(QgsModel): layername = "noeud_geo" geom_type = QgsModel.GEOM_POINT crs = CRS bounding_box = (XMIN,YMIN,XMAX,YMAX) pk = "NO_CODE" schema = {'NO_CODE': {'type': 'string', 'empty': False, 'maxlength': 30}, 'NO_NOM': {'type': 'string', 'empty': False, 'maxlength': 30}, 'NO_ID_INSE': {'type': 'string', 'empty': False, 'regex': r'50\d{3}'}, 'NO_VOIE': {'type': 'string', 'empty': False, 'maxlength': 100}, 'NO_EMPRISE': {'type': 'string', 'empty': False, 'maxlength':10}, 'NO_ETAT': {'type': 'string', 'empty': False, 'allowed': ['0', '1', '2', '3', '4']}, 'NO_OCCP': {'type': 'string', 'empty': False, 'allowed': ['0', '1.1', '1.2', '2', '3', '4']}, 'NO_TYPE': {'type': 'string', 'empty': False, 'allowed': ['CHA', 'POT', 'LTE', 'SEM', 'FAC', 'OUV', 'IMM']}, 'NO_TYPFCT': {'type': 'string', 'empty': False, 'allowed': ['INTERCONNEXION', 'SATELLITE', 'PASSAGE', 'REGARD', 'INDETERMINE']}, 'NO_TYPE_LQ': {'type': 'string', 'empty': False, 'allowed': ['CHTIR', 'CHRACC', 'POT', 'NRO', 'PM', 'MIMO', 'FAC', 'OUV', 'IMM']}, 'NO_TYPE_PH': {'type': 'string', 'empty': False, 'allowed': ['CHAMBRE', 'POTEAU', 'ARMOIRE', 'SHELTER', 'BATIMENT', 'SITE MIMO', 'FACADE', 'OUVRAGE', 'IMMEUBLE']}, 'NO_CODE_PH': {'type': 'string', 'maxlength': 20}, 'NO_TECH_PS': {'type': 'string', 'maxlength': 20, 'multiallowed': ['COAX', 'CUT', 'ECL', 'ELEC', 'VP', 'OPT', 'NC']}, 'NO_AMO': {'type': 'string', 'maxlength': 30}, 'NO_PLINOX': {'required':False, 'type': 'string', 'maxlength': 3, 'allowed': ['OUI', 'NON']}, 'NO_X': {'empty': True, 'validator': is_float}, 'NO_Y': {'empty': True, 'validator': is_float}, 'NO_PRO': {'type': 'string', 'maxlength': 20, 'empty': False, 'allowed': ['MANCHE NUMERIQUE', 'COLLECTIVITE', 'ORANGE', 'ERDF', 'PRIVE', 'ENEDIS', 'SDEM', 'AUTRE', 'NUL']}, 'NO_GEST': {'type': 'string', 'maxlength': 20, 'empty': False, 'allowed': ['MANCHE NUMERIQUE', 'MANCHE TELECOM', 'COLLECTIVITE', 'ORANGE', 'ERDF', 'ENEDIS', 'SDEM', 'MANCHE FIBRE', 'PRIVE', 'AUTRE', 'NUL']}, 'NO_HAUT': {'empty': True, 'validator': is_float}, 'NO_DATE_IN': {'empty': True, 'validator': is_modern_french_date}, 'NO_REF_PLA': {'type': 'string', 'maxlength': 100}, 'NO_SRC_GEO': {'type': 'string', 'empty': True, 'maxlength': 50}, 'NO_QLT_GEO': {'type': 'string', 'maxlength': 1, 'empty': False, 'allowed': ['A', 'B', 'C']}, 'NO_PRO_MD': {'type': 'string', 'maxlength': 20, 'empty': False, 'allowed': ['MANCHE NUMERIQUE']}, 'NO_COMMENT': {'type': 'string', 'maxlength': 300, 'empty': True}, 'NO_STATUT': {'type': 'string', 'empty': False, 'allowed': STATUTS}} class Tranchee(QgsModel): layername = "tranchee_geo" geom_type = QgsModel.GEOM_LINE crs = CRS bounding_box = (XMIN,YMIN,XMAX,YMAX) pk = "TR_CODE" schema = {'TR_CODE': {'type': 'string', 'empty': False, 'maxlength': 30}, 'TR_NOM': {'type': 'string', 'empty': False, 'maxlength': 30}, 'TR_ID_INSE': {'type': 'string', 'empty': False, 'regex': r'50\d{3}'}, 'TR_VOIE': {'type': 'string', 'empty': False, 'maxlength': 200}, 'TR_TYP_IMP': {'type': 'string', 'empty': False, 'allowed': ['ACCOTEMENT STABILISE', 'ACCOTEMENT NON STABILISE', 'CHAUSSEE LOURDE', 'CHAUSSEE LEGERE', 'FOSSE', 'TROTTOIR', 'ESPACE VERT', 'ENCORBELLEMENT']}, 'TR_MOD_POS': {'type': 'string', 'empty': False, 'allowed': ['TRADITIONNEL', 'MICRO TRANCHEE', 'FONCAGE 60', 'FONCAGE 90', 'FONCAGE 120', 'TRANCHEUSE', 'FORAGE URBAIN', 'FORAGE RURAL', 'ENCORBELLEMENT']}, 'TR_MPOS_AI': {'type': 'string', 'empty': False, 'allowed': ['TRA', 'ALL', 'FONCAGE 60', 'FON', 'FOR', 'ENC']}, 'TR_LONG': {'empty': True, 'validator': is_strictly_positive_float}, 'TR_LARG': {'empty': True, 'validator': is_strictly_positive_float}, 'TR_REVET': {'empty':True, 'type': 'string', 'allowed': ['SABLE', 'BICOUCHE', 'GRAVIER', 'ENROBE', 'BETON', 'PAVE', 'TERRAIN NATUREL']}, 'TR_CHARGE': {'empty': True, 'validator': is_positive_float}, 'TR_GRILLAG': {'empty':True, 'validator': is_positive_float}, 'TR_REMBLAI': {'type': 'string', 'maxlength': 50}, 'TR_PLYNOX': {'type': 'string', 'empty': False, 'allowed': ['OUI', 'NON']}, 'TR_PRO_VOI': {'type': 'string', 'empty': False, 'allowed': ['COMMUNE', 'COMMUNAUTE DE COMMUNES', 'DEPARTEMENT', 'ETAT', 'PRIVE']}, 'TR_GEST_VO': {'type': 'string', 'empty': False, 'allowed': ['COMMUNE', 'COMMUNAUTE DE COMMUNES', 'DEPARTEMENT', 'ETAT', 'PRIVE']}, 'TR_SCHEMA': {'maxlength': 100, 'type': 'string'}, 'TR_DATE_IN': {'empty': True, 'validator': is_modern_french_date}, 'TR_SRC_GEO': {'type': 'string', 'empty': True, 'maxlength': 50}, 'TR_QLT_GEO': {'type': 'string', 'empty': False, 'allowed': ['A', 'B', 'C']}, 'TR_PRO_MD': {'type': 'string', 'empty': False, 'maxlength': 20, 'allowed': ['MANCHE NUMERIQUE']}, 'TR_COMMENT': {'type': 'string', 'maxlength': 300, 'empty': True}, 'TR_STATUT': {'type': 'string', 'empty': False, 'allowed': STATUTS}} class Zapbo(QgsModel): layername = "zapbo_geo" geom_type = QgsModel.GEOM_POLYGON crs = CRS bounding_box = (XMIN,YMIN,XMAX,YMAX) pk = "ID_ZAPBO" required = False schema = {'ID_ZAPBO': {'type': 'string', 'empty': False, 'maxlength': 30, 'contains_any_of': ['PBO', 'BPE', 'PEC', 'BAI']}, 'COMMENTAIR': {'type': 'string', 'maxlength': 254, 'empty': True}, 'STATUT': {'type': 'string', 'empty': False, 'allowed': STATUTS}} models = [Artere, Cable, Equipement, Noeud, Tranchee, Zapbo] class Prise(QgsModel): layername = "FTTH_SITES_GEO_VALIDE" layername_bis = "SITE_GEO" ####### Validateur class Mn2Checker(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.arteres = self.dataset.get(Artere, []) self.cables = self.dataset.get(Cable, []) self.equipements = self.dataset.get(Equipement, []) self.noeuds = self.dataset.get(Noeud, []) self.tranchees = self.dataset.get(Tranchee, []) self.zapbos = self.dataset.get(Zapbo, []) for artere in self.arteres: artere.noeud_a = next((n for n in self.noeuds if n.NO_CODE == artere.AR_NOEUD_A), None) artere.noeud_b = next((n for n in self.noeuds if n.NO_CODE == artere.AR_NOEUD_B), None) for cable in self.cables: cable.equipement_a = next((e for e in self.equipements if e.EQ_CODE == cable.CA_EQ_A), None) cable.equipement_b = next((e for e in self.equipements if e.EQ_CODE == cable.CA_EQ_B), None) for equipement in self.equipements: equipement.noeud = next((n for n in self.noeuds if n.NO_CODE == equipement.EQ_NOM_NOE), None) self.prises = [] Prise.layer = next((l for l in QgsProject.instance().mapLayers().values() \ if l.name().lower() in [Prise.layername.lower(), Prise.layername_bis.lower()]), None) 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"] 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_arteres(self): """ Structure des données: Artères Contrôle les données attributaires de la couche ARTERE_GEO: présence, format, valeurs autorisées... """ self._validate_structure(Artere, self.arteres) def test_structure_cables(self): """ Structure des données: Cables Contrôle les données attributaires de la couche CABLE_GEO: présence, format, valeurs autorisées... """ self._validate_structure(Cable, self.cables) def test_structure_equipements(self): """ Structure des données: Equipements Contrôle les données attributaires de la couche EQUIPEMENT_GEO: présence, format, valeurs autorisées... """ self._validate_structure(Equipement, self.equipements) def test_structure_noeuds(self): """ Structure des données: Noeuds Contrôle les données attributaires de la couche NOEUD_GEO: présence, format, valeurs autorisées... """ self._validate_structure(Noeud, self.noeuds) def test_structure_tranchees(self): """ Structure des données: Tranchées Contrôle les données attributaires de la couche TRANCHEE_GEO: présence, format, valeurs autorisées... """ self._validate_structure(Tranchee, self.tranchees) def test_structure_zapbos(self): """ Structure des données: Zapbos Contrôle les données attributaires de la couche ZAPBO_GEO: présence, format, valeurs autorisées... """ self._validate_structure(Zapbo, self.zapbos) 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 AR_CODE tmp = [] for artere in self.arteres: if not artere.AR_CODE: continue if not artere.AR_CODE in tmp: tmp.append(artere.AR_CODE) else: self.log_error("Doublons dans le champs AR_CODE", item=artere) # doublons dans AR_NOM tmp = [] for artere in self.arteres: if not artere.AR_NOM: continue if not artere.AR_NOM in tmp: tmp.append(artere.AR_NOM) else: self.log_error("Doublons dans le champs AR_NOM", item=artere) # doublons dans CA_CODE tmp = [] for cable in self.cables: if not cable.CA_CODE: continue if not cable.CA_CODE in tmp: tmp.append(cable.CA_CODE) else: self.log_error("Doublons dans le champs CA_CODE", item=cable) # doublons dans CA_NOM tmp = [] for cable in self.cables: if not cable.CA_NOM: continue if not cable.CA_NOM in tmp: tmp.append(cable.CA_NOM) else: self.log_error("Doublons dans le champs CA_NOM", item=cable) # doublons dans EQ_CODE tmp = [] for equipement in self.equipements: if not equipement.EQ_CODE: continue if not equipement.EQ_CODE in tmp: tmp.append(equipement.EQ_CODE) else: self.log_error("Doublons dans le champs EQ_CODE", item=equipement) # doublons dans EQ_NOM tmp = [] for equipement in self.equipements: if not equipement.EQ_NOM: continue if not equipement.EQ_NOM in tmp: tmp.append(equipement.EQ_NOM) else: self.log_error("Doublons dans le champs EQ_NOM", item=equipement) # doublons dans NO_CODE tmp = [] for noeud in self.noeuds: if not noeud.NO_CODE: continue if not noeud.NO_CODE in tmp: tmp.append(noeud.NO_CODE) else: self.log_error("Doublons dans le champs NO_CODE", item=noeud) # doublons dans NO_NOM tmp = [] for noeud in self.noeuds: if not noeud.NO_NOM: continue if not noeud.NO_NOM in tmp: tmp.append(noeud.NO_NOM) else: self.log_error("Doublons dans le champs NO_NOM", item=noeud) # doublons dans TR_CODE tmp = [] for tranchee in self.tranchees: if not tranchee.TR_CODE: continue if not tranchee.TR_CODE in tmp: tmp.append(tranchee.TR_CODE) else: self.log_error("Doublons dans le champs TR_CODE", item=tranchee) # doublons dans TR_NOM tmp = [] for tranchee in self.tranchees: if not tranchee.TR_NOM: continue if not tranchee.TR_NOM in tmp: tmp.append(tranchee.TR_NOM) else: self.log_error("Doublons dans le champs TR_NOM", item=tranchee) # doublons dans ID_ZAPBO tmp = [] for zapbo in self.zapbos: if not zapbo.ID_ZAPBO: continue if not zapbo.ID_ZAPBO in tmp: tmp.append(zapbo.ID_ZAPBO) else: self.log_error("Doublons dans le champs ID_ZAPBO", item=zapbo) def test_constraints_arteres_noeuds(self): """ Application des contraintes: Artères / Noeuds Vérifie que les noeuds attachés aux artères existent """ for artere in self.arteres: if artere.noeud_a is None: self.log_error(f"Le noeud lié '{artere.AR_NOEUD_A}' n'existe pas", item=artere) if artere.noeud_b is None: self.log_error(f"Le noeud lié '{artere.AR_NOEUD_B}' n'existe pas", item=artere) def test_constraints_cables_equipements(self): """ Application des contraintes: Equipements / Cables Vérifie que les équipements attachés aux cables existent """ for cable in self.cables: if cable.equipement_a is None: self.log_error(f"L'équipement lié '{cable.CA_EQ_A}' n'existe pas", item=cable) if cable.equipement_b is None: self.log_error(f"L'équipement lié '{cable.CA_EQ_B}' n'existe pas", item=cable) def test_constraints_cables_equipements_b(self): """ Application des contraintes: Equipements B Vérifie que tous les équipements sont l'équipement B d'au moins un cable """ equipements_b = [cable.CA_EQ_B for cable in self.cables] for equipement in self.equipements: if equipement.EQ_TYPE == "BAI": continue if not equipement.EQ_NOM in equipements_b: self.log_error(f"L'equipement '{equipement.EQ_NOM}' n'est l'équipement B d'aucun cable et n'est pas un PM", item=equipement) def test_constraints_equipements_noeuds(self): """ Application des contraintes: Noeuds / Equipements Vérifie que les noeuds attachés aux équipements existent """ for equipement in self.equipements: if equipement.noeud is None: self.log_error(f"Le noeud lié '{equipement.EQ_NOM_NOE}' n'existe pas", item=equipement) def test_graphic_duplicates(self): """ Recherche de doublons graphiques """ for i, tranchee in enumerate(self.tranchees): for other in self.tranchees[i+1:]: if tranchee.geom.equals(other.geom): self.log_error("Une entité graphique est dupliquée", item=tranchee) continue for i, artere in enumerate(self.arteres): for other in self.arteres[i+1:]: if artere.geom.equals(other.geom): self.log_error("Une entité graphique est dupliquée", item=artere) continue for i, cable in enumerate(self.cables): for other in self.cables[i+1:]: if cable.geom.equals(other.geom) and cable.CA_EQ_A == other.CA_EQ_A and cable.CA_EQ_B == other.CA_EQ_B: self.log_error("Une entité graphique est dupliquée", item=cable) continue for i, noeud in enumerate(self.noeuds): for other in self.noeuds[i+1:]: if noeud.geom.equals(other.geom): self.log_error("Une entité graphique est dupliquée", item=noeud) continue for i, zapbo in enumerate(self.zapbos): for other in self.zapbos[i+1:]: if zapbo.geom.equals(other.geom): self.log_error("Une entité graphique est dupliquée", item=zapbo) continue def test_positions_noeuds(self): """ Topologie: Noeuds / Artères Compare la géométrie des noeuds à celle des artères """ for artere in self.arteres: if not artere.noeud_a or not artere.noeud_b: continue artere_points = artere.get_points() noeud_a_point = artere.noeud_a.get_points()[0] noeud_b_point = artere.noeud_b.get_points()[0] if not (artere_points[0].distanceSquared(noeud_a_point) <= TOLERANCE and \ artere_points[-1].distanceSquared(noeud_b_point) <= TOLERANCE) \ and not (artere_points[-1].distanceSquared(noeud_a_point) <= TOLERANCE and \ artere_points[0].distanceSquared(noeud_b_point) <= TOLERANCE): self.log_error(f"Pas de noeud aux coordonnées attendues (noeuds attendus: {artere.noeud_a.NO_CODE} ou {artere.noeud_b.NO_CODE})", item=artere) def test_positions_equipements(self): """ Topologie: Equipements / Cables Compare la géométrie des équipements à celle des cables """ for cable in self.cables: 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 #! attention: on utilise la géométrie du noeud associé à l'équipement, pas celle de l'équipement lui-même cable_points = cable.get_points() equip_a_point = cable.equipement_a.noeud.get_points()[0] equip_b_point = cable.equipement_b.noeud.get_points()[0] if not (cable_points[0].distanceSquared(equip_a_point) <= TOLERANCE and \ cable_points[-1].distanceSquared(equip_b_point) <= TOLERANCE) \ and not (cable_points[-1].distanceSquared(equip_a_point) <= TOLERANCE and \ cable_points[0].distanceSquared(equip_b_point) <= TOLERANCE): self.log_error(f"Pas d'équipement aux coordonnées attendues (equipements attendus: {cable.equipement_a.EQ_CODE} ou {cable.equipement_b.EQ_CODE})", 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 = Artere.full_buffer(TOLERANCE) for tranchee in self.tranchees: if not arteres_full_buffer.contains(tranchee.geom): self.log_error("Tranchée ou portion de tranchée sans artère", item=tranchee) def test_cable_artere(self): """ Topologie: Cables / Artères Compare la géométrie des cables à celle des artères """ # Vérifie que chaque cable a au moins une artère (sauf si commentaire contient 'baguette') arteres_full_buffer = Artere.full_buffer(TOLERANCE) 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) 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) for artere in self.arteres: if any(x in artere.AR_COMMENT.lower() for x in ['racco','client','adductio','attente','bus','sans cable']): continue if not cables_full_buffer.contains(artere.geom): self.log_error("Artère ou portion d'artère sans cable", item=artere) def test_boites(self): """ Données des boites Contrôle les données des équipements de type BPE / PBO """ for equipement in self.equipements: if equipement.EQ_TYPE == "BAI" or not equipement.EQ_DATE_IN: continue for field in ["EQ_CAPFO", "EQ_NBMXEQ", "EQ_NBCAB", "EQ_DIMENS", "EQ_TYPEQ", "EQ_TYPSTRC"]: if not hasattr(equipement, field): self.log_error("Le champs est obligatoire: {}".format(field), item=equipement) elif getattr(equipement, field) == "": self.log_error("Le champs doit être renseigné: {}".format(field), item=equipement) def test_arteres_enterrees(self): """ Données des artères enterrées Contrôle les données des artères enterrées ou en encorbellement """ for artere in self.arteres: if artere.AR_TYPE_FO in ["FACADE", "AERIEN"]: continue for field in ["AR_NB_FOUR", "AR_FOU_DIS", "AR_DIAM_FO", "AR_AIGUIL"]: if not hasattr(artere, field): self.log_error("Le champs est obligatoire: {}".format(field), item=artere) elif getattr(artere, field) == "": self.log_error("Le champs doit être renseigné: {}".format(field), item=artere) if artere.AR_DATE_IN: for field in ["AR_COULEUR", "AR_FAB", "AR_REFFAB"]: if not hasattr(artere, field): self.log_error("Le champs est obligatoire: {}".format(field), item=artere) elif getattr(artere, field) == "": self.log_error("Le champs doit être renseigné: {}".format(field), item=artere) def test_dimensions_fourreaux(self): """ Dimensions logiques: fourreaux Vérifie que les nombres de fourreaux renseignés sont cohérents """ for artere in self.arteres: if artere.AR_TYPE_FO in ["FACADE", "AERIEN"]: continue try: if not int(artere.AR_FOU_DIS) <= int(artere.AR_NB_FOUR): self.log_error("Le nombre de fourreaux disponibles (AR_FOU_DIS) doit être inférieur au nombre total (AR_NB_FOUR)", item=artere) except (TypeError, ValueError): pass def test_dates_install(self): """ Dates d'installation Vérifie que les dates d'installation sont renseignées pour les équipements en service """ for equipement in self.equipements: if equipement.EQ_STATUT == "EN SERVICE" and not equipement.EQ_DATE_IN: self.log_error("Date d'installation (EQ_DATE_IN) manquante", item=equipement) for cable in self.cables: if cable.CA_STATUT == "EN SERVICE" and not cable.CA_DATE_IN: self.log_error("Date d'installation (CA_DATE_IN) manquante", item=cable) for tranchee in self.tranchees: if tranchee.TR_STATUT == "EN SERVICE" and not tranchee.TR_DATE_IN: self.log_error("Date d'installation (TR_DATE_IN) manquante", item=tranchee) def _za_for_pbo(self, pbo): # retourne la ZAPBO correspondant à la PBO en parametre, None si aucune if hasattr(pbo, 'zapbo') and pbo.zapbo: return pbo.zapbo for zapbo in self.zapbos: if zapbo.geom.contains(pbo.geom) and pbo.EQ_NOM in zapbo.ID_ZAPBO: return zapbo return None def test_pbos(self): """ Topologie: PBO / ZAPBO Compare la géométrie et le nom des équipements de type PBO à celle des ZAPBO """ # Verifier que chaque equipement de type PBO est contenu dans une zapbo, et que le nom de la zapbo contient le nom de l'equipement for equipement in self.equipements: if not equipement.EQ_TYPE == "PBO": continue #zapbos englobant l'equipement candidates = [z for z in self.zapbos if z.geom.contains(equipement.noeud.geom)] # le pbo doit être contenu dans une zapbo if not candidates: self.log_error("Le PBO n'est contenu dans aucune ZAPBO", item=equipement) continue # On se base sur le nom pour trouver la zapbo correspondante equipement.zapbo = next((z for z in candidates if equipement.EQ_CODE in z.ID_ZAPBO), None) if not equipement.zapbo: self.log_error("Le nom du PBO ne coincide avec le nom d'aucune des ZAPBO qui le contiennent", item=equipement) def test_zapbos_prises(self): """ Topologie: Zapbos / Prises Toutes les zapbo contiennent au moins une prise""" self._load_prises() if not self.prises: 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): self.log_error("La Zapbo ne contient aucune prise", item=zapbo) def test_pbo_dimension(self): """ Dimensionnement des PBO Compare la couche des prises (si chargée) au dimensionnement des PBO""" self._load_prises() if not self.prises: raise CheckingException("La couche des prises n'est pas chargée") for equipement in self.equipements: if not equipement.EQ_TYPE == "PBO": continue zapbo = self._za_for_pbo(equipement) zapbo.nb_prises = sum([int(prise.NB_PRISE) for prise in self.prises if zapbo.geom.contains(prise.geom) and prise.NB_PRISE]) # Controle du dimensionnement des PBO if equipement.EQ_TYPE_PH == 'PBO 6' and not zapbo.nb_prises < 6: self.log_error("Le PBO 6 contient plus de 5 prises", item=equipement) if equipement.EQ_TYPE_PH == 'PBO 12' and not zapbo.nb_prises >= 6 and zapbo.nb_prises <= 8: self.log_error("Le PBO 12 contient moins de 6 prises ou plus de 8 prises", item=equipement) if zapbo.STATUT == "REC" and not equipement.EQ_STATUT == "REC": self.log_error("Le statut du PBO n'est pas cohérent avec le statut de sa ZAPBO", item=equipement) if equipement.EQ_STATUT == "REC" and not zapbo.STATUT == "REC" and not zapbo.ID_ZAPBO[:4].lower() == "att_": self.log_error("Le statut du PBO n'est pas cohérent avec le statut de sa ZAPBO", item=equipement) checkers = [Mn2Checker]