grass.script: Automatically parse JSON and CSV in parse_command

python/grass/script/core.py

@@ -27,6 +27,9 @@
 import string
 import random
 import shlex
+import json
+import csv
+import io
 from tempfile import NamedTemporaryFile
 from pathlib import Path
 
@@ -543,40 +546,52 @@ def read_command(*args, **kwargs):
 
 def parse_command(*args, **kwargs):
     """Passes all arguments to read_command, then parses the output
-    by parse_key_val().
+    by default with parse_key_val().
 
-    Parsing function can be optionally given by <em>parse</em> parameter
-    including its arguments, e.g.
+    If the command has parameter <em>format</em> and is called with
+    <em>format=json</em>, the output will be parsed into a dictionary.
+    Similarly, with <em>format=csv</em> the output will be parsed into
+    a list of lists (CSV rows).
 
     ::
 
-        parse_command(..., parse = (grass.parse_key_val, { 'sep' : ':' }))
+        parse_command("v.db.select", ..., format="json")
 
-    or you can simply define <em>delimiter</em>
+    Custom parsing function can be optionally given by <em>parse</em> parameter
+    including its arguments, e.g.
 
     ::
 
-        parse_command(..., delimiter = ':')
+        parse_command(..., parse=(gs.parse_key_val, {'sep': ':'}))
+
+    Parameter <em>delimiter</em> is deprecated.
 
     :param args: list of unnamed arguments (see start_command() for details)
     :param kwargs: list of named arguments (see start_command() for details)
 
     :return: parsed module output
     """
+
+    def parse_csv(result):
+        return list(csv.DictReader(io.StringIO(result)))
+
     parse = None
     parse_args = {}
     if "parse" in kwargs:
         if isinstance(kwargs["parse"], tuple):
             parse = kwargs["parse"][0]
             parse_args = kwargs["parse"][1]
         del kwargs["parse"]
-
-    if "delimiter" in kwargs:
-        parse_args = {"sep": kwargs["delimiter"]}
-        del kwargs["delimiter"]
+    elif kwargs.get("format") == "json":
+        parse = json.loads
+    elif kwargs.get("format") == "csv":
+        parse = parse_csv
 
     if not parse:
         parse = parse_key_val  # use default fn
+        if "delimiter" in kwargs:
+            parse_args = {"sep": kwargs["delimiter"]}
+            del kwargs["delimiter"]
 
     res = read_command(*args, **kwargs)
 

python/grass/script/testsuite/test_start_command_functions_nc.py

@@ -3,7 +3,8 @@
 from grass.gunittest.case import TestCase
 from grass.gunittest.main import test
 
-from grass.script.core import start_command, PIPE
+from grass.script.core import parse_command, start_command, PIPE
+from grass.script.utils import parse_key_val
 
 LOCATION = "nc"
 
@@ -49,5 +50,37 @@ def test_multiple_underscores(self):
         self.assertIn(b"raster", stderr)
 
 
+class TestParseCommand(TestCase):
+    """Tests parse_command"""
+
+    def test_parse_default(self):
+        result = parse_command("r.info", map="elevation", flags="g")
+        self.assertTrue(
+            isinstance(result, dict) and isinstance(result.get("north"), str)
+        )
+        result_2 = parse_command("r.info", map="elevation", flags="g", delimiter="=")
+        self.assertDictEqual(result, result_2)
+        result_3 = parse_command(
+            "r.info", map="elevation", flags="g", parse=(parse_key_val, {"sep": "="})
+        )
+        self.assertDictEqual(result, result_3)
+
+    def test_parse_format_json(self):
+        result = parse_command(
+            "r.what", map="elevation", coordinates=(640000, 220000), format="json"
+        )
+        self.assertTrue(
+            isinstance(result, list)
+            and isinstance(result[0].get("easting"), (int, float))
+        )
+
+    def test_parse_format_csv(self):
+        reference = parse_command("v.db.select", map="zipcodes", format="json")[
+            "records"
+        ]
+        result = parse_command("v.db.select", map="zipcodes", format="csv")
+        self.assertListEqual(list(reference[0].keys()), list(result[0].keys()))
+
+
 if __name__ == "__main__":
     test()