pygrass: add get_json_dict function

python/grass/pygrass/modules/interface/module.py

@@ -1,5 +1,7 @@
-from multiprocessing import cpu_count, Process, Queue
 import time
+
+from multiprocessing import cpu_count, Process, Queue
+from itertools import zip_longest
 from xml.etree.ElementTree import fromstring
 
 from grass.exceptions import CalledModuleError, GrassError, ParameterError
@@ -12,8 +14,6 @@
 from .read import GETFROMTAG, DOC
 from .env import G_debug
 
-from itertools import zip_longest
-
 
 def _get_bash(self, *args, **kargs):
     return self.get_bash()
@@ -569,9 +569,9 @@ def __init__(self, cmd, *args, **kargs):
             # call the command with --interface-description
             get_cmd_xml = Popen([cmd, "--interface-description"], stdout=PIPE)
         except OSError as e:
-            print("OSError error({0}): {1}".format(e.errno, e.strerror))
-            str_err = "Error running: `%s --interface-description`."
-            raise GrassError(str_err % self.name) from e
+            print(f"OSError error({e.errno}): {e.strerror}")
+            str_err = f"Error running: `{self.name} --interface-description`."
+            raise GrassError(str_err) from e
         # get the xml of the module
         self.xml = get_cmd_xml.communicate()[0]
         # transform and parse the xml into an Element class:
@@ -702,7 +702,8 @@ def update(self, *args, **kargs):
                 # verbose and quiet) work like parameters
                 self.flags[key].value = val
             else:
-                raise ParameterError("%s is not a valid parameter." % key)
+                msg = "{} is not a valid parameter.".format(key)
+                raise ParameterError(msg)
 
     def get_bash(self):
         """Return a BASH representation of the Module."""
@@ -781,6 +782,117 @@ def check(self):
                     msg = "Required parameter <%s> not set."
                     raise ParameterError(msg % k)
 
+    def get_json_dict(
+        self,
+        export: str | None = None,
+        stdout_export: str | None = None,
+        stdout_id: str = "stdout",
+        stdout_delimiter: str = "|",
+    ) -> dict:
+        """Return a dictionary with the module represented in JSON format.
+
+        The dictionary includes the module name, an id, all valid
+        inputs, outputs and flags as well as export settings for
+        usage with actinia
+        param export: string with export format for non-stdout output, one of
+                      "GTiff", "COG", for raster;
+                      "strds" for SpaceTimeRasterDatasets,
+                      "PostgreSQL", "GPKG", "GML", "GeoJSON", "ESRI_Shapefile",
+                      "SQLite" for vector and
+                      "CSV", "TXT" for files
+        param stdout_export: string with export format for stdout output, one of
+                             "table", "list", "kv", "json"
+        param stdout_id: unique string with "id" for stdout output of the module
+                         defaults to "stdout"
+        param stdout_delimiter: string with single delimiter, defaults to "|"
+        """
+        import uuid
+
+        export_dict = {
+            "GTiff": "raster",
+            "COG": "raster",
+            "strds": "GTiff",  # (multiple files packed in an tar.gz archive)
+            "PostgreSQL": "vector",
+            "GPKG": "vector",
+            "GML": "vector",
+            "GeoJSON": "vector",
+            "ESRI_Shapefile": "vector",
+            "SQLite": "vector",
+            "CSV": "file",
+            "TXT": "file",
+        }
+        stdout_export_formats = ["table", "list", "kv", "json"]
+        special_flags = ["overwrite", "verbose", "quiet"]
+        skip = ["stdin", "stdout", "stderr"]
+
+        # Check export formats
+        if export and export not in export_dict:
+            msg = f"Invalid export format <{export}>."
+            raise GrassError(msg)
+
+        # Handle inputs and flags
+        json_dict = {
+            "module": self.name,
+            "id": f"{self.name.replace('.', '_')}_{uuid.uuid4().hex}",
+            "flags": "".join(
+                [
+                    flg
+                    for flg in self.flags
+                    if self.flags[flg].value and flg not in {*special_flags, "help"}
+                ]
+            ),
+            "inputs": [
+                {
+                    "param": key,
+                    "value": (
+                        ",".join(map(str, val.value))
+                        if isinstance(val.value, list)
+                        else str(val.value)
+                    ),
+                }
+                for key, val in self.inputs.items()
+                if val.value and key not in skip
+            ],
+        }
+
+        # Handle special flags
+        for special_flag in special_flags:
+            if special_flag in self.flags:
+                json_dict[special_flag] = self.flags[special_flag].value
+
+        # Handle outputs
+        outputs = []
+        for key, val in self.outputs.items():
+            if val.value:
+                param = {
+                    "param": key,
+                    "value": (
+                        ",".join(map(str, val.value))
+                        if isinstance(val.value, list)
+                        else str(val.value)
+                    ),
+                }
+                if export:
+                    param["export"] = {
+                        "format": export,
+                        "type": export_dict[export],
+                    }
+                outputs.append(param)
+        json_dict["outputs"] = outputs
+
+        # Handle stdout
+        if stdout_export is not None:
+            if stdout_export not in stdout_export_formats:
+                msg = f"Invalid export format <{stdout_export}> for stdout."
+                raise GrassError(msg)
+            json_dict["stdout"] = {
+                "id": stdout_id,
+                "format": stdout_export,
+                "delimiter": stdout_delimiter,
+            }
+
+        return {key: val for key, val in json_dict.items() if val}
+
     def get_dict(self):
         """Return a dictionary that includes the name, all valid
         inputs, outputs and flags

python/grass/pygrass/modules/interface/testsuite/pygrass_modules_interface_json_test.py

@@ -0,0 +1,116 @@
+############################################################################
+#
+# MODULE:       Test of JSON export from pygrass.Module for usage in actinia
+# AUTHOR(S):    Stefan Blumentrath
+# PURPOSE:      Test of JSON export from pygrass.Module for usage in actinia
+# COPYRIGHT:    (C) 2024 by Stefan Blumentrath and the GRASS Development Team
+#
+#               This program is free software under the GNU General Public
+#               License (>=v2). Read the file COPYING that comes with GRASS
+#               for details.
+#
+#############################################################################
+
+"""Test of JSON export from pygrass.Module for usage in actinia"""
+
+from grass.pygrass.modules.interface import Module
+
+
+def test_rinfo_simple():
+    """Test if a Module can be exported to json dict"""
+
+    mod_json_dict = Module("r.info", map="elevation", run_=False).get_json_dict()
+
+    assert isinstance(mod_json_dict, dict)
+    assert set(mod_json_dict.keys()) == {"module", "id", "inputs"}
+    assert mod_json_dict["module"] == "r.info"
+    assert mod_json_dict["id"].startswith("r_info_")
+    assert "flags" not in mod_json_dict
+    assert set(mod_json_dict["inputs"][0].keys()) == {"param", "value"}
+
+
+def test_rinfo_ov():
+    """Test if a Module can be exported to json dict with quiet and overwrite flags"""
+
+    mod_json_dict = Module(
+        "r.info", quiet=True, map="elevation", run_=False
+    ).get_json_dict()
+
+    assert isinstance(mod_json_dict, dict)
+    assert set(mod_json_dict.keys()) == {"module", "id", "inputs", "quiet"}
+    assert mod_json_dict["module"] == "r.info"
+    assert mod_json_dict["id"].startswith("r_info_")
+    assert "flags" not in mod_json_dict
+    assert mod_json_dict["quiet"] is True
+    assert isinstance(mod_json_dict["inputs"], list)
+    assert set(mod_json_dict["inputs"][0].keys()) == {"param", "value"}
+
+
+def test_rinfo_ov_export():
+    """Test if a Module can be exported to json dict with overwrite
+    and verbose flags and results exported to CSV"""
+
+    mod_json_dict = Module(
+        "r.info",
+        verbose=True,
+        map="elevation",
+        flags="g",
+        run_=False,
+    ).get_json_dict(stdout_export="list")
+
+    assert isinstance(mod_json_dict, dict)
+    assert set(mod_json_dict.keys()) == {
+        "module",
+        "id",
+        "flags",
+        "inputs",
+        "verbose",
+        "stdout",
+    }
+    assert mod_json_dict["module"] == "r.info"
+    assert mod_json_dict["id"].startswith("r_info_")
+    assert mod_json_dict["flags"] == "g"
+    assert mod_json_dict["verbose"] is True
+    assert isinstance(mod_json_dict["inputs"], list)
+    assert set(mod_json_dict["inputs"][0].keys()) == {"param", "value"}
+    assert mod_json_dict["stdout"] == {
+        "id": "stdout",
+        "format": "list",
+        "delimiter": "|",
+    }
+
+
+def test_rslopeaspect_ov_export():
+    """Test if a Module can be exported to json dict with overwrite
+    and verbose flags and results exported to CSV"""
+
+    mod_json_dict = Module(
+        "r.slope.aspect",
+        elevation="elevation",
+        slope="slope",
+        aspect="aspect",
+        overwrite=True,
+        verbose=True,
+        run_=False,
+    ).get_json_dict(export="GTiff")
+
+    assert isinstance(mod_json_dict, dict)
+    assert set(mod_json_dict.keys()) == {
+        "module",
+        "id",
+        "inputs",
+        "outputs",
+        "overwrite",
+        "verbose",
+    }
+    assert mod_json_dict["module"] == "r.slope.aspect"
+    assert mod_json_dict["id"].startswith("r_slope_aspect_")
+    assert mod_json_dict["overwrite"] is True
+    assert mod_json_dict["verbose"] is True
+    assert isinstance(mod_json_dict["inputs"], list)
+    assert isinstance(mod_json_dict["outputs"], list)
+    assert set(mod_json_dict["outputs"][0].keys()) == {"param", "value", "export"}
+    assert mod_json_dict["outputs"][0]["export"] == {
+        "format": "GTiff",
+        "type": "raster",
+    }