python: Use context manager for subprocesses and opening files (OSGeo#4908)

* python: Use context manager for subprocesses and opening files

* style: Fix new write-whole-file (FURB103) errors

* style: Fix new read-whole-file (FURB101) errors

* checks: Remove fixed Ruff SIM115 exclusions

* grass.imaging.images2avi: Remove shell=True from subprocess.Popen

Author: echoix

pyproject.toml

@@ -324,7 +324,6 @@ ignore = [
 "python/grass/gunittest/multireport.py" = ["PYI024"]
 "python/grass/gunittest/testsu*/d*/s*/s*/subsub*/t*/test_segfaut.py" = ["B018"]
 "python/grass/gunittest/testsuite/test_assertions_rast3d.py" = ["FLY002"]
-"python/grass/imaging/images2*.py" = ["SIM115"]
 "python/grass/imaging/images2ims.py" = ["PTH208"]
 "python/grass/jupyter/testsuite/interactivemap_test.py" = ["PGH004"]
 "python/grass/jupyter/testsuite/map_test.py" = ["PGH004"]
@@ -343,9 +342,7 @@ ignore = [
 "python/grass/pygrass/vector/testsuite/test_table.py" = ["PLW0108"]
 "python/grass/script/array.py" = ["A005"]
 "python/grass/script/core.py" = ["PTH208"]
-"python/grass/script/db.py" = ["SIM115"]
-"python/grass/script/raster.py" = ["SIM115"]
-"python/grass/script/utils.py" = ["FURB189", "SIM115"]
+"python/grass/script/utils.py" = ["FURB189"]
 "python/grass/temporal/aggregation.py" = ["SIM115"]
 "python/grass/temporal/register.py" = ["SIM115"]
 "python/grass/temporal/stds_export.py" = ["SIM115"]

python/grass/gunittest/multirunner.py

@@ -111,16 +111,16 @@ def main():
     # we assume that the start script is available and in the PATH
     # the shell=True is here because of MS Windows? (code taken from wiki)
     startcmd = grass_executable + " --config path"
-    p = subprocess.Popen(
+    with subprocess.Popen(
         startcmd, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE
-    )
-    out, err = p.communicate()
-    if p.returncode != 0:
-        print(
-            "ERROR: Cannot find GRASS GIS start script (%s):\n%s" % (startcmd, err),
-            file=sys.stderr,
-        )
-        return 1
+    ) as p:
+        out, err = p.communicate()
+        if p.returncode != 0:
+            print(
+                "ERROR: Cannot find GRASS GIS start script (%s):\n%s" % (startcmd, err),
+                file=sys.stderr,
+            )
+            return 1
     gisbase = decode(out.strip())
 
     # set GISBASE environment variable
@@ -151,7 +151,7 @@ def main():
         # including also type to make it unique and preserve it for sure
         report = "report_for_" + location + "_" + location_type
         absreport = os.path.abspath(report)
-        p = subprocess.Popen(
+        with subprocess.Popen(
             [
                 sys.executable,
                 "-tt",
@@ -167,23 +167,24 @@ def main():
                 absreport,
             ],
             cwd=grasssrc,
-        )
-        returncode = p.wait()
-        reports.append(report)
+        ) as p2:
+            returncode = p2.wait()
+            reports.append(report)
 
     if main_report:
         # TODO: solve the path to source code (work now only for grass source code)
         arguments = [
             sys.executable,
-            grasssrc + "/python/grass/gunittest/" + "multireport.py",
-            "--timestapms",
+            grasssrc + "/python/grass/gunittest/multireport.py",
+            "--timestamps",
+            *reports,
         ]
-        arguments.extend(reports)
-        p = subprocess.Popen(arguments)
-        returncode = p.wait()
-        if returncode != 0:
-            print("ERROR: Creation of main report failed.", file=sys.stderr)
-            return 1
+
+        with subprocess.Popen(arguments) as p3:
+            returncode = p3.wait()
+            if returncode != 0:
+                print("ERROR: Creation of main report failed.", file=sys.stderr)
+                return 1
 
     return 0
 

python/grass/gunittest/reporters.py

@@ -160,19 +160,19 @@ def get_svn_revision():
     """
     # TODO: here should be starting directory
     # but now we are using current as starting
-    p = subprocess.Popen(
+    with subprocess.Popen(
         ["svnversion", "."], stdout=subprocess.PIPE, stderr=subprocess.PIPE
-    )
-    stdout, stderr = p.communicate()
-    rc = p.poll()
-    if not rc:
+    ) as p:
+        stdout, stderr = p.communicate()
+        rc = p.poll()
+        if rc:
+            return None
         stdout = stdout.strip()
         stdout = stdout.removesuffix("M")
         if ":" in stdout:
             # the first one is the one of source code
             stdout = stdout.split(":")[0]
         return stdout
-    return None
 
 
 def get_svn_info():

python/grass/imaging/images2avi.py

@@ -180,22 +180,22 @@ def readAvi(filename, asNumpy=True):
 
     # Run ffmpeg
     command = "ffmpeg -i input.avi im%d.jpg"
-    S = subprocess.Popen(
-        command, shell=True, cwd=tempDir, stdout=subprocess.PIPE, stderr=subprocess.PIPE
-    )
-
-    # Show what mencodec has to say
-    outPut = S.stdout.read()
-
-    if S.wait():
-        # An error occurred, show
-        print(outPut)
-        print(S.stderr.read())
-        # Clean up
-        _cleanDir(tempDir)
-        msg = "Could not read avi."
-        raise RuntimeError(msg)
-
+    with subprocess.Popen(
+        command,
+        cwd=tempDir,
+        stdout=subprocess.PIPE,
+        stderr=subprocess.PIPE,
+    ) as S:
+        # Show what mencodec has to say
+        outPut = S.stdout.read()
+        if S.wait():
+            # An error occurred, show
+            print(outPut)
+            print(S.stderr.read())
+            # Clean up
+            _cleanDir(tempDir)
+            msg = "Could not read avi."
+            raise RuntimeError(msg)
     # Read images
     images = images2ims.readIms(os.path.join(tempDir, "im*.jpg"), asNumpy)
     # Clean up

python/grass/imaging/images2gif.py

@@ -612,11 +612,8 @@ def writeGifVisvis(
     images = gifWriter.convertImagesToPIL(images, dither, nq)
 
     # Write
-    fp = open(filename, "wb")
-    try:
+    with open(filename, "wb") as fp:
         gifWriter.writeGifToFile(fp, images, duration, loops, xy, dispose)
-    finally:
-        fp.close()
 
 
 def readGif(filename, asNumpy=True):

python/grass/imaging/images2swf.py

@@ -69,6 +69,7 @@
 
 import os
 import zlib
+from pathlib import Path
 
 try:
     import numpy as np
@@ -838,11 +839,8 @@ def writeSwf(filename, images, duration=0.1, repeat=True):
         taglist.append(DoActionTag("stop"))
 
     # Build file
-    fp = open(filename, "wb")
-    try:
+    with open(filename, "wb") as fp:
         buildFile(fp, taglist, nframes=nframes, framesize=wh, fps=fps)
-    finally:
-        fp.close()
 
 
 def _readPixels(bb, i, tagType, L1):
@@ -923,67 +921,60 @@ def readSwf(filename, asNumpy=True):
     images = []
 
     # Open file and read all
-    fp = open(filename, "rb")
-    bb = fp.read()
-
-    try:
-        # Check opening tag
-        tmp = bb[0:3].decode("ascii", "ignore")
-        if tmp.upper() == "FWS":
-            pass  # ok
-        elif tmp.upper() == "CWS":
-            # Decompress movie
-            bb = bb[:8] + zlib.decompress(bb[8:])
+    bb = Path(filename).read_bytes()
+    # Check opening tag
+    tmp = bb[0:3].decode("ascii", "ignore")
+    if tmp.upper() == "FWS":
+        pass  # ok
+    elif tmp.upper() == "CWS":
+        # Decompress movie
+        bb = bb[:8] + zlib.decompress(bb[8:])
+    else:
+        raise OSError("Not a valid SWF file: " + str(filename))
+
+    # Set filepointer at first tag (skipping framesize RECT and two uin16's
+    i = 8
+    nbits = bitsToInt(bb[i : i + 1], 5)  # skip FrameSize
+    nbits = 5 + nbits * 4
+    Lrect = nbits / 8.0
+    if Lrect % 1:
+        Lrect += 1
+    Lrect = int(Lrect)
+    i += Lrect + 4
+
+    # Iterate over the tags
+    counter = 0
+    while True:
+        counter += 1
+
+        # Get tag header
+        head = bb[i : i + 6]
+        if not head:
+            break  # Done (we missed end tag)
+
+        # Determine type and length
+        T, L1, L2 = getTypeAndLen(head)
+        if not L2:
+            print("Invalid tag length, could not proceed")
+            break
+        # print(T, L2)
+
+        # Read image if we can
+        if T in {20, 36}:
+            im = _readPixels(bb, i + 6, T, L1)
+            if im is not None:
+                images.append(im)
+        elif T in {6, 21, 35, 90}:
+            print("Ignoring JPEG image: cannot read JPEG.")
         else:
-            raise OSError("Not a valid SWF file: " + str(filename))
-
-        # Set filepointer at first tag (skipping framesize RECT and two uin16's
-        i = 8
-        nbits = bitsToInt(bb[i : i + 1], 5)  # skip FrameSize
-        nbits = 5 + nbits * 4
-        Lrect = nbits / 8.0
-        if Lrect % 1:
-            Lrect += 1
-        Lrect = int(Lrect)
-        i += Lrect + 4
-
-        # Iterate over the tags
-        counter = 0
-        while True:
-            counter += 1
-
-            # Get tag header
-            head = bb[i : i + 6]
-            if not head:
-                break  # Done (we missed end tag)
-
-            # Determine type and length
-            T, L1, L2 = getTypeAndLen(head)
-            if not L2:
-                print("Invalid tag length, could not proceed")
-                break
-            # print(T, L2)
-
-            # Read image if we can
-            if T in [20, 36]:
-                im = _readPixels(bb, i + 6, T, L1)
-                if im is not None:
-                    images.append(im)
-            elif T in [6, 21, 35, 90]:
-                print("Ignoring JPEG image: cannot read JPEG.")
-            else:
-                pass  # Not an image tag
-
-            # Detect end tag
-            if T == 0:
-                break
-
-            # Next tag!
-            i += L2
+            pass  # Not an image tag
 
-    finally:
-        fp.close()
+        # Detect end tag
+        if T == 0:
+            break
 
+        # Next tag!
+        i += L2
     # Convert to normal PIL images if needed
     if not asNumpy:
         images2 = images

python/grass/script/core.py

@@ -933,15 +933,14 @@ def parser() -> tuple[dict[str, str], dict[str, bool]]:
             argv[0] = os.path.join(sys.path[0], name)
 
     prog = "g.parser.exe" if sys.platform == "win32" else "g.parser"
-    p = subprocess.Popen([prog, "-n"] + argv, stdout=subprocess.PIPE)
-    s = p.communicate()[0]
-    lines = s.split(b"\0")
-
-    if not lines or lines[0] != b"@ARGS_PARSED@":
-        stdout = os.fdopen(sys.stdout.fileno(), "wb")
-        stdout.write(s)
-        sys.exit(p.returncode)
-    return _parse_opts(lines[1:])
+    with subprocess.Popen([prog, "-n"] + argv, stdout=subprocess.PIPE) as p:
+        s = p.communicate()[0]
+        lines = s.split(b"\0")
+        if not lines or lines[0] != b"@ARGS_PARSED@":
+            stdout = os.fdopen(sys.stdout.fileno(), "wb")
+            stdout.write(s)
+            sys.exit(p.returncode)
+        return _parse_opts(lines[1:])
 
 
 # interface to g.tempfile