mncheck/core/validator.py

'''


    @author: olivier.massot, sept. 2018
'''
import logging
from qgis.core import QgsProject

from core.cerberus_ import CerberusErrorHandler, \
    _translate_messages, ExtendedValidator
from core.model import QgsModel
from core.validation_errors import CRITICAL, DataError, GeomTypeError, BoundingBoxError, \
    MissingLayer, WrongSrid, InvalidGeometry, TechnicalValidationError


logger = logging.getLogger("mncheck")

class ValidatorInterruption(BaseException): 
    pass

class Checkpoint():
    def __init__(self, name, valid=True):
        self.name = name
        self.valid = valid

class BaseValidator():
    schema_name = ""
    models = {}
    dataset = {}
    
    def __init__(self):
        self.valid = True
        self.checkpoints = []
        self.errors = []
        self._current_checkpoint_valid = True
        self.dt = 0
    
    def checkpoint(self, title):
        logger.info(f"Checkpoint: {title}")
        self.checkpoints.append(Checkpoint(title, self._current_checkpoint_valid))
        self._current_checkpoint_valid = True
        if self.errors:
            self.valid = False
            if self.critical_happened():
                logger.info("Contrôle de données interrompu en raison d'une erreur critique dans les données")
                raise ValidatorInterruption()
            
    def critical_happened(self):
        return any([err.level == CRITICAL for err in self.errors])
             
    def log_error(self, validation_error):
        self._current_checkpoint_valid = False
        self.errors.append(validation_error)
         
    @classmethod
    def submit(cls, report_name=""):
        validator = cls()
        try:
            validator.validate()
        except ValidatorInterruption:
            pass
        
        report = validator.build_report(validator.schema_name, report_name or "(nom inconnu)")
        return report
    
    def validate(self):
        
        # Chargement des données en mémoire
        self._load_layers()
        self.checkpoint("Chargement des données")
        
        # Controle la structure des données (champs, formats et types)
        self._structure_validation()
        self.checkpoint("Contrôle de la structure des données")
         
        # Validation technique
        try:
            self._technical_validation()
            self.checkpoint("Validation Métier")
        except ValidatorInterruption:
            raise
        except:
#             logger.error("".join(traceback.format_exception(*sys.exc_info())))
            logger.exception("Une erreur s'est produite")
            self.log_error(TechnicalValidationError("Contrôle des données interrompu"))
            self.checkpoint("Validation Métier [interrompu]")
    
    def _load_layers(self):
    
        for model in self.models:
            layername = model.layername
            
            layer = next((l for l in QgsProject.instance().mapLayers().values() if l.name().lower() == layername))
            if not layer:
                self.log_error(MissingLayer("Couche manquante: '{}'".format(layername)))
                continue
            
            self.dataset[model] = []
            
            if layer.crs().authid() != model.crs:
                self.log_error(WrongSrid("Mauvaise projection: {} (attendu: {})".format(layer.crs().authid(), model.crs)))
            
            for feature in layer.getFeatures():
                
                item = model(feature)
                
                self.dataset[model].append(item)

    

    def _structure_validation(self):
        
        for model in self.models:
            v = ExtendedValidator(model.schema, purge_unknown=True, error_handler=CerberusErrorHandler, require_all=True)
            xmin, ymin, xmax, ymax = model.bounding_box
            
            for item in self.dataset[model]:

                # geom valid
                if not item.is_geometry_valid():
                    self.log_error(InvalidGeometry("La géométrie de l'objet est invalide: {}".format(item), layername=model.layername, field="geom"))

                # geom type
                if item.get_geom_type() != model.geom_type:
                    self.log_error(GeomTypeError("Type de géométrie invalide: {} (attendu: {})".format(item.geom_name, QgsModel.GEOM_NAMES[model.geom_type]), layername=model.layername, field="geom"))

                # bounding box
                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(BoundingBoxError("Situé hors de l'emprise autorisée", layername=model.layername, field="geom"))

                v.validate(item.__dict__)
                
                for field, verrors in v.errors.items():
                    for err in verrors:
                        self.log_error(DataError(_translate_messages(err), layername=model.layername, field=field))
    
    @classmethod
    def _technical_validation(cls):
        raise NotImplementedError()
    
    def build_report(self, schema, layername):
        report = {}
        report["schema"] = schema
        report["reportname"] = ""
        report["checkpoints"] = [{"name": chk.name, "valid": chk.valid} for chk in self.checkpoints]
        
        report["errors"] = {}
        
        for err in self.errors:
            if not err.name in report["errors"]:
                report["errors"][err.name] = {"help": err.help, "order_": err.order_, "level": err.level, "list": []}
            
            err_report = {"layername": err.layername or "-",
                          "field": err.field or "-",
                          "message": err.message}
            if err_report not in report["errors"][err.name]["list"]:
                report["errors"][err.name]["list"].append(err_report)
            
        return report