Merge pull request #617 from kartoza/timlinux/issue612

Fix poly classification for Digital Inclusion and Education

geest/core/generate_model.py

@@ -53,9 +53,10 @@ def load_spreadsheet(self):
                 "Use Mapillary Downloader",
                 "Use Other Downloader",
                 "Use Add Layers Manually",
-                "Use Classify Poly into Classes",
+                "Use Classify Polygon into Classes",
+                "Use Classify Safety Polygon into Classes",
                 "Use CSV to Point Layer",
-                "Use Poly per Cell",
+                "Use Polygon per Cell",
                 "Use Polyline per Cell",
                 "Use Point per Cell",
                 "Use Nighttime Lights",
@@ -74,7 +75,7 @@ def create_id(self, name):
         """
         Helper method to create a lowercase, underscore-separated id from the name.
         """
-        return name.lower().replace(" ", "_")
+        return name.lower().replace(" ", "_").replace("'", "_")
 
     def parse_to_json(self):
         """
@@ -242,19 +243,24 @@ def parse_to_json(self):
                     if not pd.isna(row["Use Add Layers Manually"])
                     else ""
                 ),
-                "use_classify_poly_into_classes": (
-                    row["Use Classify Poly into Classes"]
-                    if not pd.isna(row["Use Classify Poly into Classes"])
+                "use_classify_polygon_into_classes": (
+                    row["Use Classify Polygon into Classes"]
+                    if not pd.isna(row["Use Classify Polygon into Classes"])
+                    else ""
+                ),
+                "use_classify_safety_polygon_into_classes": (
+                    row["Use Classify Safety Polygon into Classes"]
+                    if not pd.isna(row["Use Classify Safety Polygon into Classes"])
                     else ""
                 ),
                 "use_csv_to_point_layer": (
                     row["Use CSV to Point Layer"]
                     if not pd.isna(row["Use CSV to Point Layer"])
                     else ""
                 ),
-                "use_poly_per_cell": (
-                    row["Use Poly per Cell"]
-                    if not pd.isna(row["Use Poly per Cell"])
+                "use_polygon_per_cell": (
+                    row["Use Polygon per Cell"]
+                    if not pd.isna(row["Use Polygon per Cell"])
                     else ""
                 ),
                 "use_polyline_per_cell": (

geest/core/workflow_factory.py

@@ -20,6 +20,7 @@
     SafetyRasterWorkflow,
     RasterReclassificationWorkflow,
     StreetLightsBufferWorkflow,
+    ClassifiedPolygonWorkflow,
 )
 
 from .json_tree_item import JsonTreeItem
@@ -79,7 +80,7 @@ def create_workflow(
             elif analysis_mode == "use_polyline_per_cell":
                 return PolylinePerCellWorkflow(item, cell_size_m, feedback, context)
             # TODO fix inconsistent abbreviation below for Poly
-            elif analysis_mode == "use_poly_per_cell":
+            elif analysis_mode == "use_polygon_per_cell":
                 return PolygonPerCellWorkflow(item, cell_size_m, feedback, context)
             elif analysis_mode == "factor_aggregation":
                 return FactorAggregationWorkflow(item, cell_size_m, feedback, context)
@@ -91,7 +92,9 @@ def create_workflow(
                 return AnalysisAggregationWorkflow(item, cell_size_m, feedback, context)
             elif analysis_mode == "use_csv_to_point_layer":
                 return AcledImpactWorkflow(item, cell_size_m, feedback, context)
-            elif analysis_mode == "use_classify_poly_into_classes":
+            elif analysis_mode == "use_classify_polygon_into_classes":
+                return ClassifiedPolygonWorkflow(item, cell_size_m, feedback, context)
+            elif analysis_mode == "use_classify_safety_polygon_into_classes":
                 return SafetyPolygonWorkflow(item, cell_size_m, feedback, context)
             elif analysis_mode == "use_nighttime_lights":
                 return SafetyRasterWorkflow(item, cell_size_m, feedback, context)

geest/core/workflows/__init__.py

@@ -13,3 +13,4 @@
 from .safety_raster_workflow import SafetyRasterWorkflow
 from .raster_reclassification_workflow import RasterReclassificationWorkflow
 from .street_lights_buffer_workflow import StreetLightsBufferWorkflow
+from .classified_polygon_workflow import ClassifiedPolygonWorkflow

geest/core/workflows/classified_polygon_workflow.py

@@ -0,0 +1,177 @@
+import os
+from qgis.core import (
+    Qgis,
+    QgsFeedback,
+    QgsGeometry,
+    QgsVectorLayer,
+    QgsProcessingContext,
+    edit,
+    QgsField,
+)
+from qgis.PyQt.QtCore import QVariant
+from .workflow_base import WorkflowBase
+from geest.core import JsonTreeItem
+from geest.utilities import log_message
+
+
+class ClassifiedPolygonWorkflow(WorkflowBase):
+    """
+    Concrete implementation of a 'use_classify_polygon_into_classes' workflow.
+    """
+
+    def __init__(
+        self,
+        item: JsonTreeItem,
+        cell_size_m: float,
+        feedback: QgsFeedback,
+        context: QgsProcessingContext,
+    ):
+        """
+        Initialize the workflow with attributes and feedback.
+        :param item: Item containing workflow parameters.
+        :param cell_size_m: Cell size in meters.
+        :param feedback: QgsFeedback object for progress reporting and cancellation.
+        :param context: QgsProcessingContext object for processing. This can be used to pass objects to the thread. e.g. the QgsProject Instance
+        """
+        super().__init__(
+            item, cell_size_m, feedback, context
+        )  # ⭐️ Item is a reference - whatever you change in this item will directly update the tree
+        self.workflow_name = "use_classify_polygon_into_classes"
+        layer_path = self.attributes.get(
+            "classify_polygon_into_classes_shapefile", None
+        )
+
+        if not layer_path:
+            log_message(
+                "Invalid layer found in use_classify_polygon_into_classes_shapefile, trying use_classify_polygon_into_classes_source.",
+                tag="Geest",
+                level=Qgis.Warning,
+            )
+            layer_path = self.attributes.get(
+                "classify_polygon_into_classes_layer_source", None
+            )
+            if not layer_path:
+                log_message(
+                    "No layer found in use_classify_polygon_into_classes_layer_source.",
+                    tag="Geest",
+                    level=Qgis.Warning,
+                )
+                return False
+
+        self.features_layer = QgsVectorLayer(layer_path, "features_layer", "ogr")
+
+        self.selected_field = self.attributes.get(
+            "classify_polygon_into_classes_selected_field", ""
+        )
+
+    def _process_features_for_area(
+        self,
+        current_area: QgsGeometry,
+        current_bbox: QgsGeometry,
+        area_features: QgsVectorLayer,
+        index: int,
+    ) -> str:
+        """
+        Executes the actual workflow logic for a single area
+        Must be implemented by subclasses.
+
+        :current_area: Current polygon from our study area.
+        :current_bbox: Bounding box of the above area.
+        :area_features: A vector layer of features to analyse that includes only features in the study area.
+        :index: Iteration / number of area being processed.
+
+        :return: A raster layer file path if processing completes successfully, False if canceled or failed.
+        """
+        area_features_count = area_features.featureCount()
+        log_message(
+            f"Features layer for area {index+1} loaded with {area_features_count} features.",
+            tag="Geest",
+            level=Qgis.Info,
+        )
+        # Step 1: Assign reclassification values based on perceived safety
+        reclassified_layer = self._assign_reclassification_to_safety(area_features)
+
+        # Step 2: Rasterize the data
+        raster_output = self._rasterize(
+            reclassified_layer,
+            current_bbox,
+            index,
+            value_field="value",
+            default_value=255,
+        )
+        return raster_output
+
+    def _assign_reclassification_to_safety(
+        self, layer: QgsVectorLayer
+    ) -> QgsVectorLayer:
+        """
+        Assign reclassification values to polygons based on thresholds.
+        """
+        with edit(layer):
+            # Remove all other columns except the selected field and the new 'value' field
+            fields_to_keep = {self.selected_field, "value"}
+            fields_to_remove = [
+                field.name()
+                for field in layer.fields()
+                if field.name() not in fields_to_keep
+            ]
+            layer.dataProvider().deleteAttributes(
+                [layer.fields().indexFromName(field) for field in fields_to_remove]
+            )
+            layer.updateFields()
+            if layer.fields().indexFromName("value") == -1:
+                layer.dataProvider().addAttributes([QgsField("value", QVariant.Int)])
+                layer.updateFields()
+
+            for feature in layer.getFeatures():
+                score = feature[self.selected_field]
+                # Scale values between 0 and 5
+                reclass_val = self._scale_value(score, 0, 100, 0, 5)
+                log_message(f"Scaled {score} to: {reclass_val}")
+                feature.setAttribute("value", reclass_val)
+                layer.updateFeature(feature)
+        return layer
+
+    def _scale_value(self, value, min_in, max_in, min_out, max_out):
+        """
+        Scale value from input range (min_in, max_in) to output range (min_out, max_out).
+        """
+        try:
+            result = (value - min_in) / (max_in - min_in) * (
+                max_out - min_out
+            ) + min_out
+            return result
+        except:
+            log_message(f"Invalid value, returning 0: {value}")
+            return 0
+
+    # Default implementation of the abstract method - not used in this workflow
+    def _process_raster_for_area(
+        self,
+        current_area: QgsGeometry,
+        current_bbox: QgsGeometry,
+        area_raster: str,
+        index: int,
+    ):
+        """
+        Executes the actual workflow logic for a single area using a raster.
+
+        :current_area: Current polygon from our study area.
+        :current_bbox: Bounding box of the above area.
+        :area_raster: A raster layer of features to analyse that includes only bbox pixels in the study area.
+        :index: Index of the current area.
+
+        :return: Path to the reclassified raster.
+        """
+        pass
+
+    def _process_aggregate_for_area(
+        self,
+        current_area: QgsGeometry,
+        current_bbox: QgsGeometry,
+        index: int,
+    ):
+        """
+        Executes the workflow, reporting progress through the feedback object and checking for cancellation.
+        """
+        pass

geest/core/workflows/polygon_per_cell_workflow.py

@@ -37,7 +37,7 @@ def __init__(
             item, cell_size_m, feedback, context
         )  # ⭐️ Item is a reference - whatever you change in this item will directly update the tree
         # TODO fix inconsistent abbreviation below for Poly
-        self.workflow_name = "use_poly_per_cell"
+        self.workflow_name = "use_polygon_per_cell"
 
         layer_path = self.attributes.get("polygon_per_cell_shapefile", None)
 

geest/core/workflows/safety_polygon_workflow.py

@@ -16,7 +16,7 @@
 
 class SafetyPolygonWorkflow(WorkflowBase):
     """
-    Concrete implementation of a 'use_classify_poly_into_classes' workflow.
+    Concrete implementation of a 'use_classify_polygon_into_classes' workflow.
     """
 
     def __init__(
@@ -36,21 +36,23 @@ def __init__(
         super().__init__(
             item, cell_size_m, feedback, context
         )  # ⭐️ Item is a reference - whatever you change in this item will directly update the tree
-        self.workflow_name = "use_classify_poly_into_classes"
-        layer_path = self.attributes.get("classify_poly_into_classes_shapefile", None)
+        self.workflow_name = "use_classify_safety_polygon_into_classes"
+        layer_path = self.attributes.get(
+            "classify_safety_polygon_into_classes_shapefile", None
+        )
 
         if not layer_path:
             log_message(
-                "Invalid raster found in classify_poly_into_classes_shapefile, trying classify_poly_into_classes_layer_source.",
+                "Invalid layer found in classify_safety_polygon_into_classes_shapefile, trying classify_safety_polygon_into_classes_layer_source.",
                 tag="Geest",
                 level=Qgis.Warning,
             )
             layer_path = self.attributes.get(
-                "classify_poly_into_classes_layer_source", None
+                "classify_safety_polygon_into_classes_layer_source", None
             )
             if not layer_path:
                 log_message(
-                    "No points layer found in classify_poly_into_classes_layer_source.",
+                    "No layer found in classify_safety_polygon_into_classes_layer_source.",
                     tag="Geest",
                     level=Qgis.Warning,
                 )
@@ -59,12 +61,12 @@ def __init__(
         self.features_layer = QgsVectorLayer(layer_path, "features_layer", "ogr")
 
         self.selected_field = self.attributes.get(
-            "classify_poly_into_classes_selected_field", ""
+            "classify_safety_polygon_into_classes_selected_field", ""
         )
         # This is a dict with keys being unique values from the selected field
         # and values from the aggregation dialog configuration table
         self.safety_mapping_table = self.attributes.get(
-            "classify_poly_into_classes_unique_values", None
+            "classify_safety_polygon_into_classes_unique_values", None
         )
         if not isinstance(self.safety_mapping_table, dict):
             raise Exception("Safety scoring table not configured.")

geest/gui/combined_widget_factory.py

@@ -13,6 +13,7 @@
     SafetyRasterWidget,
     RasterReclassificationWidget,
     StreetLightsWidget,
+    ClassifiedPolygonWidget,
 )
 from geest.core import setting
 from geest.utilities import log_message
@@ -49,15 +50,17 @@ def create_radio_button(
                 return MultiBufferDistancesWidget(label_text=key, attributes=attributes)
             if key == "use_single_buffer_point" and value == 1:
                 return SingleBufferDistanceWidget(label_text=key, attributes=attributes)
-            if key == "use_poly_per_cell" and value == 1:
+            if key == "use_polygon_per_cell" and value == 1:
                 return PolygonWidget(label_text=key, attributes=attributes)
             if key == "use_polyline_per_cell" and value == 1:
                 return PolylineWidget(label_text=key, attributes=attributes)
             if key == "use_point_per_cell" and value == 1:
                 return PointLayerWidget(label_text=key, attributes=attributes)
             if key == "use_csv_to_point_layer" and value == 1:
                 return AcledCsvLayerWidget(label_text=key, attributes=attributes)
-            if key == "use_classify_poly_into_classes" and value == 1:
+            if key == "use_classify_polygon_into_classes" and value == 1:
+                return ClassifiedPolygonWidget(label_text=key, attributes=attributes)
+            if key == "use_classify_safety_polygon_into_classes" and value == 1:
                 return SafetyPolygonWidget(label_text=key, attributes=attributes)
             if key == "use_nighttime_lights" and value == 1:
                 return SafetyRasterWidget(label_text=key, attributes=attributes)

geest/gui/configuration_widget_factory.py

@@ -1,16 +1,17 @@
 from qgis.core import Qgis
 from geest.gui.widgets.configuration_widgets import (
+    AcledCsvConfigurationWidget,
     BaseConfigurationWidget,
+    ClassifiedPolygonConfigurationWidget,
     DontUseConfigurationWidget,
-    AcledCsvConfigurationWidget,
+    FeaturePerCellConfigurationWidget,
     IndexScoreConfigurationWidget,
     MultiBufferConfigurationWidget,
-    SingleBufferConfigurationWidget,
-    FeaturePerCellConfigurationWidget,
-    SafetyPolygonConfigurationWidget,
-    StreetLightsConfigurationWidget,
     RasterReclassificationConfigurationWidget,
+    SafetyPolygonConfigurationWidget,
     SafetyRasterConfigurationWidget,
+    SingleBufferConfigurationWidget,
+    StreetLightsConfigurationWidget,
 )
 from geest.core import setting
 from geest.utilities import log_message
@@ -57,7 +58,7 @@ def create_radio_button(
             # ------------------------------------------------
             # These three all use the same configuration widgets
             # but will have different datasource widgets generated as appropriate
-            if key == "use_poly_per_cell" and value == 1:  # poly = polygon
+            if key == "use_polygon_per_cell" and value == 1:  # poly = polygon
                 return FeaturePerCellConfigurationWidget(
                     analysis_mode=key, attributes=attributes
                 )
@@ -74,7 +75,11 @@ def create_radio_button(
                     analysis_mode=key, attributes=attributes
                 )
             # ------------------------------------------------
-            if key == "use_classify_poly_into_classes" and value == 1:
+            if key == "use_classify_polygon_into_classes" and value == 1:
+                return ClassifiedPolygonConfigurationWidget(
+                    analysis_mode=key, attributes=attributes
+                )
+            if key == "use_classify_safety_polygon_into_classes" and value == 1:
                 return SafetyPolygonConfigurationWidget(
                     analysis_mode=key, attributes=attributes
                 )