pygrass: add get_json_dict function

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

@@ -1,15 +1,7 @@
-from __future__ import (
-    nested_scopes,
-    generators,
-    division,
-    absolute_import,
-    with_statement,
-    print_function,
-    unicode_literals,
-)
-import sys
-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
@@ -22,13 +14,6 @@
 from .read import GETFROMTAG, DOC
 from .env import G_debug
 
-if sys.version_info[0] == 2:
-    from itertools import izip_longest as zip_longest
-else:
-    from itertools import zip_longest
-
-    unicode = str
-
 
 def _get_bash(self, *args, **kargs):
     return self.get_bash()
@@ -528,9 +513,7 @@ def f(*args, **kargs):
     """
 
     def __init__(self, cmd, *args, **kargs):
-        if isinstance(cmd, unicode):
-            self.name = str(cmd)
-        elif isinstance(cmd, str):
+        if isinstance(cmd, str):
             self.name = cmd
         else:
             raise GrassError("Problem initializing the module {s}".format(s=cmd))
@@ -539,8 +522,9 @@ def __init__(self, cmd, *args, **kargs):
             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)
+            raise GrassError(
+                "Error running: `{} --interface-description`.".format(self.name)
+            )
         # get the xml of the module
         self.xml = get_cmd_xml.communicate()[0]
         # transform and parse the xml into an Element class:
@@ -675,7 +659,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."""
@@ -755,6 +739,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("Invalid Export format.")
+
+        # 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

@@ -5,19 +5,15 @@
 """
 import sys
 from fnmatch import fnmatch
+from io import BytesIO as StringIO
+
 from grass.gunittest.case import TestCase
 from grass.gunittest.main import test
 
 from grass.script.core import get_commands
 from grass.exceptions import ParameterError
 from grass.pygrass.modules.interface import Module
 
-PY2 = sys.version_info[0] == 2
-if PY2:
-    from StringIO import StringIO
-else:
-    from io import BytesIO as StringIO
-
 
 SKIP = [
     "g.parser",
@@ -71,6 +67,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"""