2024-03-15 16:43:27 -04:00
|
|
|
from qgis.core import QgsApplication, QgsField, QgsProject, \
|
|
|
|
QgsProcessingFeedback, QgsVectorLayer, QgsVectorDataProvider, \
|
2024-03-26 16:04:43 -04:00
|
|
|
QgsExpressionContext, QgsExpressionContextUtils, edit
|
|
|
|
from qgis.PyQt.QtCore import QVariant
|
2024-03-15 12:08:52 -04:00
|
|
|
from qgis.analysis import QgsNativeAlgorithms
|
2024-03-15 16:43:27 -04:00
|
|
|
import processing
|
2024-03-15 11:26:23 -04:00
|
|
|
|
|
|
|
|
|
|
|
class ScrubLayer:
|
2024-03-15 12:08:52 -04:00
|
|
|
def __init__(self, qgis_path, layer_path, layer_name):
|
|
|
|
# Set the path to QGIS installation
|
|
|
|
QgsApplication.setPrefixPath(qgis_path, True)
|
|
|
|
|
|
|
|
# Initialize QGIS application
|
|
|
|
qgs = QgsApplication([], False)
|
|
|
|
qgs.initQgis()
|
|
|
|
|
2024-03-15 11:26:23 -04:00
|
|
|
self.layer_path = layer_path
|
|
|
|
self.layer_name = layer_name
|
2024-03-15 12:08:52 -04:00
|
|
|
self.layer = self.load_layer()
|
2024-03-16 16:33:47 -04:00
|
|
|
self.data_count = self.layer.featureCount()
|
2024-03-15 11:26:23 -04:00
|
|
|
|
2024-03-15 12:08:52 -04:00
|
|
|
def load_layer(self):
|
|
|
|
the_layer = QgsVectorLayer(self.layer_path, self.layer_name, 'ogr')
|
2024-03-15 11:26:23 -04:00
|
|
|
if not the_layer.isValid():
|
2024-03-15 12:08:52 -04:00
|
|
|
raise ValueError(f'Failed to load layer {self.layer_name} from {self.layer_path}')
|
2024-03-15 11:26:23 -04:00
|
|
|
else:
|
2024-03-15 12:08:52 -04:00
|
|
|
QgsProject.instance().addMapLayer(the_layer)
|
|
|
|
return the_layer
|
|
|
|
|
2024-03-16 16:27:25 -04:00
|
|
|
def fix_geometries(self, fixed_layer):
|
|
|
|
QgsApplication.processingRegistry().addProvider(QgsNativeAlgorithms())
|
|
|
|
fix_geometries_params = {
|
|
|
|
'INPUT': self.layer,
|
|
|
|
'METHOD': 0,
|
|
|
|
'OUTPUT': fixed_layer
|
|
|
|
}
|
|
|
|
processing.run("native:fixgeometries", fix_geometries_params)
|
|
|
|
|
2024-03-15 16:43:27 -04:00
|
|
|
def create_spatial_index(self):
|
|
|
|
QgsApplication.processingRegistry().addProvider(QgsNativeAlgorithms())
|
|
|
|
create_spatial_index_params = {
|
2024-03-16 16:27:25 -04:00
|
|
|
'INPUT': self.layer,
|
2024-03-15 16:43:27 -04:00
|
|
|
'OUTPUT': 'Output'
|
|
|
|
}
|
|
|
|
processing.run("native:createspatialindex", create_spatial_index_params)
|
2024-03-26 16:04:43 -04:00
|
|
|
print(f'Creating Spatial index for {self.layer_name} is completed.')
|
|
|
|
|
|
|
|
def spatial_join(self, joining_layer_path, joined_layer_path):
|
|
|
|
"""In QGIS, it is called 'Join attributes by Location'"""
|
|
|
|
params = {'INPUT': self.layer,
|
|
|
|
'PREDICATE': [0],
|
|
|
|
'JOIN': joining_layer_path,
|
|
|
|
'JOIN_FIELDS': [],
|
|
|
|
'METHOD': 0,
|
|
|
|
'DISCARD_NONMATCHING': False,
|
|
|
|
'PREFIX': '',
|
|
|
|
'OUTPUT': joined_layer_path}
|
|
|
|
|
|
|
|
feedback = QgsProcessingFeedback()
|
|
|
|
processing.run('native:joinattributesbylocation', params, feedback=feedback)
|
|
|
|
print(f'Spatial Join with input layer {self.layer_name} is completed.')
|
2024-03-16 16:27:25 -04:00
|
|
|
|
|
|
|
def clip_layer(self, overlay_layer, clipped_layer):
|
2024-03-26 16:04:43 -04:00
|
|
|
"""This must be tested"""
|
2024-03-16 16:27:25 -04:00
|
|
|
QgsApplication.processingRegistry().addProvider(QgsNativeAlgorithms())
|
|
|
|
clip_layer_params = {
|
|
|
|
'INPUT': self.layer_path,
|
|
|
|
'OVERLAY': overlay_layer,
|
|
|
|
'FILTER_EXPRESSION': '',
|
|
|
|
'FILTER_EXTENT': None,
|
|
|
|
'OUTPUT': clipped_layer
|
|
|
|
}
|
|
|
|
processing.run("native:clip", clip_layer_params)
|
|
|
|
print(f'Clipping of {self.layer_name} is completed.')
|
2024-03-15 16:43:27 -04:00
|
|
|
|
2024-03-26 16:04:43 -04:00
|
|
|
def merge_layers(self, layers_path, mered_layer_path):
|
|
|
|
pass
|
|
|
|
|
2024-03-26 17:12:14 -04:00
|
|
|
def multipart_to_singleparts(self, singleparts_layer_path):
|
|
|
|
QgsApplication.processingRegistry().addProvider(QgsNativeAlgorithms())
|
|
|
|
params = {'INPUT': self.layer,
|
|
|
|
'OUTPUT': singleparts_layer_path}
|
|
|
|
processing.run("native:multiparttosingleparts", params)
|
2024-03-26 16:04:43 -04:00
|
|
|
|
|
|
|
def split_layer(self):
|
|
|
|
pass
|
|
|
|
|
|
|
|
def delete_duplicates(self, deleted_duplicates_layer):
|
|
|
|
QgsApplication.processingRegistry().addProvider(QgsNativeAlgorithms())
|
|
|
|
params = {'INPUT': self.layer_path,
|
|
|
|
'OUTPUT': deleted_duplicates_layer}
|
|
|
|
processing.run("native:deleteduplicategeometries", params)
|
|
|
|
|
|
|
|
def delete_field(self, field_name):
|
|
|
|
QgsApplication.processingRegistry().addProvider(QgsNativeAlgorithms())
|
|
|
|
with edit(self.layer):
|
|
|
|
# Get the index of the column to delete
|
|
|
|
idx = self.layer.fields().indexFromName(field_name)
|
|
|
|
|
|
|
|
# Delete the field
|
|
|
|
self.layer.deleteAttribute(idx)
|
|
|
|
|
|
|
|
# Update layer fields
|
|
|
|
self.layer.updateFields()
|
|
|
|
|
|
|
|
def add_field(self, new_field_name):
|
|
|
|
functionalities = self.layer.dataProvider().capabilities()
|
|
|
|
|
|
|
|
if functionalities & QgsVectorDataProvider.AddAttributes:
|
|
|
|
new_field = QgsField(new_field_name, QVariant.Double)
|
|
|
|
self.layer.dataProvider().addAttributes([new_field])
|
|
|
|
self.layer.updateFields()
|
|
|
|
|
|
|
|
def assign_area(self, field_name):
|
|
|
|
self.layer.startEditing()
|
|
|
|
idx = self.layer.fields().indexFromName(field_name)
|
|
|
|
|
|
|
|
context = QgsExpressionContext()
|
|
|
|
context.appendScopes(QgsExpressionContextUtils.globalProjectLayerScopes(self.layer))
|
|
|
|
|
|
|
|
for feature in self.layer.getFeatures():
|
|
|
|
area = feature.geometry().area()
|
|
|
|
feature[idx] = area
|
|
|
|
self.layer.updateFeature(feature)
|
|
|
|
|
|
|
|
self.layer.commitChanges()
|
|
|
|
|
2024-03-16 16:33:47 -04:00
|
|
|
def __str__(self):
|
|
|
|
return f'The {self.layer_name} has {self.data_count} records.'
|
|
|
|
|
2024-03-15 12:08:52 -04:00
|
|
|
@staticmethod
|
|
|
|
def cleanup():
|
|
|
|
QgsApplication.exitQgis()
|
2024-03-15 11:26:23 -04:00
|
|
|
|
|
|
|
|
2024-03-15 12:08:52 -04:00
|
|
|
|
|
|
|
|