Mise en place du mécanisme de validation des données

.gitignore

@@ -6,4 +6,4 @@ Output/
 htmlcov/
 *.log
 *.log.1
-work/*
+/work/*

core/model.py

@@ -87,28 +87,3 @@ class Model():
         except KeyError:
             return value
 
-    # Fonctions CSV
-    def dump_to_csv(self, path):
-        """ Ajoute les données du modèle au format CSV dans le fichier spécifié en paramètre.
-        Créé le fichier s'il n'existe pas, avec une ligne d'en-tête """
-        if not path.exists():
-            logger.debug("Génère le fichier %s", path)
-            with open(path, 'w+', newline='') as f:
-                writer = csv.writer(f, 'tsv')
-                writer.writerow(self._fields)
-
-        with open(path, "a", newline='') as f:
-            writer = csv.writer(f, 'tsv')
-            writer.writerow([str(getattr(self, field)).replace("\t", " ") for field in self._fields])
-
-    @classmethod
-    def load_csv(cls, path):
-        """ parcourt les lignes du fichier csv et renvoie chaque ligne sous forme d'un objet Model
-        ATTENTION: chaque propriété dont le type n'est pas précisé dans _mapping aura le type 'string'
-        """
-        with open(path) as f:
-            reader = csv.reader(f, 'tsv')
-            fields = next(reader)
-            for row in csv.reader(f, 'tsv'):
-                data = {key: cls._parse(key, value) for key, value in zip(fields, row)}
-                yield(cls.from_dict(data))

main.py

@@ -1,11 +1,16 @@
 '''
+Python 3.7+
 
 @author: olivier.massot, sept 2018
 '''
+from dataclasses import dataclass
 from datetime import datetime
 import logging
+import sys
 
+from path import Path
 import shapefile
+import yaml
 
 from core import logconf
 from core.constants import MAIN
@@ -14,214 +19,119 @@ from core.constants import MAIN
 logger = logging.getLogger("netgeo_checker")
 logconf.start("netgeo_checker", logging.INFO)
 
-# # POUR TESTER, décommenter les lignes suivantes
-##-----------------------------------------------
-
-# logger.handlers = [h for h in logger.handlers if (type(h) == logging.StreamHandler)]
-# logger.warning("<<<<<<<<<<<<<<   Mode TEST   >>>>>>>>>>>>>>>>>")
+ETUDE, IN_DIR = False, MAIN / "work" / "SCOPELEC_CAP_097AP0_REC_180829_OK"
+# ETUDE, IN_DIR = True, MAIN / "rsc" / "data_in" / "SCOPELEC_101BP0_APD_180725"
 
-##-----------------------------------------------
 
-ETUDE, IN_DIR = False, MAIN / "rsc" / "data_in" / "SCOPELEC_CAP_097AP0_REC_180829_OK"
-# ETUDE, IN_DIR = True, MAIN / "rsc" / "data_in" / "SCOPELEC_101BP0_APD_180725"
+subject = MAIN / "work" / "SCOPELEC_CAP_097AP0_REC_180829_OK"
+checker = MAIN / "resources" / "netgeo_v2-2_doe.yaml"
 
-COUCHES = ["artere_geo.shp", "cable_geo.shp", "equipement_passif.shp", "noeud_geo.shp", "tranchee_geo.shp", "zapbo_geo.shp"]
+# prendre une archive en entrée
+# contrôler la presence des 6 fichiers shp
+# Pour chaque couche, vérifier le type de géometrie, la presence de données, la projection, l'inclusion du bounding rect dans le departement
+# Charger les modèles
+# Vérifier enregistrement par enregistrement la validité des données
 
 
 
-def main(etude=False):
+def check(subject, checker):    
+    """ prends un dossier ou une archive en entier et vérifie son contenu selon les règles données par le fichier de config """
+    archive, checker = Path(subject), Path(checker)
     
-    # Controle des tranchées
-    filename = IN_DIR / "tranchee_geo.shp"
+    if archive.isfile():
+        # extraire vers un dossier temp
+        # dirname = tempdir
+        pass
+    elif archive.isdir():
+        dirname = subject
+    else:
+        raise IOError(f"Impossible de trouver le fichier ou répertoire: {subject}")
     
-    logging.info("*****   Traitement du fichier '%s'   *****", filename.name)
-    logging.info("Type de dossier: %s", ("ADP" if etude else "DOE"))
+    logging.info("*****   Traitement de '%s'   *****", subject.name)
     
-    sf = shapefile.Reader(filename)
-
-    if sf.shapeType != shapefile.POLYLINE:
-        logger.error("Le fichier shapefile n'est pas de type PolyLigne", filename)
-        return
-
-    tranchees = sf.shapeRecords()
-    if not tranchees:
-        logger.warning("Le fichier shapefile ne contient aucune donnees")
+    logging.info("> Controlleur: '%s'", checker.name)
+    
+    with open(checker, "r") as cf:
+        config = yaml.load(cf)
+    
+    for filename, model in config["files"].items():
+        path_ = dirname / filename
+        logging.info("* Traitement de %s", path_.name)
         
-    for i, tr_record in enumerate(tranchees):
+        try:
+            sf = shapefile.Reader(path_)
+        except shapefile.ShapefileException:
+            logger.error("Fichier SHAPE illisible")
+            continue
         
-        fields = [f for f in sf.fields if f[0] != 'DeletionFlag']
-        tr = {field[0].lower(): tr_record.record[i] for i, field in enumerate(fields)}
+        shape_names = {1:"Point", 3:"Polyligne", 5:"Polygone"}
+        if sf.shapeType != model["shape_type"]:
+            logger.error("Le fichier shapefile n'est pas de type %s", shape_names[model["shape_type"]])
+            del sf
+            continue
+
+        records = sf.shapeRecords()
+        if not records:
+            if not model["can_be_empty"]:
+                logger.error("Le fichier shapefile ne contient aucune donnees")
+                del sf, records
+                continue
+            else:
+                logger.warning("Le fichier shapefile ne contient aucune donnees")
         
-        logging.info("\n> Enregistrement %s\n", i)        
+        if not "fields" in model:
+            continue
         
+        fields = [f[0] for f in sf.fields if f[0] != 'DeletionFlag']
         
-        if not tr.get("TR_CODE"):
-            logger.error("- TR_CODE est manquant")
-            
-        if not tr.get("TR_NOM"):
-            logger.error("- TR_NOM est manquant")
-            
-        if tr.get("TR_NOM") != tr.get("TR_CODE"):
-            logging.error("- Les champs TR_CODE et TR_NOM sont différents")
-
-        # Facultatif en etude
-        if not tr.get("TR_ID_INSE"):
-            if not etude:
-                logger.error("- TR_ID_INSE est manquant")
-        else:
-            pass
-#             if not tr["TR_ID_INSE"] in LISTE_INSEE:
-#                 logger.error("- TR_ID_INSE n'existe pas dans la table ...")
+        # controle d'éventuels champs inconnus
+        for f in fields:
+            if f not in model["fields"]:
+                logger.warning("Champs inconnu: %s", f)
         
-        # Facultatif en etude
-        if not tr.get("TR_VOIE"):
-            if not etude:
-                logger.error("- TR_VOIE est manquant")
-
-        if not tr.get("TR_TYP_IMP"):
-            logger.error("- TR_TYP_IMP est manquant")
-        else:
-            if not tr["TR_TYP_IMP"] in ("ACCOTEMENT STABILISE", "ACCOTEMENT NON STABILISE", "CHAUSSEE LOURDE", "CHAUSSEE LEGERE", "FOSSE", "TROTTOIR", "ESPACE VERT", "ENCORBELLEMENT"):
-                logger.error("- TR_TYP_IMP ne possède pas une valeur valide ('%s')", tr["TR_TYP_IMP"])
+        # parcours et controle des enregistrements
+        for i, record in enumerate(records):
             
-        if not tr.get("TR_MOD_POS"):
-            logger.error("- TR_MOD_POS est manquant")
-        else:
-            if not tr["TR_MOD_POS"] in ("TRADITIONNEL", "MICRO TRANCHEE", "FONCAGE 60", "FONCAGE 90", "FONCAGE 120", "TRANCHEUSE", "FORAGE URBAIN", "FORAGE RURAL", "ENCORBELLEMENT"):
-                logger.error("- TR_MOD_POS ne possède pas une valeur valide ('%s')", tr["TR_MOD_POS"])
+            logging.info("\n> Enregistrement n°%s\n", i)
+            record_data = {field: record.record[i] for i, field in enumerate(fields)}
             
-        if not tr.get("TR_MOD_POS_AI"):
-            logger.error("- TR_MOD_POS_AI est manquant")
-        else:
-            if not tr["TR_MOD_POS_AI"] in ("TRA", "ALL", "FONCAGE 60", "FON", "FOR", "ENC"):
-                logger.error("- TR_MOD_POS_AI ne possède pas une valeur valide ('%s')", tr["TR_MOD_POS_AI"])
-              
-        if not tr.get("TR_LONG"):
-            logger.error("- TR_LONG est manquant")
-        else:
-            try:
-                _ = float(tr["TR_LONG"])
-            except ValueError:
-                logger.error("- TR_LONG ne possède pas une valeur valide ('%s')", tr["TR_LONG"])
+            for fieldname, fieldmodel in model["fields"].items():
                 
-        # Facultatif en etude
-        if not tr.get("TR_LARG"):
-            if not etude:
-                logger.error("- TR_LARG est manquant")
-        else:
-            try:
-                _ = float(tr["TR_LARG"])
-            except ValueError:
-                logger.error("- TR_LARG ne possède pas une valeur valide ('%s')", tr["TR_LARG"])
+                try:
+                    val = record_data[fieldname]
+                except KeyError:
+                    if fieldmodel.get("required", True):
+                        logger.error("%s - Champs manquant", fieldname)
+                    continue
                 
-        if not tr.get("TR_REVET"):
-            logger.error("- TR_REVET est manquant")       
-        else:
-            if not tr["TR_REVET"] in ("SABLE", "BICOUCHE", "ENROBE", "BETON", "PAVE", "TERRAIN NATUREL"):
-                logger.error("- TR_REVET ne possède pas une valeur valide ('%s')", tr["TR_REVET"])
-                
-        # Facultatif en etude
-        if not tr.get("TR_CHARGE"):
-            if not etude:
-                logger.error("- TR_CHARGE est manquant")
-        else:
-            try:
-                _ = float(tr["TR_CHARGE"])
-            except ValueError:
-                logger.error("- TR_CHARGE ne possède pas une valeur valide ('%s')", tr["TR_CHARGE"])
-                
-        # Facultatif en etude
-        if not tr.get("TR_GRILLAG"):
-            if not etude:
-                logger.error("- TR_GRILLAG est manquant")
-        else:
-            try:
-                _ = float(tr["TR_GRILLAG"])
-            except ValueError:
-                logger.error("- TR_GRILLAG ne possède pas une valeur valide ('%s')", tr["TR_GRILLAG"])
+                type_ = fieldmodel.get("type", "str")
+                if type_ == "float":
+                    try:
+                        _ = float(val)
+                    except (TypeError, ValueError):
+                        logger.error("%s - Valeur invalide, un flottant est attendu ('%s')", fieldname, val)
+                        continue
+                elif type_ == "datetime":
+                    try:
+                        _ = datetime.strptime(val, fieldmodel.get("date_format", "%d/%m/%Y"))
+                    except ValueError:
+                        logger.error("%s - Valeur invalide, une date est attendu ('%s')", fieldname, val)
+                        continue
+                else: 
+                    if not fieldmodel.get("allow_empty", False) and not val:
+                        logger.error("%s - Champs vide", fieldname)
+                        continue
             
-        # Facultatif en etude
-        if not tr.get("TR_REMBLAI"):
-            if not etude:
-                logger.error("- TR_REMBLAI est manquant")
-        else:
-            pass
-#             if not tr["TR_REMBLAI"] in LISTE_REMBLAIS:
-#                 logger.error("- TR_REMBLAI ne possède pas une valeur valide ('%s')", tr["TR_REMBLAI"])
-
-        # Facultatif en etude
-        if not tr.get("TR_PLYNOX"):
-            if not etude:
-                logger.error("- TR_PLYNOX est manquant")
-        else:
-            if not tr["TR_PLYNOX"] in ("OUI", "NON"):
-                logger.error("- TR_PLYNOX ne possède pas une valeur valide ('%s')", tr["TR_PLYNOX"])
-                
-        # Facultatif en etude
-        if not tr.get("TR_PRO_VOI"):
-            if not etude:
-                logger.error("- TR_PRO_VOI est manquant")
-        else:
-            if not tr["TR_PRO_VOI"] in ("COMMUNE", "COMMUNAUTE DE COMMUNES", "DEPARTEMENT", "ETAT", "PRIVE"):
-                logger.error("- TR_PRO_VOI ne possède pas une valeur valide ('%s')", tr["TR_PRO_VOI"])
-                
-        # Facultatif en etude
-        if not tr.get("TR_GEST_VO"):
-            if not etude:
-                logger.error("- TR_GEST_VO est manquant")
-        else:
-            if not tr["TR_GEST_VO"] in ("COMMUNE", "COMMUNAUTE DE COMMUNES", "DEPARTEMENT", "ETAT", "PRIVE"):
-                logger.error("- TR_GEST_VO ne possède pas une valeur valide ('%s')", tr["TR_GEST_VO"])
-                
-        # Facultatif en etude
-        if not tr.get("TR_SCHEMA"):
-            if not etude:
-                logger.error("- TR_SCHEMA est manquant")
-        else:
-            pass
-#             if not tr["TR_SCHEMA"] in LISTE_SCHEMAS:
-#                 logger.error("- TR_SCHEMA ne possède pas une valeur valide ('%s')", tr["TR_REMBLAI"])
-
-        # Facultatif en etude
-        if not tr.get("TR_DATE_IN"):
-            if not etude:
-                logger.error("- TR_DATE_IN est manquant")
-        else:
-            try:
-                _ = datetime.strptime(tr["TR_DATE_IN"], "%d-%m-%Y")
-            except ValueError:
-                logger.error("- TR_DATE_IN ne possède pas une valeur ou un format valide ('%s')", tr["TR_DATE_IN"])
-                
-        # Facultatif en etude
-        if not tr.get("TR_SRC_GEO"):
-            if not etude:
-                logger.error("- TR_SRC_GEO est manquant")
-            
-        # Facultatif en etude
-        if not tr.get("TR_QLT_GEO"):
-            if not etude:
-                logger.error("- TR_QLT_GEO est manquant")
-            
-        # Facultatif en etude
-        if not tr.get("TR_PRO_MD"):
-            if not etude:
-                logger.error("- TR_PRO_MD est manquant")
+                try:
+                    if not val in fieldmodel["in_list"]:
+                        logger.error("%s - Valeur invalide, pas dans la liste ('%s')", fieldname, val)
+                        continue
+                except KeyError:
+                    pass
         
-        # Facultatif en etude
-        if not tr.get("TR_COMMENT"):
-            if not etude:
-                logger.error("- TR_COMMENT est manquant")
-            
-        # Facultatif en etude
-        if not tr.get("TR_STATUT"):
-            if not etude:
-                logger.error("- TR_STATUT est manquant")
-        else:
-            if not tr["TR_STATUT"] in ("EN ETUDE", "EN REALISATION", "EN SERVICE", "HORS SERVICE"):
-                logger.error("- TR_STATUT ne possède pas une valeur valide ('%s')", tr["TR_STATUT"])
-    del sf, tranchees
+        del sf, records
 
 
 if __name__ == "__main__":
-    main(ETUDE)
+    check(subject, checker)
     logger.info("-- Fin --")

resources/netgeo_v2-2_doe.yaml

@@ -0,0 +1,81 @@
+files:
+  "artere_geo.shp":
+    shape_type: 3
+    
+  "cable_geo.shp":
+    shape_type: 3
+    
+  "equipement_passif.shp":
+    shape_type: 1
+    
+  "noeud_geo.shp":
+    shape_type: 1
+    
+  "site_geo.shp":
+    shape_type: 1
+    
+  "tranchee_geo.shp":
+    shape_type: 3
+    can_be_empty: False
+    srid: 3949
+    fields:
+      TR_CODE:
+        type: str
+      TR_NOM:
+        type: str
+        tester: "lambda t, x: x=t.TR_CODE"
+      TR_ID_INSE:
+        type: str
+        in_list: []
+      TR_VOIE:
+        type: str
+      TR_TYP_IMP:
+        type: str
+        in_list: ["ACCOTEMENT STABILISE", "ACCOTEMENT NON STABILISE", "CHAUSSEE LOURDE", "CHAUSSEE LEGERE", "FOSSE", "TROTTOIR", "ESPACE VERT", "ENCORBELLEMENT"]
+      TR_MOD_POS:
+        type: str
+        in_list: ["TRADITIONNEL", "MICRO TRANCHEE", "FONCAGE 60", "FONCAGE 90", "FONCAGE 120", "TRANCHEUSE", "FORAGE URBAIN", "FORAGE RURAL", "ENCORBELLEMENT"]
+      TR_MPOS_AI:
+        type: str
+        in_list: ["TRA", "ALL", "FONCAGE 60", "FON", "FOR", "ENC"]
+      TR_LONG:
+        type: float
+      TR_LARG:
+        type: float
+      TR_REVET:
+        type: str
+        in_list: ["SABLE", "BICOUCHE", "ENROBE", "BETON", "PAVE", "TERRAIN NATUREL"]
+      TR_CHARGE:
+        type: float
+      TR_GRILLAG:
+        type: float
+      TR_REMBLAI:
+        type: str
+      TR_PLYNOX:
+        type: str
+        in_list: ["OUI", "NON"]
+      TR_PRO_VOI:
+        type: str
+        in_list: ["COMMUNE", "COMMUNAUTE DE COMMUNES", "DEPARTEMENT", "ETAT", "PRIVE"]
+      TR_GEST_VO:
+        type: str
+        in_list: ["COMMUNE", "COMMUNAUTE DE COMMUNES", "DEPARTEMENT", "ETAT", "PRIVE"]
+      TR_SCHEMA:
+        type: str
+      TR_DATE_IN:
+        type: datetime
+        date_format: "%d/%m/%Y"
+      TR_SRC_GEO:
+        type: str
+      TR_QLT_GEO:
+        type: str
+      TR_PRO_MD:
+        type: str
+      TR_COMMENT:
+        type: str
+        allow_empty: True
+      TR_STATUT:
+        type: str
+        in_list: ["EN ETUDE", "EN REALISATION", "EN SERVICE", "HORS SERVICE"]
+  "zapbo_geo.shp":
+    shape_type: 5