Ejercicio 7: Cálculo de indicadores hídricos

Este ejercicio es distinto a los seis anteriores: en vez de trabajar con capas ráster (imágenes), trabaja con una capa vectorial de puntos (localidades) y modifica directamente su tabla de atributos, calculando 3 indicadores de acceso a servicios de vivienda a partir de datos censales.

Conceptos clave de POO y PyQGIS usados en el ejercicio

Sin capa de salida separada: a diferencia de todos los ejercicios anteriores, initAlgorithm solo declara una entrada (la capa vectorial) y ninguna salida de tipo "destino": el resultado se escribe directamente sobre la misma capa de entrada, agregando columnas nuevas a su tabla de atributos.
Manipular una capa vectorial en memoria: QgsVectorLayer y su dataProvider() permiten leer y modificar los campos (columnas) y las entidades (filas) de una capa vectorial directamente desde código, algo que no aparecía en los ejercicios de raster.
Expresiones de QGIS evaluadas por código: QgsExpression permite construir una fórmula de QGIS (como las que se escriben en el "calculador de campos") como texto y evaluarla programáticamente para cada fila de la tabla.
Sesión de edición con with edit(...): es el equivalente en PyQGIS a iniciar y guardar automáticamente la edición de una capa; si ocurre un error dentro del bloque, los cambios no se confirman, evitando dejar la capa en un estado inconsistente.
Iterar sobre las entidades (features) de una capa: for renglon in localidades.getFeatures(): recorre cada punto (cada localidad) de la capa, uno por uno, para calcular y asignar su indicador correspondiente.

Análisis y explicación de la estructura y lógica del código

07_indicadores_hidricos.py

"""
Script para calcular indicadores hídricos de localidades.
Se oupan los atributos asociados a una capa vectorial de puntos
(shapefile)
"""

#Importamos las blibliotecas o módulos necesarios para el correcto funcionamiento del script
from qgis.core import (
    QgsProcessingAlgorithm, #Clase base de nuestra clase creada
    QgsProcessingParameterVectorLayer, #Define la capa vectorial de entrada
    QgsWkbTypes, # Restringe la entrada a geometrías de puntos
    QgsProcessingMultiStepFeedback, #Muestra ek avance y perminte cancelar el proceso
    QgsVectorLayer, #Carga y manipula la capa vectorial
    QgsField, #Para la creación de los campos
    QgsExpression, #Contruye y evalúa las expresiones
    QgsExpressionContext, #Define el contexto para evakluar cada feature
    QgsExpressionContextUtils, #Añade los ámbitos globalres del proyecto y de la capa
    edit, #Abre una sesión de edición para modificar atributos
    QgsProject #Para agregar la capa de salida el proyecto de QGIS
    )
from qgis.PyQt.QtCore import QVariant #Para definir el tipo de dato de los campos nuevos

#Se crea la clase para la herramienta
class indicadores_hidricos(QgsProcessingAlgorithm):
    #1. Nombre de la herramienta
    def name(self):
        return 'indicadores_hidricos'
        
    #2. Nombre a mostrar en la GUI
    def displayName(self):
        return 'Cálculo de indicadores hídricos'
        
    #3. Grupo de herramientas al que pertenece la herramienta
    def group(self):
        return 'Herramientas vectoriales'
        
    #4. ID del grupo de herramientas
    def groupId(self):
        return 'herramientas_vectoriales'
        
    #5. Descripción breve de la heramienta
    def shortHelpString(self):
        descripcion = """
            Esta herramienta calcula indicadores hídricos básicos a partir de los
            datos de un censo de población, un conteo de población o una encuesta
            intercensal. En todos los casos los datos deben estar a nivel de localidad.
            
            Los indicadores requieren las variables:
            1. Total de viviendas particulares habitadas (VPH).
            2. Total de VPH con servicio de agua dentro de la vivienda.
            3. Total de VPH con servicio de sanitario.
            4. Total de VPH con servicio de drenaje.
            
            Al final se presentan los indicadores como proporción de viviendas
            con dichos servicios. Los datos de salida se almacenan en la misma tabla
            de atributos de la capa vectorial de entrada.
        """
        return descripcion
        
    #6. Cargar las entradas como parámetros de algorítmo
    def initAlgorithm(self, config=None):
        #Capa vectorial de entrada
        self.addParameter(QgsProcessingParameterVectorLayer('capa_entrada', 'Capa vectorial de entrada con datos de viviendas', defaultValue=None))
        
        
    #7. Algoritmo de procesamiento
    def processAlgorithm(self, parameters, context, model_feedback):
        feedback = QgsProcessingMultiStepFeedback(3, model_feedback)
        outputs = {}
        results = {}
        
        #Asignar el vector a una variable para manejarla en el código
        localidades = QgsVectorLayer(parameters["capa_entrada"],'Localidades','ogr')
        #Recuperar el proveedor de la capa vectorial
        localidades_provider = localidades.dataProvider()
        #Crear un diccionario para iterar en el cálcilo de los indicadores
        indicadores_dict = {'ind1':['IndAguaDV', 'VPH_AGUADV'],
                            'ind2':['IndExcsa', 'VPH_EXCSA'],
                            'ind3':['IndDrenaj', 'VPH_DRENAJ']
                            }
        #Variable para el total de viviendas habitadas
        total_VPH = 'TVIVPARHAB'
        
        step = 0
        
        #Verificación de campos y cálculo de indicadores
        for indicador in indicadores_dict:
            #Verificar si ya existe el campo que se desea crear. En caso de que exista, borrar.
            indicador_hidrico = indicadores_dict[indicador][0]
            campo_de_tabla = indicadores_dict[indicador][1]
            for campo in localidades.fields():
                nombre_campo = campo.name()
                if nombre_campo == indicador_hidrico:
                    index_campo = localidades_provider.fieldNameIndex(nombre_campo)
                    localidades_provider.deleteAttributes([index_campo])
                    localidades.updateFields()
                    print('Se borró un campo ya existente.')
            
            #Agergar el nuevo campo a la tabla de atributos
            localidades_provider.addAttributes([QgsField(indicador_hidrico, QVariant.Double)])
            localidades.updateFields()
            
            #Expresión con el cálculo del indicador
            exp = QgsExpression(f'({campo_de_tabla} / {total_VPH}) * 100')
            #Iteracion 1: '(VPH_AGUADV / TVIVPARHAB) * 100'
            #Iteracion 2: '(VPH_EXCSA / TVIVPARHAB) * 100'
            #Iteracion 3: '(VPH_DRENAJ / TVIVPARHAB) * 100'
            #Se valida la expresión
            validacion_expresion = QgsExpression.isValid(exp)
            print('¿Expresion válida?', validacion_expresion)
            
            #Si la expresión es válida, entonces de aplica en la tabla de atributos
            if validacion_expresion == True:
                contexto_capa = QgsExpressionContext()
                contexto_capa.appendScopes(QgsExpressionContextUtils.globalProjectLayerScopes(localidades))
                #Aplicación de la expresion
                with edit(localidades):
                    for renglon in localidades.getFeatures():
                        contexto_capa.setFeature(renglon)
                        renglon[indicador_hidrico] = exp.evaluate(contexto_capa)
                        localidades.updateFeature(renglon)
                print(f'Expresión aplicadada para calcular {indicador_hidrico}')
            else:
                print(f'No se pudo aplicar la expresión para calcular {indicador_hidrico}')
                
            #Señalar el progreso
            step += 1
            feedback.setCurrentStep(step)
            if feedback.isCanceled():
                return{}
                
        #Regresar el resultado
        outputs['indicadores'] = parameters['capa_entrada']
        results['indicadores'] = outputs['indicadores']
        
        #Despliegue de la capa con el resultado
        if localidades.isValid() == True:
            QgsProject.instance().addMapLayer(localidades)
        else:
            print('No se pudo cargar la capa vectorial.')
            
        return results
    
    #8. Creación de la instancia
    def createInstance(self):
        return self.__class__()

Sección 1 · Para qué sirve cada biblioteca importada

from qgis.core import (
    QgsProcessingAlgorithm, #Clase base de nuestra clase creada
    QgsProcessingParameterVectorLayer, #Define la capa vectorial de entrada
    QgsWkbTypes, # Restringe la entrada a geometrías de puntos
    QgsProcessingMultiStepFeedback, #Muestra ek avance y perminte cancelar el proceso
    QgsVectorLayer, #Carga y manipula la capa vectorial
    QgsField, #Para la creación de los campos
    QgsExpression, #Contruye y evalúa las expresiones
    QgsExpressionContext, #Define el contexto para evakluar cada feature
    QgsExpressionContextUtils, #Añade los ámbitos globalres del proyecto y de la capa
    edit, #Abre una sesión de edición para modificar atributos
    QgsProject #Para agregar la capa de salida el proyecto de QGIS
    )
from qgis.PyQt.QtCore import QVariant #Para definir el tipo de dato de los campos nuevos

from qgis.core import (...): en vez de escribir una línea from ... import ... por cada elemento (como en los ejercicios anteriores), aquí se agrupan todas las importaciones de qgis.core dentro de un solo paréntesis. Es otra forma válida de organizar los import cuando son muchos.
QgsProcessingAlgorithm: clase base de la que hereda indicadores_hidricos, igual que en todos los ejercicios anteriores.
QgsProcessingParameterVectorLayer: a diferencia de QgsProcessingParameterRasterLayer (usado en los ejercicios 1 a 6), este parámetro declara una capa vectorial de entrada (puntos, líneas o polígonos).
QgsWkbTypes: se importa para poder restringir el tipo de geometría aceptado (en este script se usaría para exigir que la capa sea de puntos), aunque en la versión actual del código no se aplica esa restricción dentro de initAlgorithm.
QgsProcessingMultiStepFeedback: reporta el progreso de los 3 pasos (uno por cada indicador calculado) y permite cancelar la ejecución.
QgsVectorLayer: es la clase que representa una capa vectorial cargada en memoria; se usa para volver a abrir la capa de entrada y poder leer/escribir su tabla de atributos.
QgsField: se usa para definir un nuevo campo (columna) que se agregará a la tabla de atributos, junto con su tipo de dato.
QgsExpression: permite construir, como texto, una fórmula de QGIS (por ejemplo una división y una multiplicación) y luego evaluarla.
QgsExpressionContext: representa el "contexto" necesario para evaluar una expresión sobre una entidad (feature) específica de la capa.
QgsExpressionContextUtils: aporta funciones auxiliares para completar ese contexto con la información del proyecto y de la capa.
edit: función especial de PyQGIS que abre automáticamente una sesión de edición sobre una capa (y la guarda o la cancela sola, según si hubo errores).
QgsProject: se usa al final para agregar la capa ya modificada al proyecto actual de QGIS, de modo que el usuario la vea reflejada en el panel de capas.
from qgis.PyQt.QtCore import QVariant: QVariant proviene de Qt (la biblioteca gráfica en la que está basada la interfaz de QGIS) y se usa para indicar el tipo de dato del nuevo campo (en este caso, QVariant.Double, un número decimal).

Sección 2 · Para qué sirve cada método de la clase

name(): devuelve 'indicadores_hidricos', el identificador interno de la herramienta.
displayName(): devuelve 'Cálculo de indicadores hídricos', el nombre visible en QGIS.
group() / groupId(): ubican la herramienta en el grupo 'Herramientas vectoriales', distinto de los grupos usados en los ejercicios anteriores.
shortHelpString(): explica qué variables censales necesita la capa de entrada (viviendas con agua, sanitario y drenaje) y qué calculará la herramienta.
initAlgorithm(): declara la única entrada necesaria: la capa vectorial de localidades.
processAlgorithm(): contiene toda la lógica de cálculo: crea los campos nuevos, arma y evalúa las expresiones, y escribe los resultados en la tabla de atributos.
createInstance(): permite a QGIS crear nuevas copias de la herramienta.

Sección 3 · Declaración de parámetros (`initAlgorithm`)

    def initAlgorithm(self, config=None):
        #Capa vectorial de entrada
        self.addParameter(QgsProcessingParameterVectorLayer('capa_entrada', 'Capa vectorial de entrada con datos de viviendas', defaultValue=None))

Solo se declara un parámetro: 'capa_entrada', la capa vectorial con los datos censales de vivienda. No hay parámetro de salida, porque el resultado se escribirá directamente sobre esta misma capa.

Sección 4 · Preparar la capa y el diccionario de indicadores

        feedback = QgsProcessingMultiStepFeedback(3, model_feedback)
        outputs = {}
        results = {}
        
        #Asignar el vector a una variable para manejarla en el código
        localidades = QgsVectorLayer(parameters["capa_entrada"],'Localidades','ogr')
        #Recuperar el proveedor de la capa vectorial
        localidades_provider = localidades.dataProvider()
        #Crear un diccionario para iterar en el cálcilo de los indicadores
        indicadores_dict = {'ind1':['IndAguaDV', 'VPH_AGUADV'],
                            'ind2':['IndExcsa', 'VPH_EXCSA'],
                            'ind3':['IndDrenaj', 'VPH_DRENAJ']
                            }
        #Variable para el total de viviendas habitadas
        total_VPH = 'TVIVPARHAB'
        
        step = 0

feedback = QgsProcessingMultiStepFeedback(3, model_feedback): se declaran 3 pasos, uno por cada indicador que se calculará.
localidades = QgsVectorLayer(parameters["capa_entrada"], 'Localidades', 'ogr'): vuelve a abrir la capa de entrada como un objeto QgsVectorLayer manipulable, usando 'ogr' como proveedor de datos (el controlador estándar para shapefiles y formatos similares).
localidades_provider = localidades.dataProvider(): obtiene el "proveedor de datos" de la capa, que es el objeto que permite agregar o eliminar campos de la tabla de atributos.
indicadores_dict = {...}: un diccionario donde cada llave ('ind1', 'ind2', 'ind3') guarda una lista con el nombre del nuevo campo a crear y el nombre del campo existente que se usará en el cálculo. Esto evita repetir 3 veces casi el mismo bloque de código: en su lugar, se recorre este diccionario con un ciclo.
total_VPH = 'TVIVPARHAB': guarda el nombre del campo que contiene el total de viviendas particulares habitadas, usado como denominador en los 3 indicadores.
step = 0: variable contador que llevará el registro de en qué paso va la ejecución, para actualizar la barra de progreso.

Sección 5 · Ciclo principal: crear el campo y calcular cada indicador

        for indicador in indicadores_dict:
            #Verificar si ya existe el campo que se desea crear. En caso de que exista, borrar.
            indicador_hidrico = indicadores_dict[indicador][0]
            campo_de_tabla = indicadores_dict[indicador][1]
            for campo in localidades.fields():
                nombre_campo = campo.name()
                if nombre_campo == indicador_hidrico:
                    index_campo = localidades_provider.fieldNameIndex(nombre_campo)
                    localidades_provider.deleteAttributes([index_campo])
                    localidades.updateFields()
                    print('Se borró un campo ya existente.')
            
            #Agergar el nuevo campo a la tabla de atributos
            localidades_provider.addAttributes([QgsField(indicador_hidrico, QVariant.Double)])
            localidades.updateFields()

for indicador in indicadores_dict:: recorre las 3 llaves del diccionario ('ind1', 'ind2', 'ind3'), repitiendo todo el bloque interior una vez por cada indicador.
indicador_hidrico = indicadores_dict[indicador][0]: obtiene el nombre del campo nuevo a crear (por ejemplo, 'IndAguaDV').
campo_de_tabla = indicadores_dict[indicador][1]: obtiene el nombre del campo existente que se usará en la fórmula (por ejemplo, 'VPH_AGUADV').
for campo in localidades.fields():: recorre todos los campos actuales de la tabla de atributos, para revisar si el campo que se quiere crear ya existe de una ejecución anterior.
if nombre_campo == indicador_hidrico:: si encuentra un campo con ese mismo nombre, lo elimina primero (deleteAttributes) para evitar duplicados o valores de una ejecución previa.
localidades.updateFields(): refresca la lista de campos de la capa después de eliminar uno (es necesario llamarlo para que QGIS "vea" el cambio).
localidades_provider.addAttributes([QgsField(indicador_hidrico, QVariant.Double)]): agrega el nuevo campo a la tabla, indicando que su tipo de dato será decimal (QVariant.Double).

Sección 6 · Construir, validar y aplicar la expresión

            #Expresión con el cálculo del indicador
            exp = QgsExpression(f'({campo_de_tabla} / {total_VPH}) * 100')
            #Se valida la expresión
            validacion_expresion = QgsExpression.isValid(exp)
            print('¿Expresion válida?', validacion_expresion)
            
            #Si la expresión es válida, entonces de aplica en la tabla de atributos
            if validacion_expresion == True:
                contexto_capa = QgsExpressionContext()
                contexto_capa.appendScopes(QgsExpressionContextUtils.globalProjectLayerScopes(localidades))
                #Aplicación de la expresion
                with edit(localidades):
                    for renglon in localidades.getFeatures():
                        contexto_capa.setFeature(renglon)
                        renglon[indicador_hidrico] = exp.evaluate(contexto_capa)
                        localidades.updateFeature(renglon)
                print(f'Expresión aplicadada para calcular {indicador_hidrico}')
            else:
                print(f'No se pudo aplicar la expresión para calcular {indicador_hidrico}')
                
            #Señalar el progreso
            step += 1
            feedback.setCurrentStep(step)
            if feedback.isCanceled():
                return{}

exp = QgsExpression(f'({campo_de_tabla} / {total_VPH}) * 100'): construye, como texto, la fórmula del indicador (por ejemplo, '(VPH_AGUADV / TVIVPARHAB) * 100'), es decir, el porcentaje de viviendas con ese servicio.
validacion_expresion = QgsExpression.isValid(exp): revisa que la fórmula esté bien escrita (sin errores de sintaxis) antes de intentar usarla.
if validacion_expresion == True:: solo continúa con el cálculo si la expresión es válida; si no lo es, simplemente informa el error por consola y sigue con el siguiente indicador.
contexto_capa = QgsExpressionContext() y .appendScopes(...): preparan el "entorno" necesario para que la expresión pueda usar variables del proyecto y de la capa al evaluarse.
with edit(localidades):: abre una sesión de edición segura sobre la capa; todos los cambios hechos dentro de este bloque se confirman juntos al final (o se cancelan si algo falla).
for renglon in localidades.getFeatures():: recorre cada punto (cada localidad) de la capa, uno por uno.
contexto_capa.setFeature(renglon): le indica al contexto de la expresión sobre cuál entidad (fila) debe evaluarse en este momento.
renglon[indicador_hidrico] = exp.evaluate(contexto_capa): calcula el valor de la fórmula para esta localidad y lo guarda en el campo nuevo.
localidades.updateFeature(renglon): guarda el cambio hecho en esta entidad dentro de la sesión de edición.
step += 1 / feedback.setCurrentStep(step): avanza el contador de pasos y actualiza la barra de progreso después de terminar con un indicador completo.
if feedback.isCanceled(): return {}: permite cancelar la ejecución entre un indicador y el siguiente.

Sección 7 · Entregar el resultado y mostrar la capa

        #Regresar el resultado
        outputs['indicadores'] = parameters['capa_entrada']
        results['indicadores'] = outputs['indicadores']
        
        #Despliegue de la capa con el resultado
        if localidades.isValid() == True:
            QgsProject.instance().addMapLayer(localidades)
        else:
            print('No se pudo cargar la capa vectorial.')
            
        return results

outputs['indicadores'] = parameters['capa_entrada'] / results['indicadores'] = outputs['indicadores']: como no hay una capa de salida nueva, el resultado que se reporta es la referencia a la misma capa de entrada, ya modificada.
if localidades.isValid() == True:: comprueba que la capa siga siendo válida después de todos los cambios realizados.
QgsProject.instance().addMapLayer(localidades): agrega (o vuelve a mostrar) la capa ya actualizada en el panel de capas del proyecto de QGIS, para que el usuario pueda verla de inmediato con sus nuevos campos calculados.
return results: entrega el resultado final a QGIS, cerrando la ejecución de la herramienta.

Pon a prueba lo analizado

Volver a PASIG II