mncheck/MnCheck/schemas/mn2_rec.py

'''

    Schéma de validation des données MN2

@author: olivier.massot, 2018
'''

import logging
from qgis.core import QgsProject  #@UnresolvedImport

from MnCheck.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, _tofloat
from MnCheck.core.checking import BaseChecker, CheckingException
from MnCheck.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', 'ENEDIS', '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', 'ENEDIS', '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)
                
            if cable.CA_STATUT != cable.equipement_b.EQ_STATUT:
                self.log_error(f"L'équipement B du cable n'a pas le même statut que le cable", 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_CODE in equipements_b:
                self.log_error(f"L'equipement '{equipement.EQ_CODE}' 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]
            
            a_seen_at = []
            b_seen_at = []
            for i, p in enumerate(artere_points):
                if p.distanceSquared(noeud_a_point) <= TOLERANCE:
                    a_seen_at.append(i)
                if p.distanceSquared(noeud_b_point) <= TOLERANCE:
                    b_seen_at.append(i)
            
            if a_seen_at:
                if 0 in a_seen_at or (len(artere_points) - 1) in a_seen_at:
                    pass # ok
                else:
                    self.log_error(f"Le noeud A n'est pas situé à l'extremité de l'artère", item=artere)
            else:
                self.log_error(f"Le noeud A n'est pas aux coordonnées attendues ({artere.noeud_a.NO_CODE})", item=artere)
            
            if b_seen_at:
                if 0 in b_seen_at or (len(artere_points) - 1) in b_seen_at:
                    pass # ok
                else:
                    self.log_error(f"Le noeud B n'est pas situé à l'extremité de l'artère", item=artere)
            else:
                self.log_error(f"Le noeud B n'est pas aux coordonnées attendues ({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]
            
            a_seen_at = []
            b_seen_at = []
            for i, p in enumerate(cable_points):
                if p.distanceSquared(equip_a_point) <= TOLERANCE:
                    a_seen_at.append(i)
                if p.distanceSquared(equip_b_point) <= TOLERANCE:
                    b_seen_at.append(i)
            
            if a_seen_at:
                if 0 in a_seen_at or (len(cable_points) - 1) in a_seen_at:
                    pass # ok
                else:
                    self.log_error(f"L'équipement A n'est pas situé à l'extremité du cable", item=cable)
            else:
                self.log_error(f"L'équipement A n'est pas aux coordonnées attendues ({cable.equipement_a.EQ_CODE})", item=cable)
            
            if b_seen_at:
                if 0 in b_seen_at or (len(cable_points) - 1) in b_seen_at:
                    pass # ok
                else:
                    self.log_error(f"L'équipement B n'est pas situé à l'extremité du cable", item=cable)
            else:
                self.log_error(f"L'équipement B n'est pas aux coordonnées attendues ({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)
    
            if equipement.EQ_TYPE == "BPE" and equipement.noeud.NO_TYPE_PH == "POTEAU":
                self.log_error("Une BPE ne peut pas être sur un poteau", item=equipement)
                
            if equipement.EQ_TYPE == "PBO" and equipement.noeud.NO_TYPE_PH == "POTEAU":
                try:
                    if not 2.4 <= _tofloat(equipement.EQ_HAUT) <= 4.0:
                        self.log_error("PBO sur poteau: La hauteur doit être comprise entre 2.40m et 4.00m", item=equipement)
                except (TypeError, ValueError):
                    self.log_error("PBO sur poteau: Hauteur manquante ou invalide", 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_largeur_tranchees(self):
        """ Tranchées: Dimensions
        Vérifie que la cohérence des dimensions des tranchées """
        for tranchee in self.tranchees:
            if not ("FONCAGE" in tranchee.TR_MOD_POS or 
                    "FORAGE" in tranchee.TR_MOD_POS or 
                    "ENCORBELLEMENT" in tranchee.TR_MOD_POS):
                try:
                    if float(tranchee.TR_LARG) > 0:
                        continue
                except (TypeError, ValueError):
                    pass
                self.log_error("La largeur de la tranchée doit être supérieure à 0", item=tranchee)
                    
    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_prop_gest(self):
        """ Propriétaires / Gestionnaires
        Vérifie la cohérence des propriétaires et gestionnaires renseignés """
        cases = {
                 "ORANGE": ["ORANGE"],
                 "ERDF": ["ERDF"],
                 "ENEDIS": ["ENEDIS"],
                 "MANCHE NUMERIQUE": ["MANCHE FIBRE", "MANCHE TELECOM"] 
                 }
        for artere in self.arteres:
            for prop, gests in cases.items():
                if artere.AR_PRO_FOU.upper() == prop and not artere.AR_GEST_FO.upper() in gests:
                    self.log_error("Propriétaire: {}, gestionnaire(s) possible(s): {} (renseigné: {})".format(prop, ', '.join(gests), artere.AR_GEST_FO), item=artere)

        for equipement in self.equipements:
            for prop, gests in cases.items():
                if equipement.EQ_PRO.upper() == prop and not equipement.EQ_GEST.upper() in gests:
                    self.log_error("Propriétaire: {}, gestionnaire(s) possible(s): {} (renseigné: {})".format(prop, ', '.join(gests), equipement.EQ_GEST), item=equipement)

        for noeud in self.noeuds:
            for prop, gests in cases.items():
                if noeud.NO_PRO.upper() == prop and not noeud.NO_GEST.upper() in gests:
                    self.log_error("Propriétaire: {}, gestionnaire(s) possible(s): {} (renseigné: {})".format(prop, ', '.join(gests), noeud.NO_GEST), item=noeud)
                     
    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 de la PBO ne coincide avec le nom d'aucune des ZAPBO qui le contiennent", item=equipement)
                continue
            
            if equipement.EQ_STATUT != equipement.zapbo.STATUT:
                self.log_error("Le statut de la PBO ne coincide pas avec celui de sa ZAPBO", item=equipement)
                continue
                
    def test_comments(self):
        """ Présence de commentaires
        Vérifie la présence de commentaires là où ils sont attendus
        """
        for equipement in self.equipements:
            if equipement.EQ_STATUT.upper() != 'EN SERVICE' and not equipement.EQ_COMMENT:
                self.log_error("L'equipement n'est pas en REC, un commentaire devrait en préciser la raison", item=equipement)
                
        for zapbo in self.zapbos:
            if zapbo.STATUT.upper() != 'EN SERVICE' and not zapbo.COMMENTAIR:
                self.log_error("La Zapbo n'est pas en REC, un commentaire devrait en préciser la raison", item=zapbo)
                
    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)
            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])
            
            # 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]