Merge pull request #329 from kartoza/timlinux/issue286

Timlinux/issue286

.gitignore

@@ -72,4 +72,4 @@ app.py
 /.idea/modules.xml
 /.idea/inspectionProfiles/profiles_settings.xml
 /.idea/vcs.xml
-core
+

geest/core/crs_converter.py

@@ -1,6 +1,12 @@
-from qgis.core import QgsCoordinateTransform, QgsCoordinateReferenceSystem, QgsProcessingFeedback, QgsVectorLayer
+from qgis.core import (
+    QgsCoordinateTransform,
+    QgsCoordinateReferenceSystem,
+    QgsProcessingFeedback,
+    QgsVectorLayer,
+)
 from qgis import processing
 
+
 class CRSConverter:
     def __init__(self, layer):
         """
@@ -22,7 +28,9 @@ def convert_to_crs(self, target_crs_epsg):
 
         # Check if the current CRS is the same as the target CRS
         if current_crs != target_crs:
-            print(f"Converting layer from {current_crs.authid()} to {target_crs.authid()}")
+            print(
+                f"Converting layer from {current_crs.authid()} to {target_crs.authid()}"
+            )
 
             layer = processing.run(
                 "native:reprojectlayer",

geest/core/workflow_factory.py

@@ -1,4 +1,4 @@
-from .workflows import RasterLayerWorkflow, DontUseWorkflow
+from .workflows import RasterLayerWorkflow, DontUseWorkflow, DefaultIndexScoreWorkflow
 from qgis.core import QgsFeedback
 
 
@@ -17,6 +17,8 @@ def create_workflow(self, attributes, feedback: QgsFeedback):
 
         if analysis_mode == "Spatial Analysis":
             return RasterLayerWorkflow(attributes, feedback)
+        elif analysis_mode == "Use Default Index Score":
+            return DefaultIndexScoreWorkflow(attributes, feedback)
         elif analysis_mode == "Don’t Use":
             return DontUseWorkflow(attributes, feedback)
         elif analysis_mode == "Temporal Analysis":

geest/core/workflow_job.py

@@ -35,36 +35,36 @@ def run(self) -> bool:
         if not self._workflow:
             QgsMessageLog.logMessage(
                 f"Error: No workflow assigned to {self.description()}",
-                "Custom Workflows",
-                Qgis.Critical,
+                tag="Geest",
+                level=Qgis.Critical,
             )
             return False
 
         try:
             QgsMessageLog.logMessage(
-                f"Running workflow: {self.description()}", "Custom Workflows", Qgis.Info
+                f"Running workflow: {self.description()}", tag="Geest", level=Qgis.Info
             )
 
             result = self._workflow.execute()
 
             if result:
                 QgsMessageLog.logMessage(
                     f"Workflow {self.description()} completed.",
-                    "Custom Workflows",
-                    Qgis.Info,
+                    tag="Geest",
+                    level=Qgis.Info,
                 )
                 return True
             else:
                 QgsMessageLog.logMessage(
                     f"Workflow {self.description()} did not complete successfully.",
-                    "Custom Workflows",
-                    Qgis.Warning,
+                    tag="Geest",
+                    level=Qgis.Info,
                 )
                 return False
 
         except Exception as e:
             QgsMessageLog.logMessage(
-                f"Error during task execution: {e}", "Custom Workflows", Qgis.Critical
+                f"Error during task execution: {e}", tag="Geest", level=Qgis.Critical
             )
             return False
 

geest/core/workflow_queue_manager.py

@@ -33,27 +33,27 @@ def add_task(self, attributes: dict) -> None:
         task = WorkflowJob(description="Workflow Task", attributes=attributes)
         self.workflow_queue.add_job(task)
         QgsMessageLog.logMessage(
-            f"Task added: {task.description()}", "Workflow Manager", Qgis.Info
+            f"Task added: {task.description()}", tag="Geest", level=Qgis.Info
         )
 
     def start_processing(self) -> None:
         """Start processing the tasks in the WorkflowQueue."""
         QgsMessageLog.logMessage(
-            "Starting workflow queue processing...", "Workflow Manager", Qgis.Info
+            "Starting workflow queue processing...", tag="Geest", level=Qgis.Info
         )
         self.workflow_queue.start_processing()
 
     def cancel_processing(self) -> None:
         """Cancels all tasks in the WorkflowQueue."""
         QgsMessageLog.logMessage(
-            "Cancelling workflow queue...", "Workflow Manager", Qgis.Warning
+            "Cancelling workflow queue...", tag="Geest", level=Qgis.Info
         )
         self.workflow_queue.cancel_processing()
 
     def update_status(self) -> None:
         """Update the status of the workflow queue (for UI updates, etc.)."""
         QgsMessageLog.logMessage(
-            "Workflow queue status updated.", "Workflow Manager", Qgis.Info
+            "Workflow queue status updated.", tag="Geest", level=Qgis.Info
         )
 
     def on_processing_completed(self, success: bool) -> None:
@@ -64,17 +64,17 @@ def on_processing_completed(self, success: bool) -> None:
         if success:
             QgsMessageLog.logMessage(
                 "All workflow tasks completed successfully.",
-                "Workflow Manager",
-                Qgis.Success,
+                tag="Geest",
+                level=Qgis.Info,
             )
         else:
             QgsMessageLog.logMessage(
-                "Workflow processing was canceled.", "Workflow Manager", Qgis.Warning
+                "Workflow processing was canceled.", tag="Geest", level=Qgis.Info
             )
 
     def log_status_message(self, message: str) -> None:
         """
         Logs status messages from the WorkflowQueue.
         :param message: Status message to log
         """
-        QgsMessageLog.logMessage(message, "Workflow Manager", Qgis.Info)
+        QgsMessageLog.logMessage(message, tag="Geest", level=Qgis.Info)

geest/core/workflows/__init__.py

@@ -1,2 +1,3 @@
 from .raster_layer_workflow import RasterLayerWorkflow
 from .dont_use_workflow import DontUseWorkflow
+from .default_index_score_workflow import DefaultIndexScoreWorkflow

geest/core/workflows/default_index_score_workflow.py

@@ -0,0 +1,150 @@
+import os
+from qgis.core import (
+    QgsMessageLog,
+    Qgis,
+    QgsFeedback,
+    QgsFeature,
+    QgsVectorLayer,
+    QgsField,
+    QgsGeometry,
+    QgsRectangle,
+)
+from qgis.PyQt.QtCore import QVariant
+import processing  # QGIS processing toolbox
+from .workflow_base import WorkflowBase
+
+
+class DefaultIndexScoreWorkflow(WorkflowBase):
+    """
+    Concrete implementation of a 'Use Default Index Score' workflow.
+    """
+
+    def __init__(self, attributes: dict, feedback: QgsFeedback):
+        """
+        Initialize the TemporalAnalysisWorkflow with attributes and feedback.
+        :param attributes: Dictionary containing workflow parameters.
+        :param feedback: QgsFeedback object for progress reporting and cancellation.
+        """
+        super().__init__(attributes, feedback)
+
+    def execute(self):
+        """
+        Executes the workflow, reporting progress through the feedback object and checking for cancellation.
+        """
+
+        QgsMessageLog.logMessage(
+            "Executing Use Default Index Score", tag="Geest", level=Qgis.Info
+        )
+        QgsMessageLog.logMessage(
+            "----------------------------------", tag="Geest", level=Qgis.Info
+        )
+        for item in self.attributes.items():
+            QgsMessageLog.logMessage(
+                f"{item[0]}: {item[1]}", tag="Geest", level=Qgis.Info
+            )
+        QgsMessageLog.logMessage(
+            "----------------------------------", tag="Geest", level=Qgis.Info
+        )
+
+        # loop through self.bboxes_layer and the self.areas_layer  and create a raster mask for each feature
+        index_score = self.attributes["Default Index Score"]
+        for feature in self.bboxes_layer.getFeatures():
+            geom = feature.geometry()  # todo this shoudl come from the areas layer
+            aligned_box = geom
+            mask_name = f"bbox_{feature.id()}"
+            self.create_raster(
+                geom=geom,
+                aligned_box=aligned_box,
+                mask_name=mask_name,
+                index_score=index_score,
+            )
+        # TODO Jeff copy create_raster_vrt from study_area.py
+
+        steps = 10
+        for i in range(steps):
+            if self.feedback.isCanceled():
+                QgsMessageLog.logMessage(
+                    "Dont use workflow canceled.", tag="Geest", level=Qgis.Warning
+                )
+                return False
+
+            # Simulate progress and work
+            self.attributes["progress"] = f"Dont use workflow Step {i + 1} completed"
+            self.feedback.setProgress(
+                (i + 1) / steps * 100
+            )  # Report progress in percentage
+            QgsMessageLog.logMessage(
+                f"Assigning index score: {self.attributes['Default Index Score']}",
+                tag="Geest",
+                level=Qgis.Info,
+            )
+
+        self.attributes["result"] = "Use Default Index Score Workflow Completed"
+        QgsMessageLog.logMessage(
+            "Use Default Index Score workflow workflow completed",
+            tag="Geest",
+            level=Qgis.Info,
+        )
+        return True
+
+    def create_raster(
+        self,
+        geom: QgsGeometry,
+        aligned_box: QgsGeometry,
+        mask_name: str,
+        index_score: float,
+    ) -> None:
+        """
+        Creates a byte raster mask for a single geometry.
+
+        :param geom: Geometry to be rasterized.
+        :param aligned_box: Aligned bounding box geometry for the geometry.
+        :param mask_name: Name for the output raster file.
+        """
+        aligned_box = QgsRectangle(aligned_box.boundingBox())
+        mask_filepath = os.path.join(self.workflow_directory, f"{mask_name}.tif")
+
+        # Create a memory layer to hold the geometry
+        temp_layer = QgsVectorLayer(
+            f"Polygon?crs={self.output_crs.authid()}", "temp_mask_layer", "memory"
+        )
+        temp_layer_data_provider = temp_layer.dataProvider()
+
+        # Define a field to store the mask value
+        temp_layer_data_provider.addAttributes([QgsField("area_name", QVariant.String)])
+        temp_layer.updateFields()
+
+        # Add the geometry to the memory layer
+        temp_feature = QgsFeature()
+        temp_feature.setGeometry(geom)
+        temp_feature.setAttributes(["1"])  # Setting an arbitrary value for the mask
+        temp_layer_data_provider.addFeature(temp_feature)
+
+        # Ensure resolution parameters are properly formatted as float values
+        x_res = 100.0  # 100m pixel size in X direction
+        y_res = 100.0  # 100m pixel size in Y direction
+
+        # Define rasterization parameters for the temporary layer
+        params = {
+            "INPUT": temp_layer,
+            "FIELD": None,
+            "BURN": 78,  # todo Jeff put on likert scale properly
+            "USE_Z": False,
+            "UNITS": 1,
+            "WIDTH": x_res,
+            "HEIGHT": y_res,
+            "EXTENT": f"{aligned_box.xMinimum()},{aligned_box.xMaximum()},"
+            f"{aligned_box.yMinimum()},{aligned_box.yMaximum()}",  # Extent of the aligned bbox
+            "NODATA": 0,
+            "OPTIONS": "",
+            "DATA_TYPE": 0,  # byte
+            "INIT": None,
+            "INVERT": False,
+            "EXTRA": "",
+            "OUTPUT": mask_filepath,
+        }
+        # Run the rasterize algorithm
+        processing.run("gdal:rasterize", params)
+        QgsMessageLog.logMessage(
+            f"Created raster mask: {mask_filepath}", tag="Geest", level=Qgis.Info
+        )

geest/core/workflows/workflow_base.py

@@ -1,5 +1,7 @@
+import os
 from abc import ABC, abstractmethod
-from qgis.core import QgsFeedback
+from qgis.core import QgsFeedback, QgsVectorLayer
+from qgis.PyQt.QtCore import QSettings
 
 
 class WorkflowBase(ABC):
@@ -16,6 +18,28 @@ def __init__(self, attributes: dict, feedback: QgsFeedback):
         """
         self.attributes = attributes
         self.feedback = feedback
+        # This is set in the setup panel
+        self.settings = QSettings()
+        # This is the top level folder for work files
+        self.working_directory = self.settings.value("last_working_directory", "")
+        if not self.working_directory:
+            raise ValueError("Working directory not set.")
+        # This is the lower level directory for this workflow
+        self.workflow_directory = self._create_workflow_directory()
+        self.gpkg_path: str = os.path.join(
+            self.working_directory, "study_area", "study_area.gpkg"
+        )
+        if not os.path.exists(self.gpkg_path):
+            raise ValueError(f"Study area geopackage not found at {self.gpkg_path}.")
+        self.bboxes_layer = QgsVectorLayer(
+            f"{self.gpkg_path}|layername=study_area_bboxes", "study_area_bboxes", "ogr"
+        )
+        self.areas_layer = QgsVectorLayer(
+            f"{self.gpkg_path}|layername=study_area_polygons",
+            "study_area_polygons",
+            "ogr",
+        )
+        self.output_crs = self.bboxes_layer.crs()
 
     @abstractmethod
     def execute(self) -> bool:
@@ -24,3 +48,30 @@ def execute(self) -> bool:
         :return: True if the workflow completes successfully, False if canceled or failed.
         """
         pass
+
+    def _create_workflow_directory(self) -> str:
+        """
+        Creates the directory for this workflow if it doesn't already exist.
+        It will be in the scheme of working_dir/dimension/factor/indicator
+
+        :return: The path to the workflow directory
+        """
+        workflow_dir = os.path.join(
+            self.working_directory,
+            "contextual",
+            "workplace_discrimination",
+            "wbl_2024_workplace_index_score",
+        )
+        if not os.path.exists(workflow_dir):
+            try:
+                os.makedirs(workflow_dir)
+                QgsMessageLog.logMessage(
+                    f"Created study area directory: {workflow_dir}",
+                    tag="Geest",
+                    level=Qgis.Info,
+                )
+            except Exception as e:
+                QgsMessageLog.logMessage(
+                    f"Error creating directory: {e}", tag="Geest", level=Qgis.Critical
+                )
+        return workflow_dir