mncheck/core/mncheck.py

'''

@author: olivier.massot, 2018
'''
import importlib
import inspect
import logging
import pkgutil
from qgis.core import QgsProject, QgsWkbTypes, QgsGeometry, QgsPoint

from PyQt5.QtCore import QVariant

from core.checking import BaseChecker
from plugins import processing


logger = logging.getLogger("mncheck")


def list_schemas():
    import schemas
    return [name for _, name, ispkg in pkgutil.iter_modules(schemas.__path__) if not (ispkg or name[0] == '_')]

def get_schema(schema_name):
    return importlib.import_module("schemas." + schema_name)

def get_checkers(schema):
    return [cls for _, cls in inspect.getmembers(schema, predicate=inspect.isclass) \
            if issubclass(cls, BaseChecker) and not cls is BaseChecker]

class QgsModel():
    
    GEOM_UNKNOWN = 0
    GEOM_POINT = 1
    GEOM_LINE = 2
    GEOM_POLYGON = 3
    GEOM_MULTIPOINT = 4
    GEOM_MULTILINE = 5
    GEOM_MULTIPOLYGON = 6
    
    GEOM_NAMES = {0: "(AUCUN)", 1: "POINT", 2: "LIGNE", 3: "POLYGONE", 
                  4: "MULTI-POINT", 5:"MULTI-LIGNE", 6:"MULTI-POLYGONE"}
    
    layername = ""
    geom_type = 0
    bounding_box = (0,0,1,1)
    schema = {}
    pk = ""

    def __init__(self, qgs_feature):
        self._feature = qgs_feature
        
        for attr, value in self.attributes().items():
            if isinstance(value, QVariant):
                value = value.value() if not value.isNull() else ""
            setattr(self, attr, value)
    
    @property
    def feature(self):
        return self._feature
    
    def attributes(self):
        return dict(zip([f.name() for f in self._feature.fields()], self._feature.attributes()))
    
    @property
    def geom(self):
        return self._feature.geometry()
    
    def is_geometry_valid(self):
        return self._feature.geometry().isGeosValid()
        
    def get_geom_type(self):
        return QgsWkbTypes.singleType(self._feature.geometry().wkbType())
        
    def get_bounding_box(self):
        bb = self._feature.geometry().boundingBox()
        return (bb.xMinimum(), bb.yMinimum(), bb.xMaximum(), bb.yMaximum())
    
    def get_points(self):
        if self.geom.isNull():
            return []
        
        multi_geom = QgsGeometry()
        temp_geom = []
        
        if self.geom.type() == 0: # it's a point
            if self.geom.isMultipart():
                temp_geom = self.geom.asMultiPoint()
            else:
                temp_geom.append(self.geom.asPoint())
        elif self.geom.type() == 1: # it's a line
            if self.geom.isMultipart():
                multi_geom = self.geom.asMultiPolyline() #multi_geog is a multiline
                for i in multi_geom: #i is a line
                    temp_geom.extend( i )
            else:
                temp_geom = self.geom.asPolyline()
        elif self.geom.type() == 2: # it's a polygon
            if self.geom.isMultipart():
                multi_geom = self.geom.asMultiPolygon() #multi_geom is a multipolygon
                for i in multi_geom: #i is a polygon
                    for j in i: #j is a line
                        temp_geom.extend( j )
            else:
                multi_geom = self.geom.asPolygon() #multi_geom is a polygon
                for i in multi_geom: #i is a line
                    temp_geom.extend( i )
                
        return [QgsPoint(p) for p in temp_geom]

    @classmethod
    def full_buffer(cls, distance):
        layer = next((l for l in QgsProject.instance().mapLayers().values() if l.name().lower() == cls.layername.lower()))

        result = processing.run("native:buffer", {'INPUT':layer.dataProvider().dataSourceUri(),  #@UndefinedVariable
                                                 'DISTANCE':distance,
                                                 'SEGMENTS':5,
                                                 'END_CAP_STYLE':0,
                                                 'JOIN_STYLE':0,
                                                 'MITER_LIMIT':2,
                                                 'DISSOLVE':True,
                                                 'OUTPUT':'memory:mncheck_temp'})

        buffer = next((f for f in result['OUTPUT'].getFeatures()))
        
        return buffer.geometry()


def validate(schema_name):
    try:
        schema = get_schema(schema_name)
    except ModuleNotFoundError:
        logger.critical(f"Le schéma {schema_name} n'existe pas")
        return
    
    results = schema.checker.run()
    
    return results