grass.tools: Add API to access tools as functions

python/grass/app/cli.py

@@ -20,10 +20,30 @@
 import tempfile
 import os
 import sys
+import subprocess
 from pathlib import Path
 
+
 import grass.script as gs
 from grass.app.data import lock_mapset, unlock_mapset, MapsetLockingException
+from grass.experimental.tools import Tools
+
+
+def subcommand_run_tool(args, tool_args: list, help: bool):
+    command = [args.tool_name, *tool_args]
+    with tempfile.TemporaryDirectory() as tmp_dir_name:
+        project_name = "project"
+        project_path = Path(tmp_dir_name) / project_name
+        gs.create_project(project_path)
+        with gs.setup.init(project_path) as session:
+            if help:
+                result = subprocess.run(command, env=session.env)
+                return result.returncode
+            tools = Tools(capture_output=False)
+            try:
+                tools.run_from_list(command)
+            except subprocess.CalledProcessError as error:
+                return error.returncode
 
 
 def subcommand_lock_mapset(args):
@@ -77,6 +97,10 @@ def main(args=None, program=None):
 
     # Subcommand parsers
 
+    subparser = subparsers.add_parser("run", help="run a tool")
+    subparser.add_argument("tool_name", type=str)
+    subparser.set_defaults(func=subcommand_run_tool)
+
     subparser = subparsers.add_parser("lock", help="lock a mapset")
     subparser.add_argument("mapset_path", type=str)
     subparser.add_argument(
@@ -120,5 +144,27 @@ def main(args=None, program=None):
     subparser.add_argument("page", type=str)
     subparser.set_defaults(func=subcommand_show_man)
 
-    parsed_args = parser.parse_args(args)
+    # Parsing
+
+    if not args:
+        args = sys.argv[1:]
+    raw_args = args.copy()
+    add_back = None
+    if len(raw_args) > 2 and raw_args[0] == "run":
+        # Getting the --help of tools needs to work around the standard help mechanism
+        # of argparse.
+        # Maybe a better workaround is to use custom --help, action="help", print_help,
+        # and dedicated tool help function complimentary with g.manual subcommand
+        # interface.
+        if "--help" in raw_args[2:]:
+            raw_args.remove("--help")
+            add_back = "--help"
+        elif "--h" in raw_args[2:]:
+            raw_args.remove("--h")
+            add_back = "--h"
+    parsed_args, other_args = parser.parse_known_args(raw_args)
+    if parsed_args.subcommand == "run":
+        if add_back:
+            other_args.append(add_back)
+        return parsed_args.func(parsed_args, other_args, help=bool(add_back))
     return parsed_args.func(parsed_args)

python/grass/experimental/Makefile

@@ -7,7 +7,8 @@ DSTDIR = $(ETC)/python/grass/experimental
 
 MODULES = \
 	create \
-	mapset
+	mapset \
+	tools
 
 PYFILES := $(patsubst %,$(DSTDIR)/%.py,$(MODULES) __init__)
 PYCFILES := $(patsubst %,$(DSTDIR)/%.pyc,$(MODULES) __init__)

python/grass/experimental/tests/conftest.py

@@ -33,6 +33,14 @@ def xy_session_for_module(tmp_path_factory):
         yield session
 
 
+@pytest.fixture
+def xy_dataset_session(tmp_path):
+    """Creates a session with a mapset which has vector with a float column"""
+    gs.core._create_location_xy(tmp_path, "test")  # pylint: disable=protected-access
+    with gs.setup.init(tmp_path / "test") as session:
+        yield session
+
+
 @pytest.fixture
 def unique_id():
     """A unique alphanumeric identifier"""

python/grass/experimental/tests/grass_tools_test.py

@@ -0,0 +1,266 @@
+"""Test grass.experimental.Tools class"""
+
+import os
+import json
+
+import pytest
+
+
+import grass.script as gs
+from grass.experimental.mapset import TemporaryMapsetSession
+from grass.experimental.tools import Tools
+from grass.exceptions import CalledModuleError
+
+
+def test_key_value_parser_number(xy_dataset_session):
+    """Check that numbers are parsed as numbers"""
+    tools = Tools(session=xy_dataset_session)
+    assert tools.g_region(flags="g").keyval["nsres"] == 1
+
+
+@pytest.mark.xfail
+def test_key_value_parser_multiple_values(xy_dataset_session):
+    """Check that strings and floats are parsed"""
+    tools = Tools(session=xy_dataset_session)
+    name = "surface"
+    tools.r_surf_gauss(output=name)  # needs seed
+    result = tools.r_info(map=name, flags="g").keyval
+    assert result["datatype"] == "DCELL"
+    assert result["nsres"] == 1
+    result = tools.r_univar(map=name, flags="g").keyval
+    assert result["mean"] == pytest.approx(-0.756762744552762)
+
+
+def test_json_parser(xy_dataset_session):
+    """Check that JSON is parsed"""
+    tools = Tools(session=xy_dataset_session)
+    assert (
+        tools.g_search_modules(keyword="random", flags="j").json[0]["name"]
+        == "r.random"
+    )
+
+
+def test_json_direct_access(xy_dataset_session):
+    """Check that JSON is parsed"""
+    tools = Tools(session=xy_dataset_session)
+    assert tools.g_search_modules(keyword="random", flags="j")[0]["name"] == "r.random"
+
+
+def test_json_direct_access_bad_key_type(xy_dataset_session):
+    """Check that JSON is parsed"""
+    tools = Tools(session=xy_dataset_session)
+    with pytest.raises(TypeError):
+        tools.g_search_modules(keyword="random", flags="j")["name"]
+
+
+def test_json_direct_access_bad_key_value(xy_dataset_session):
+    """Check that JSON is parsed"""
+    tools = Tools(session=xy_dataset_session)
+    high_number = 100_000_000
+    with pytest.raises(IndexError):
+        tools.g_search_modules(keyword="random", flags="j")[high_number]
+
+
+def test_json_direct_access_not_json(xy_dataset_session):
+    """Check that JSON is parsed"""
+    tools = Tools(session=xy_dataset_session)
+    with pytest.raises(json.JSONDecodeError):
+        tools.g_search_modules(keyword="random")[0]["name"]
+
+
+def test_stdout_as_text(xy_dataset_session):
+    """Check that simple text is parsed and has no whitespace"""
+    tools = Tools(session=xy_dataset_session)
+    assert tools.g_mapset(flags="p").text == "PERMANENT"
+
+
+def test_stdout_as_space_items(xy_dataset_session):
+    """Check that whitespace-separated items are parsed"""
+    tools = Tools(session=xy_dataset_session)
+    assert tools.g_mapset(flags="l").space_items == ["PERMANENT"]
+
+
+def test_stdout_split_whitespace(xy_dataset_session):
+    """Check that whitespace-based split function works"""
+    tools = Tools(session=xy_dataset_session)
+    assert tools.g_mapset(flags="l").text_split() == ["PERMANENT"]
+
+
+def test_stdout_split_space(xy_dataset_session):
+    """Check that the split function works with space"""
+    tools = Tools(session=xy_dataset_session)
+    # Not a good example usage, but it tests the functionality.
+    assert tools.g_mapset(flags="l").text_split(" ") == ["PERMANENT", ""]
+
+
+def test_stdout_without_capturing(xy_dataset_session):
+    """Check that text is not present when not capturing it"""
+    tools = Tools(session=xy_dataset_session, capture_output=False)
+    assert not tools.g_mapset(flags="p").text
+    assert tools.g_mapset(flags="p").text is None
+
+
+def test_direct_overwrite(xy_dataset_session):
+    """Check overwrite as a parameter"""
+    tools = Tools(session=xy_dataset_session)
+    tools.r_random_surface(output="surface", seed=42)
+    tools.r_random_surface(output="surface", seed=42, overwrite=True)
+
+
+def test_object_overwrite(xy_dataset_session):
+    """Check overwrite as parameter of the tools object"""
+    tools = Tools(session=xy_dataset_session, overwrite=True)
+    tools.r_random_surface(output="surface", seed=42)
+    tools.r_random_surface(output="surface", seed=42)
+
+
+def test_no_overwrite(xy_dataset_session):
+    """Check that it fails without overwrite"""
+    tools = Tools(session=xy_dataset_session)
+    tools.r_random_surface(output="surface", seed=42)
+    with pytest.raises(CalledModuleError, match="overwrite"):
+        tools.r_random_surface(output="surface", seed=42)
+
+
+def test_env_overwrite(xy_dataset_session):
+    """Check that overwrite from env parameter is used"""
+    # env = xy_dataset_session.env.copy()  # ideally
+    env = os.environ.copy()  # for now
+    env["GRASS_OVERWRITE"] = "1"
+    tools = Tools(session=xy_dataset_session, env=env)
+    tools.r_random_surface(output="surface", seed=42)
+    tools.r_random_surface(output="surface", seed=42)
+
+
+def test_global_overwrite_vs_env(xy_dataset_session):
+    """Check that global overwrite is not used when separate env is used"""
+    # env = xy_dataset_session.env.copy()  # ideally
+    env = os.environ.copy()  # for now
+    os.environ["GRASS_OVERWRITE"] = "1"  # change to xy_dataset_session.env
+    tools = Tools(session=xy_dataset_session, env=env)
+    tools.r_random_surface(output="surface", seed=42)
+    with pytest.raises(CalledModuleError, match="overwrite"):
+        tools.r_random_surface(output="surface", seed=42)
+    del os.environ["GRASS_OVERWRITE"]  # check or ideally remove this
+
+
+def test_global_overwrite_vs_init(xy_dataset_session):
+    """Check that global overwrite is not used when separate env is used"""
+    tools = Tools(session=xy_dataset_session)
+    os.environ["GRASS_OVERWRITE"] = "1"  # change to xy_dataset_session.env
+    tools.r_random_surface(output="surface", seed=42)
+    with pytest.raises(CalledModuleError, match="overwrite"):
+        tools.r_random_surface(output="surface", seed=42)
+    del os.environ["GRASS_OVERWRITE"]  # check or ideally remove this
+
+
+def test_stdin(xy_dataset_session):
+    """Test that stdin is accepted"""
+    tools = Tools(session=xy_dataset_session)
+    tools.feed_input_to("13.45,29.96,200").v_in_ascii(
+        input="-", output="point", separator=","
+    )
+
+
+def test_raises(xy_dataset_session):
+    """Test that exception is raised for wrong parameter value"""
+    tools = Tools(session=xy_dataset_session)
+    wrong_name = "wrong_standard"
+    with pytest.raises(CalledModuleError, match=wrong_name):
+        tools.feed_input_to("13.45,29.96,200").v_in_ascii(
+            input="-",
+            output="point",
+            format=wrong_name,
+        )
+
+
+def test_run_command(xy_dataset_session):
+    """Check run_command and its overwrite parameter"""
+    tools = Tools(session=xy_dataset_session)
+    tools.run_command("r.random.surface", output="surface", seed=42)
+    tools.run_command("r.random.surface", output="surface", seed=42, overwrite=True)
+
+
+def test_parse_command_key_value(xy_dataset_session):
+    tools = Tools(session=xy_dataset_session)
+    assert tools.parse_command("g.region", flags="g")["nsres"] == "1"
+
+
+def test_parse_command_json(xy_dataset_session):
+    tools = Tools(session=xy_dataset_session)
+    assert (
+        tools.parse_command("g.region", flags="g", format="json")["region"]["ns-res"]
+        == 1
+    )
+
+
+def test_with_context_managers(tmpdir):
+    project = tmpdir / "project"
+    gs.create_project(project)
+    with gs.setup.init(project) as session:
+        tools = Tools(session=session)
+        tools.r_random_surface(output="surface", seed=42)
+        with TemporaryMapsetSession(env=tools.env) as mapset:
+            tools.r_random_surface(output="surface", seed=42, env=mapset.env)
+            with gs.MaskManager(env=mapset.env) as mask:
+                # TODO: Do actual test
+                tools.r_univar(map="surface", env=mask.env, format="json")[0]["mean"]
+
+
+def test_misspelling(xy_dataset_session):
+    tools = Tools(session=xy_dataset_session)
+    with pytest.raises(AttributeError, match=r"r\.slope\.aspect"):
+        tools.r_sloppy_respect()
+
+
+def test_multiple_suggestions(xy_dataset_session):
+    tools = Tools(session=xy_dataset_session)
+    with pytest.raises(AttributeError, match=r"v\.db\.univar|db\.univar"):
+        tools.db_v_uni_var()
+
+
+def test_tool_group_vs_model_name(xy_dataset_session):
+    tools = Tools(session=xy_dataset_session)
+    with pytest.raises(AttributeError, match=r"r\.sim\.water"):
+        tools.rSIMWEwater()
+
+
+def test_wrong_attribute(xy_dataset_session):
+    tools = Tools(session=xy_dataset_session)
+    with pytest.raises(AttributeError, match="execute_big_command"):
+        tools.execute_big_command()
+
+import numpy as np
+
+def test_numpy_one_input(xy_dataset_session):
+    """Check that global overwrite is not used when separate env is used"""
+    tools = Tools(session=xy_dataset_session)
+    tools.r_slope_aspect(elevation=np.ones((1, 1)), slope="slope")
+    assert tools.r_info(map="slope", format="json")["datatype"] == "FCELL"
+
+# Other possible ways how to handle the syntax:
+
+# class ToNumpy:
+#     pass
+
+# class AsInput:
+#     pass
+
+# def test_numpy_one_input(xy_dataset_session):
+#     """Check that global overwrite is not used when separate env is used"""
+#     tools = Tools(session=xy_dataset_session)
+#     tools.r_slope_aspect(elevation=np.ones((1, 1)), slope="slope", aspect="aspect", force_numpy_for_output=True)
+#     tools.r_slope_aspect(elevation=np.ones((1, 1)), slope=np.nulls(0,0), aspect="aspect")
+#     tools.r_slope_aspect(elevation=np.ones((1, 1)), slope=ToNumpy(), aspect="aspect")
+#     tools.r_slope_aspect(elevation=np.ones((1, 1)), slope=np.ndarray, aspect="aspect")
+#     tools.r_slope_aspect.ufunc(np.ones((1, 1)), slope=True, aspect=True, overwrite=True)  # (np.array, np.array)
+#     tools.r_slope_aspect(elevation=np.ones((1, 1)), slope=AsInput, aspect=AsInput)  # {"slope": np.array(...), "aspect": np.array(...) }
+#     assert tools.r_info(map="slope", format="json")["datatype"] == "FCELL"
+
+def test_numpy_one_input_one_output(xy_dataset_session):
+    """Check that global overwrite is not used when separate env is used"""
+    tools = Tools(session=xy_dataset_session)
+    tools.g_region(rows=2, cols=3)
+    slope = tools.r_slope_aspect(elevation=np.ones((2, 3)), slope=np.ndarray)
+    assert slope.shape == (2, 3)
+    assert slope[0] == 0