Merge remote-tracking branch 'origin/main' into flake8_update

.flake8

@@ -18,6 +18,10 @@ exclude = .git
 max-line-length = 80
 
 per-file-ignores =
+    ./src/grass_gis_helpers/cleanup.py: F821
+    ./src/grass_gis_helpers/general.py: F821
+    ./src/grass_gis_helpers/location.py: F821
     ./src/grass_gis_helpers/mapset.py: F821
     ./src/grass_gis_helpers/parallel.py: F821
     ./src/grass_gis_helpers/tiling.py: F821
+    ./src/grass_gis_helpers/validation.py: F821

.github/workflows/linting.yml

@@ -1,6 +1,6 @@
 name: Python Flake8, black and pylint code quality check
 
-on: [push]
+on: [push,pull_request]
 
 jobs:
   lint:

README.md

@@ -11,14 +11,25 @@ pip install grass-gis-helpers
 ```
 
 ## Small example
+
 Small example how the library can be used inside a GRASS GIS session:
+
 ```python3
 from grass_gis_helpers import general
 general.set_nprocs(2)
 ```
 
 ## DEV setup
 
+pip-tools is required for DEV setup:
+
+```bash
+# only once
+pip3 install pip-tools
+```
+
+then install grass-gis-helpers from the local repository:
+
 ```bash
 pip3 install -e .
 

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "grass_gis_helpers"
-version = "0.0.1"
+version = "0.1.0"
 authors = [
   { name="Anika Weinmann", email="weinmann@mundialis.de" },
 ]

src/grass_gis_helpers/cleanup.py

@@ -16,16 +16,90 @@
 #
 #############################################################################
 
-from os import devnull
+import os
+import shutil
+import gc
 
 import grass.script as grass
+from .location import get_location_size, switch_back_original_location
+
+
+def general_cleanup(
+    rm_rasters=[],
+    rm_vectors=[],
+    rm_files=[],
+    rm_dirs=[],
+    rm_groups=[],
+    rm_groups_wo_rasters=[],
+    rm_regions=[],
+    rm_strds=[],
+    orig_region=None,
+    rm_mask=False,
+):
+    """General cleanup function"""
+
+    grass.message(_("Cleaning up..."))
+    nulldev = open(os.devnull, "w")
+    kwargs = {"flags": "f", "quiet": True, "stderr": nulldev}
+    for rmg in rm_groups:
+        if grass.find_file(name=rmg, element="group")["file"]:
+            group_rasters = grass.parse_command(
+                "i.group", flags="lg", group=rmg
+            )
+            rm_rasters.extend(group_rasters)
+            grass.run_command("g.remove", type="group", name=rmg, **kwargs)
+    for rmg_wor in rm_groups_wo_rasters:
+        if grass.find_file(name=rmg_wor, element="group")["file"]:
+            grass.run_command("g.remove", type="group", name=rmg_wor, **kwargs)
+    for rmrast in rm_rasters:
+        if grass.find_file(name=rmrast, element="raster")["file"]:
+            grass.run_command("g.remove", type="raster", name=rmrast, **kwargs)
+    for rmvect in rm_vectors:
+        if grass.find_file(name=rmvect, element="vector")["file"]:
+            grass.run_command("g.remove", type="vector", name=rmvect, **kwargs)
+    for rmfile in rm_files:
+        if os.path.isfile(rmfile):
+            os.remove(rmfile)
+    for rmdir in rm_dirs:
+        if os.path.isdir(rmdir):
+            shutil.rmtree(rmdir)
+    if orig_region is not None:
+        if grass.find_file(name=orig_region, element="windows")["file"]:
+            grass.run_command("g.region", region=orig_region)
+            grass.run_command(
+                "g.remove", type="region", name=orig_region, **kwargs
+            )
+    for rmreg in rm_regions:
+        if grass.find_file(name=rmreg, element="windows")["file"]:
+            grass.run_command("g.remove", type="region", name=rmreg, **kwargs)
+    strds = grass.parse_command("t.list", type="strds")
+    mapset = grass.gisenv()["MAPSET"]
+    for rm_s in rm_strds:
+        if f"{rm_s}@{mapset}" in strds:
+            grass.run_command(
+                "t.remove",
+                flags="rf",
+                type="strds",
+                input=rm_s,
+                quiet=True,
+                stderr=nulldev,
+            )
+    if rm_mask:
+        if grass.find_file(name="MASK", element="raster")["file"]:
+            grass.run_command("r.mask", flags="r")
+
+    # get location size
+    get_location_size()
+
+    # Garbage Collector: release unreferenced memory
+    gc.collect()
 
 
 def rm_vects(vects):
     """Function to remove clean vector maps
     Args:
         vects (list): list of vector maps which should be removed"""
-    nuldev = open(devnull, "w")
+    nuldev = open(os.devnull, "w")
     kwargs = {"flags": "f", "quiet": True, "stderr": nuldev}
     for rmv in vects:
         if grass.find_file(name=rmv, element="vector")["file"]:
@@ -38,9 +112,28 @@ def reset_region(region):
         region (str): the name of the saved region which should be set and
                       deleted
     """
-    nulldev = open(devnull, "w")
+    nulldev = open(os.devnull, "w")
     kwargs = {"flags": "f", "quiet": True, "stderr": nulldev}
     if region:
         if grass.find_file(name=region, element="windows")["file"]:
             grass.run_command("g.region", region=region)
             grass.run_command("g.remove", type="region", name=region, **kwargs)
+
+
+def cleaning_tmp_location(original_gisrc, tmp_loc, gisdbase, tmp_gisrc):
+    """Cleaning up things from temporary location
+
+    Args:
+        original_gisrc (str): The path to the original GISRC file
+        tmp_loc (str): The name of the temporary location
+        gisdbase (str): The GISDBASE info
+        tmp_gisrc (str): The path to the temporary GISRC file
+    """
+    # switch back to original gisrc
+    if original_gisrc:
+        switch_back_original_location(original_gisrc)
+    # remove temporary location
+    if tmp_loc:
+        grass.try_rmdir(os.path.join(gisdbase, tmp_loc))
+    if tmp_gisrc:
+        grass.try_remove(tmp_gisrc)

src/grass_gis_helpers/general.py

@@ -17,6 +17,8 @@
 #############################################################################
 
 import multiprocessing as mp
+import subprocess
+import psutil
 
 import grass.script as grass
 
@@ -35,3 +37,77 @@ def set_nprocs(nprocs):
                 f"{nprocs_real} CPUs available."
             )
         return nprocs
+
+
+def communicate_grass_command(*args, **kwargs):
+    """Return stdout and stderr from executed GRASS command"""
+    kwargs["stdout"] = grass.PIPE
+    kwargs["stderr"] = grass.PIPE
+    grass_ps = grass.start_command(*args, **kwargs)
+    return grass_ps.communicate()
+
+
+def check_grass_version(comp_version=(8, 0, 0)):
+    """Returns boolean, if current GRASS version is >= some compare version"""
+    cur_version = tuple(
+        [
+            int(x.replace("dev", "")) if x != "dev" else 0
+            for x in grass.version()["version"].split(".")
+        ]
+    )
+    return cur_version >= comp_version
+
+
+def log_memory(grassenv=None):
+    """Log memory usage"""
+    if not grassenv:
+        grassenv = grass.gisenv()
+    cmd = grass.Popen(
+        f"df -h {grassenv['GISDBASE']}", shell=True, stdout=subprocess.PIPE
+    )
+    grass.message(
+        _(
+            (
+                "\nDisk usage of GRASS GIS database:\n",
+                f"{cmd.communicate()[0].decode('utf-8').rstrip()}\n",
+            ),
+        )
+    )
+
+    grass.message(_(f"\nmemory: \n{str(psutil.virtual_memory())}"))
+    grass.message(_(f"\nswap memory: \n{str(psutil.swap_memory())}"))
+    # ulimit -a
+    cmd = grass.Popen("ulimit -a", shell=True, stdout=subprocess.PIPE)
+    grass.message(
+        _(f"\nulimit -a: \n{cmd.communicate()[0].decode('utf-8').rstrip()}")
+    )
+
+
+def free_ram(unit, percent=100):
+    """The function gives the amount of the percentages of the available
+    RAM memory and free swap space.
+    Args:
+        unit(string): 'GB' or 'MB'
+        percent(int): number of percent which shoud be used of the available
+                      RAM memory and free swap space
+                      default 100%
+    Returns:
+        memory_MB_percent/memory_GB_percent(int): percent of the the available
+                                                  memory and free swap in MB or
+                                                  GB
+
+    """
+    # use psutil cause of alpine busybox free version for RAM/SWAP usage
+    mem_available = psutil.virtual_memory().available
+    swap_free = psutil.swap_memory().free
+    memory_gb = (mem_available + swap_free) / 1024.0**3
+    memory_mb = (mem_available + swap_free) / 1024.0**2
+
+    if unit == "MB":
+        memory_mb_percent = memory_mb * percent / 100.0
+        return int(round(memory_mb_percent))
+    elif unit == "GB":
+        memory_gb_percent = memory_gb * percent / 100.0
+        return int(round(memory_gb_percent))
+    else:
+        grass.fatal(f"Memory unit {unit} not supported")

src/grass_gis_helpers/location.py

@@ -0,0 +1,111 @@
+#!/usr/bin/env python3
+
+############################################################################
+#
+# MODULE:       lib with location related helper functions for GRASS GIS
+#
+# AUTHOR(S):    Lina Krisztian, Anika Weinmann
+#
+# PURPOSE:      lib with location related helper functions for GRASS GIS
+#
+# COPYRIGHT:	(C) 2023 by mundialis 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.
+#
+#############################################################################
+
+import os
+import subprocess
+
+import grass.script as grass
+
+
+def get_location_size():
+    """Log size of current location"""
+    current_gisdbase = grass.gisenv()["GISDBASE"]
+    cmd = grass.Popen(
+        f"df -h {current_gisdbase}", shell=True, stdout=subprocess.PIPE
+    )
+    grass.message(
+        _(
+            (
+                "\nDisk usage of GRASS GIS database:\n",
+                f"{cmd.communicate()[0].decode('utf-8').rstrip()}\n",
+            ),
+        )
+    )
+
+
+def create_tmp_location(epsg=4326):
+    """Creation of a new temporary location
+
+    Args:
+        epsg (int): The number of the EPSG code
+
+    Returns:
+        tmp_loc (str): The name of the temporary location
+        tmp_gisrc (str): The path to the original GISRC file
+    """
+    current_gisdbase = grass.gisenv()["GISDBASE"]
+    srcgisrc = grass.tempfile()
+    tmp_loc = f"temp_epsg{epsg}_location_{os.getpid()}"
+    gisrc_file = open(srcgisrc, "w")
+    gisrc_file.write("MAPSET: PERMANENT\n")
+    gisrc_file.write(f"GISDBASE: {current_gisdbase}\n")
+    gisrc_file.write(f"LOCATION_NAME: {tmp_loc}\n")
+    gisrc_file.write("GUI: text\n")
+    gisrc_file.close()
+
+    proj_test = grass.parse_command("g.proj", flags="g")
+    if "epsg" in proj_test:
+        epsg_arg = {"epsg": epsg}
+    else:
+        epsg_arg = {"srid": f"EPSG:{epsg}"}
+    # create temp location from input without import
+    grass.verbose(_(f"Creating temporary location with EPSG:{epsg}..."))
+    grass.run_command(
+        "g.proj", flags="c", location=tmp_loc, quiet=True, **epsg_arg
+    )
+
+    # switch to temp location
+    os.environ["GISRC"] = str(srcgisrc)
+    proj = grass.parse_command("g.proj", flags="g")
+    if "epsg" in proj:
+        new_epsg = proj["epsg"]
+    else:
+        new_epsg = proj["srid"].split("EPSG:")[1]
+    if new_epsg != str(epsg):
+        grass.fatal(_("Creation of temporary location failed!"))
+
+    return tmp_loc, srcgisrc
+
+
+def get_current_location():
+    """Get infos to current location
+
+    Returns:
+        loc (str): The name of the current location
+        mapset (str): The name of the current mapset
+        gisdbase (str): The current GISDBASE info
+        gisrc (str): The path to the current GISRC file
+    """
+    # get current location, mapset, ...
+    grassenv = grass.gisenv()
+    loc = grassenv["LOCATION_NAME"]
+    mapset = grassenv["MAPSET"]
+    gisdbase = grassenv["GISDBASE"]
+    gisrc = os.environ["GISRC"]
+    return loc, mapset, gisdbase, gisrc
+
+
+def switch_back_original_location(original_gisrc):
+    """Switching back to original location after the computation in tmp
+    location
+
+    Args:
+        original_gisrc (str): The path to the original GISRC file
+    """
+    # switch to target location
+    os.environ["GISRC"] = str(original_gisrc)

src/grass_gis_helpers/parallel.py

@@ -70,7 +70,8 @@ def run_module_parallel(
         msg = proc.outputs["stderr"].value.strip()
         grass.message(_(f"\nLog of {proc.get_bash()}:"))
         for msg_part in msg.split("\n"):
-            grass.message(_(msg_part))
+            if len(msg_part) > 0:
+                grass.message(msg_part)
 
     # verify that switching the mapset worked
     location_path = verify_mapsets(start_cur_mapset)