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
@@ -553,9 +555,8 @@ 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)
+            print(f"OSError error({e.errno}): {e.strerror}")
+            raise GrassError(f"Error running: `{self.name} --interface-description`.")
         # get the xml of the module
         self.xml = get_cmd_xml.communicate()[0]
         # transform and parse the xml into an Element class:
@@ -690,7 +691,7 @@ 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)
+                raise ParameterError("{} is not a valid parameter.".format(key))
 
     def get_bash(self):
         """Return a BASH representation of the Module."""
@@ -770,6 +771,82 @@ def check(self):
                     msg = "Required parameter <%s> not set."
                     raise ParameterError(msg % k)
 
+    def get_json_dict(self, export=None):
+        """Return a dictionary that includes the name, all valid
+        inputs, outputs and flags as well as export settings for
+        usage with actinia
+        param export: string with export format, e.g. GTiff
+        """
+        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",
+        }
+        special_flags = ["overwrite", "verbose", "quiet"]
+        skip = ["stdin", "stdout", "stderr"]
+
+        # Check export formats
+        if export and export not in export_dict:
+            raise GrassError(f"Invalid export format <{export}>.")
+
+        # 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(val.value)
+                    if type(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(val.value)
+                    if type(val.value) == list
+                    else str(val.value),
+                }
+                if export:
+                    param["export"] = {
+                        "format": export,
+                        "type": export_dict[export],
+                    }
+                outputs.append(param)
+        json_dict["outputs"] = outputs
+
+        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/test_modules.py

@@ -53,6 +53,44 @@ def test_rsun(self):
         out.close()
 
 
+class TestModulesJsonDictExport(TestCase):
+    def test_rinfo_simple(self):
+        """Test if a Module can be exported to json dict"""
+
+        Module("r.info", map="elevation", run_=False).get_json_dict()
+
+    def test_rinfo_ov(self):
+        """Test if a Module can be exported to json dict with verbose flag"""
+
+        Module("r.info", verbose=True, map="elevation", run_=False).get_json_dict()
+
+    def test_rinfo_ov_export(self):
+        """Test if a Module can be exported to json dict with overwrite
+        and verbose flags and results exported to CSV"""
+
+        Module(
+            "r.info",
+            verbose=True,
+            map="elevation",
+            flags="g",
+            run_=False,
+        ).get_json_dict(export="CSV")
+
+    def test_rslopeaspect_ov_export(self):
+        """Test if a Module can be exported to json dict with overwrite
+        and verbose flags and results exported to CSV"""
+
+        Module(
+            "r.slope.aspect",
+            elevation="elevation",
+            slope="slope",
+            aspect="aspect",
+            overwrite=True,
+            verbose=True,
+            run_=False,
+        ).get_json_dict(export="GTiff")
+
+
 class TestModulesCheck(TestCase):
     def test_flags_with_suppress_required(self):
         """Test if flags with suppress required are handle correctly"""