Merge pull request #734 from kartoza/timlinux/issue702

Fix tests and styling for aggregates

geest/core/algorithms/subnational_aggregation_processor.py

@@ -103,7 +103,9 @@ def __init__(
             "ogr",
         )
         if not self.aggregation_layer.isValid():
-            raise Exception("Invalid aggregation areas layer.")
+            raise Exception(
+                f"Invalid aggregation areas layer:\n{self.aggregation_areas_path}"
+            )
 
         self.output_dir = os.path.join(working_directory, "subnational_aggregation")
         os.makedirs(self.output_dir, exist_ok=True)
@@ -114,11 +116,11 @@ def __init__(
 
         if not os.path.exists(self.population_folder):
             raise Exception(
-                "Population folder not found. Please run population raster processing first."
+                f"Population folder not found:\n{self.population_folder}\nPlease run population raster processing first."
             )
         if not os.path.exists(self.wee_folder):
             raise Exception(
-                "WEE folder not found. Please run WEE raster processing first."
+                f"WEE folder not found.\n{self.wee_folder}\nPlease run WEE raster processing first."
             )
 
         self.force_clear = force_clear
@@ -134,6 +136,10 @@ def __init__(
                 "ogr",
             )
             self.target_crs = layer.crs()
+            log_message(
+                f"Target CRS not set. Using CRS from study area clip polygon: {self.target_crs.authid()}"
+            )
+            log_message(f"{self.study_area_gpkg_path}|ayername=study_area_clip_polygon")
             del layer
 
         log_message("Initialized WEE Subnational Area Aggregation Processing Task")
@@ -145,10 +151,10 @@ def run(self) -> bool:
         try:
             self.aggregate()
             self.apply_qml_style(
-                source_qml=resources_path("qml/wee_by_population_vector_score.qml.qml"),
-                qml_path=os.path.join(
-                    self.output_dir, "wee_by_population_vector_score.qml"
+                source_qml=resources_path(
+                    "resources", "qml", "wee_by_population_vector_score.qml"
                 ),
+                qml_path=os.path.join(self.output_dir, "subnational_aggregation.qml"),
             )
             return True
         except Exception as e:
@@ -180,10 +186,10 @@ def aggregate(self) -> None:
 
     def apply_qml_style(self, source_qml: str, qml_path: str) -> None:
 
+        log_message(f"Copying QML style from {source_qml} to {qml_path}")
         # Apply QML Style
         if os.path.exists(source_qml):
             shutil.copy(source_qml, qml_path)
-            log_message(f"Copied QML style from {source_qml} to {qml_path}")
         else:
             log_message("QML style file not found. Skipping QML copy.")
 

geest/core/workflows/analysis_aggregation_workflow.py

@@ -79,7 +79,7 @@ def __init__(
             force_clear=False,
         )
         subnational_processor.run()
-        # Shamelessly hard coded for now, needs to move to the wee processor class
+        # Shamelessly hard coded for now, needs to move to the aggregation processor class
         output = os.path.join(
             self.working_directory,
             "subnational_aggregation",

geest/gui/panels/tree_panel.py

@@ -888,7 +888,7 @@ def add_study_area_to_map(self):
                     f"Added layer: {layer.name()} to group: {geest_group.name()}"
                 )
 
-    def add_to_map(self, item, key="result_file", layer_name=None):
+    def add_to_map(self, item, key="result_file", layer_name=None, qml_key=None):
         """Add the item to the map."""
 
         layer_uri = item.attribute(f"{key}")
@@ -900,6 +900,10 @@ def add_to_map(self, item, key="result_file", layer_name=None):
             if "gpkg" in layer_uri:
                 log_message(f"Adding GeoPackage layer: {layer_name}")
                 layer = QgsVectorLayer(layer_uri, layer_name, "ogr")
+                if qml_key:
+                    qml_path = item.attribute(qml_key)
+                    if qml_path:
+                        result = layer.loadNamedStyle(qml_path)
             else:
                 log_message(f"Adding raster layer: {layer_name}")
                 layer = QgsRasterLayer(layer_uri, layer_name)

geest/resources/icons/credits.txt

@@ -16,3 +16,4 @@ Factor and Indicator icons were further modified by Tim Sutton, Oct 2024.
 
 queued.svg - original art by Tim Sutton
 
+run-tests.svg cog icon: https://www.svgrepo.com/svg/348985/cog

test/test_subnational_aggregation.py

@@ -0,0 +1,53 @@
+import os
+import unittest
+from qgis.core import (
+    QgsVectorLayer,
+    QgsProcessingContext,
+    QgsFeedback,
+)
+from geest.core.tasks import (
+    StudyAreaProcessingTask,
+)  # Adjust the import path as necessary
+from utilities_for_testing import prepare_fixtures
+from geest.core.algorithms import SubnationalAggregationProcessingTask
+
+
+class TestSubnationalAggregationProcessingTask(unittest.TestCase):
+
+    @classmethod
+    def setUpClass(cls):
+        """Set up shared resources for the test suite."""
+
+        cls.working_directory = os.path.join(prepare_fixtures(), "wee_score")
+        cls.context = QgsProcessingContext()
+        cls.feedback = QgsFeedback()
+        cls.aggregation_areas_path = os.path.join(
+            cls.working_directory, "aggregation", "boundaries.gpkg|layername=boundaries"
+        )
+        cls.study_area_gpkg_path = os.path.join(
+            cls.working_directory, "study_area", "study_area.gpkg"
+        )
+
+    def setUp(self):
+        self.task = SubnationalAggregationProcessingTask(
+            # geest_raster_path=f"{self.working_directory}/wee_masked_0.tif",
+            # pop_raster_path=f"{self.working_directory}/population/reclassified_0.tif",
+            study_area_gpkg_path=self.study_area_gpkg_path,
+            aggregation_areas_path=self.aggregation_areas_path,
+            working_directory=self.working_directory,
+            target_crs=None,
+            force_clear=True,
+        )
+
+    def test_initialization(self):
+        self.assertTrue(
+            self.task.output_dir.endswith("subnational_aggregation"),
+            msg=f"Output directory is {self.task.output_dir}",
+        )
+        self.assertEqual(self.task.target_crs.authid(), "EPSG:32620")
+
+    def test_run_task(self):
+        result = self.task.run()
+        self.assertTrue(
+            result, msg=f"Subnational Aggregation failed in {self.working_directory}"
+        )