Merge pull request #60081 from JanCaha/fix-postgreraster-grass-provider

Allow postgresraster layers as grass processing tools input

python/plugins/grassprovider/grass_algorithm.py

@@ -923,9 +923,35 @@ def loadRasterLayer(
         if not destName:
             destName = f"rast_{os.path.basename(getTempFilename(context=context))}"
         self.exportedLayers[name] = destName
+
+        if layer.providerType() == "postgresraster":
+            source = ""
+            uri = layer.dataProvider().uri()
+            source = f"PG: {uri.connectionInfo()}"
+            schema = uri.schema()
+            if schema:
+                source += f" schema='{schema}'"
+            table = uri.table()
+            source += f" table='{table}'"
+            column = uri.param("column") or uri.geometryColumn()
+            if column:
+                source += f" column='{column}'"
+            is_tiled = any(
+                [
+                    layer.dataProvider().xSize() != layer.dataProvider().xBlockSize(),
+                    layer.dataProvider().ySize() != layer.dataProvider().yBlockSize(),
+                ]
+            )
+            source += f" mode={2 if is_tiled else 1}"
+            where = layer.dataProvider().subsetString()
+            if where:
+                source += f" where='{where}'"
+        else:
+            source = os.path.normpath(layer.source())
+
         command = '{} input="{}" {}output="{}" --overwrite -o'.format(
             "r.external" if external else "r.in.gdal",
-            os.path.normpath(layer.source()),
+            source,
             f"band={band} " if band else "",
             destName,
         )

python/plugins/grassprovider/tests/CMakeLists.txt

@@ -8,4 +8,5 @@ if(ENABLE_TESTS)
   ADD_PYTHON_TEST(ProcessingGrassAlgorithmsRasterTestPt1 grass_algorithms_raster_test_pt1.py)
   ADD_PYTHON_TEST(ProcessingGrassAlgorithmsRasterTestPt2 grass_algorithms_raster_test_pt2.py)
   ADD_PYTHON_TEST(ProcessingGrassAlgorithmsVectorTest grass_algorithms_vector_test.py)
+  ADD_PYTHON_TEST(ProcessingGrassAlgsPostgreRasterProvider grass_algorithm_postgreraster_test.py)
 endif()

python/plugins/grassprovider/tests/grass_algorithm_postgreraster_test.py

@@ -0,0 +1,147 @@
+"""QGIS Unit tests for Grass Algorithm with postgreraster provider
+
+.. note:: This program is free software; you can redistribute it and/or modify
+it under the terms of the GNU General Public License as published by
+the Free Software Foundation; either version 2 of the License, or
+(at your option) any later version.
+"""
+
+__author__ = "Jan Caha"
+__date__ = "01/08/2025"
+__copyright__ = "Copyright 2025, The QGIS Project"
+
+
+import os
+import tempfile
+from shutil import rmtree
+
+import unittest
+from qgis.testing import start_app, QgisTestCase
+from qgis.core import (
+    QgsApplication,
+    QgsRasterLayer,
+    QgsDataSourceUri,
+    QgsAuthMethodConfig,
+    QgsProcessingContext,
+    QgsProcessingFeedback,
+)
+from qgis import processing
+from grassprovider.grass_provider import GrassProvider
+from grassprovider.grass_utils import GrassUtils
+
+QGIS_AUTH_DB_DIR_PATH = tempfile.mkdtemp()
+os.environ["QGIS_AUTH_DB_DIR_PATH"] = QGIS_AUTH_DB_DIR_PATH
+
+start_app()
+
+
+class TestProcessingGrassAlgsPostgreRasterProvider(QgisTestCase):
+
+    @classmethod
+    def setUpClass(cls):
+        super().setUpClass()
+        cls.provider = GrassProvider()
+        QgsApplication.processingRegistry().addProvider(cls.provider)
+        cls.cleanup_paths = []
+
+        assert GrassUtils.installedVersion()
+
+    @classmethod
+    def tearDownClass(cls):
+        """Run after all tests"""
+        rmtree(QGIS_AUTH_DB_DIR_PATH)
+        del os.environ["QGIS_AUTH_DB_DIR_PATH"]
+        super().tearDownClass()
+
+    def test_algorithm_r_buffer(self):
+        """
+        Test grass algorithm r.buffer with postgreraster provider
+        """
+
+        context = QgsProcessingContext()
+        feedback = QgsProcessingFeedback()
+
+        rl = QgsRasterLayer(
+            "dbname='mydb' host=localhost port=5432 user='asdf' password='42'"
+            " sslmode=disable table=some_table schema=some_schema column=rast sql=pk = 2",
+            "pg_layer",
+            "postgresraster",
+        )
+
+        alg = QgsApplication.processingRegistry().algorithmById("grass:r.buffer")
+
+        self.assertTrue(alg)
+
+        params = {
+            "input": rl,
+            "distances": "100,200,500",
+            "units": 0,
+            "-z": False,
+            "output": "TEMPORARY_OUTPUT",
+            "GRASS_REGION_PARAMETER": None,
+            "GRASS_REGION_CELLSIZE_PARAMETER": 0,
+            "GRASS_RASTER_FORMAT_OPT": "",
+            "GRASS_RASTER_FORMAT_META": "",
+        }
+
+        can_run, _ = alg.checkParameterValues(parameters=params, context=context)
+
+        self.assertTrue(can_run)
+
+        res = alg.run(params, context, feedback)
+
+        # should be tuple
+        self.assertTrue(res)
+        # should be true if algorithm run successfully
+        self.assertTrue(res[1])
+        # should be dict with output keys
+        self.assertTrue(isinstance(res[0], dict))
+        # should be path to result file
+        self.assertTrue(isinstance(res[0]["output"], str))
+
+    def test_algorithm_r_info(self):
+        """
+        Test grass algorithm r.info with postgreraster provider
+        """
+
+        context = QgsProcessingContext()
+        feedback = QgsProcessingFeedback()
+
+        rl = QgsRasterLayer(
+            "dbname='mydb' host=localhost port=5432 user='asdf' password='42'"
+            " sslmode=disable table=some_table schema=some_schema column=rast sql=pk = 2",
+            "pg_layer",
+            "postgresraster",
+        )
+
+        alg = QgsApplication.processingRegistry().algorithmById("grass:r.info")
+
+        params = {
+            "map": rl,
+            "-r": False,
+            "-g": False,
+            "-h": False,
+            "-e": False,
+            "html": "./report.html",
+            "GRASS_REGION_PARAMETER": None,
+            "GRASS_REGION_CELLSIZE_PARAMETER": 0,
+        }
+
+        can_run, _ = alg.checkParameterValues(parameters=params, context=context)
+
+        self.assertTrue(can_run)
+
+        res = alg.run(params, context, feedback)
+
+        # should be tuple
+        self.assertTrue(res)
+        # should be true if algorithm run successfully
+        self.assertTrue(res[1])
+        # should be dict with output keys
+        self.assertTrue(isinstance(res[0], dict))
+        # should be path to result file
+        self.assertTrue(isinstance(res[0]["html"], str))
+
+
+if __name__ == "__main__":
+    unittest.main()