Layer Builder

ExtractFeatureInfo(layer_pattern, feature_type, filter='', formula='', default_value={})

Dataclass that store features extraction informations.

Parameters:

Name	Type	Description	Default
layer_pattern	str	Pattern to match a layer file.	required
filter	str	Filter expression to filter layer. It shall be in QgsExpression format.	''
formula	str	Expression to compute buffer distance parameter. It shall be in QgsExpression format.	''
feature_type	str	Type of feature that will be extracted. i.e 'building', 'road', etc.	required
default_value	typing.Dict[str, float]	Dict of str-values that map field used in formula with their default values if the feature got NULL.	{}

Source code in niva/core/layer_builder/csv_parse.py

def __init__(
    self,
    layer_pattern: str,
    feature_type: str,
    filter: str = "",
    formula: str = "",
    default_value: Dict[str, float] = {},
):
    self.layer_pattern = layer_pattern
    self.feature_type = feature_type
    self.filter = filter
    self.formula = formula
    self.default_value = default_value
    super().__init__()

FeatureExtractor(extract_info, roi, **kwargs)

Bases: niva.core.mixins.QgsLogicMixin

Manage extraction of features from a layer.

Parameters:

Name	Type	Description	Default
extract_info	niva.core.layer_builder.csv_parse.ExtractFeatureInfo	All information to filter & extract features.	required
roi	qgis.core.QgsVectorLayer	Region of interest.	required

Source code in niva/core/layer_builder/csv_parse.py

def __init__(self, extract_info: ExtractFeatureInfo, roi: QgsVectorLayer, **kwargs):
    self.extract_info = extract_info
    self.roi = roi
    super().__init__(**kwargs)

fill_null(layer)

Replace NULL values for field needed in buffer distance calculation by median of values for each combination of fields used in filter expression. If all values are NULL use extract informations default value for field.

Parameters:

Name	Type	Description	Default
layer	qgis.core.QgsVectorLayer	Layer where to fill NULL values.	required

Source code in niva/core/layer_builder/csv_parse.py

def fill_null(self, layer: QgsVectorLayer):
    """Replace NULL values for field needed in buffer distance calculation by median of values for each combination of fields used in filter expression.
    If all values are NULL use extract informations default value for field.

    Args:
        layer (QgsVectorLayer): Layer where to fill NULL values.
    """
    if self.extract_info.filter_fields:
        self._fill_null_categorical(layer)
    else:
        self._fill_null_non_categorical(layer)

extract_data(path)

Extract features from layer at path.

Parameters:

Name	Type	Description	Default
path	str	Path of layer to extract features from.	required

Returns:

Name	Type	Description
QgsVectorLayer	qgis.core.QgsVectorLayer	Layer of extracted features.

Source code in niva/core/layer_builder/csv_parse.py

def extract_data(
    self, path: str
) -> QgsVectorLayer:  # Add exception management if not layer at path.
    """Extract features from layer at path.

    Args:
        path (str): Path of layer to extract features from.

    Returns:
        QgsVectorLayer:
            - Layer of extracted features.
    """

    layer_crs = QgsVectorLayer(path).crs()
    self._projected_roi = CRSHandler.reproject(
        self.roi, layer_crs, context=self.context
    )

    # get layer that fits roi extent
    extract_params = pt.ExtractByLocation.build_parameters()
    extract_params["INPUT"] = path
    extract_params["PREDICATE"] = [0]
    extract_params["INTERSECT"] = self._projected_roi
    self.fit_roi_layer = pt.ExtractByLocation.apply(
        extract_params, context=self.context
    )["OUTPUT"]

    # Fix geometries

    fixed_params = pt.FixGeometries.build_parameters()
    fixed_params["INPUT"] = self.fit_roi_layer
    fixed_params["METHOD"] = 1
    self._fixed = pt.FixGeometries.apply(fixed_params, context=self.context)[
        "OUTPUT"
    ]

    # apply filter
    if self.extract_info.filter_expression.expression():
        expression_params = pt.ExtractByExpression.build_parameters(build_opt=True)
        expression_params["INPUT"] = self._fixed
        expression_params["EXPRESSION"] = (
            self.extract_info.filter_expression.expression()
        )

        self._filtered_layer = pt.ExtractByExpression.apply(
            expression_params, context=self.context
        )["OUTPUT"]
    else:
        self._filtered_layer = self._fixed
        self.context.temporaryLayerStore().addMapLayer(self._filtered_layer)

    if self.extract_info.buffer_fields:
        self.fill_null(self._filtered_layer)

    # apply buffer
    if self.extract_info.buffer_expression.expression():  # TODO assert layer is polygon if no buffer or it will fail when insert (assertion befor this step, when load layer)
        buffer_params = pt.Buffer.build_parameters()
        buffer_params["INPUT"] = self._filtered_layer
        buffer_params["DISTANCE"] = QgsProperty.fromExpression(
            self.extract_info.buffer_expression.expression()
        )
        self._buffer_layer = pt.Buffer.apply(buffer_params, context=self.context)[
            "OUTPUT"
        ]
    else:
        self._buffer_layer = self._filtered_layer

    # reproject result to roi crs
    self._reprojected: QgsVectorLayer = CRSHandler.reproject(
        self._buffer_layer, self.roi.crs(), context=self.context
    )
    return self._reprojected

LayerBuilder(csv_path, roi, **kwargs)

Bases: niva.core.mixins.QgsLogicMixin

Mange layer creation by extracting features from a csv that detail layers to parse and features transformations and a region of interest.

Parameters:

Name	Type	Description	Default
csv_path	str	Path to CSV file.	required
roi	qgis.core.QgsVectorLayer	Region Of Interest.	required

Source code in niva/core/layer_builder/csv_parse.py

def __init__(self, csv_path: str, roi: QgsVectorLayer, **kwargs):
    self.roi = roi
    self.extract_infos = []
    self._build_layer()
    super().__init__(**kwargs)
    self.extract_infos = self.parse_csv(csv_path)

parse_csv(csv)

Parse a CSV file and build list of ExtractFeatureInfo at self.extract_infos.

Parameters:

Name	Type	Description	Default
csv	str	Path to CSV file to parse.	required

Returns:

Type	Description
typing.List[niva.core.layer_builder.csv_parse.ExtractFeatureInfo]	List[ExtractFeatureInfo]: - List of ExtractFeatureInfo.

Source code in niva/core/layer_builder/csv_parse.py

def parse_csv(self, csv: str) -> List[ExtractFeatureInfo]:
    """Parse a CSV file and build list of ExtractFeatureInfo at self.extract_infos.

    Args:
        csv (str): Path to CSV file to parse.

    Returns:
        List[ExtractFeatureInfo]:
            - List of ExtractFeatureInfo.
    """
    csv_layer = QgsVectorLayer(csv)
    tasks = []
    for row_id, feature in enumerate(csv_layer.getFeatures()):
        try:
            task = ExtractFeatureInfo.from_feature(feature)
            tasks.append(task)
        except NivaException as e:
            row_display = dict(
                zip([f.name() for f in feature.fields()], feature.attributes())
            )
            raise CSVParsingError(
                f"Error at row {row_id + 1}: {row_display}"
            ) from e

    return tasks

update(data, feature_type)

Update self.layer with features in data.

Parameters:

Name	Type	Description	Default
data	qgis.core.QgsVectorLayer	Layer with extracted features.	required
feature_type	str	Feature type to write in self.layer.	required

Source code in niva/core/layer_builder/csv_parse.py

def update(self, data: QgsVectorLayer, feature_type: str):
    """Update self.layer with features in data.

    Args:
        data (QgsVectorLayer): Layer with extracted features.
        feature_type (str): Feature type to write in self.layer.
    """
    extracted_feats = []

    if self._layer.wkbType() != data.wkbType():
        convert_p = pt.ConvertGeometryType.build_parameters()
        convert_p["INPUT"], convert_p["TYPE"] = data, 4
        data = pt.ConvertGeometryType.apply(convert_p, context=self.context)[
            "OUTPUT"
        ]

    if QgsWkbTypes.hasZ(data.wkbType()):
        drop_p = pt.DropMZValues.build_parameters()
        drop_p["INPUT"], drop_p["DROP_Z_VALUES"] = data, True
        data = pt.DropMZValues.apply(drop_p, context=self.context)["OUTPUT"]

    for feature in data.getFeatures():
        feat = QgsFeature()
        feat.setGeometry(feature.geometry())
        feat.setFields(self._layer.fields())
        feat.setAttributes([feature_type])
        extracted_feats.append(feat)

    self._layer.dataProvider().addFeatures(extracted_feats)

run_extraction(extrac_info, folder)

Extract data from layers in folder using extrac_info.

Parameters:

Name	Type	Description	Default
extrac_info	``ExtractFeatureInfo``	Informations of extraction.	required
folder	``str``	Folder to recursively match layers.	required

Source code in niva/core/layer_builder/csv_parse.py

def run_extraction(self, extrac_info: ExtractFeatureInfo, folder: str):
    """Extract data from layers in folder using extrac_info.

    Args:
        extrac_info (``ExtractFeatureInfo``): Informations of extraction.
        folder (``str``): Folder to recursively match layers.
    """
    layers_paths = self._find_layers(extrac_info.layer_pattern, folder)
    if layers_paths:
        self.logger.info(
            f"Found {len(layers_paths)} path that fits {extrac_info.layer_pattern} pattern in folder {folder}."
        )
    else:
        self.logger.warning(
            f"No paths fitting {extrac_info.layer_pattern} pattern in folder {folder} "
        )
    feature_extractor = FeatureExtractor(extract_info=extrac_info, roi=self.roi)
    for path in layers_paths:
        data = feature_extractor.extract_data(path)
        self.update(data, extrac_info.feature_type)

build(folder)

Launch all extractions to build layer.

Parameters:

Name	Type	Description	Default
folder	``str``	Folder to match layers	required

Source code in niva/core/layer_builder/csv_parse.py

def build(self, folder: str):
    """Launch all extractions to build layer.

    Args:
        folder (``str``): Folder to match layers"""
    total = len(self.extract_infos)
    for i, extraction_info in enumerate(self.extract_infos):
        try:
            self.run_extraction(extraction_info, folder)
        except NivaException as e:
            raise DataExtractionError(
                f"Data extraction error for task {extraction_info}"
            ) from e

        self.feedback.setProgress(((i + 1) / total) * 100)
        self.cancel()