refactor: modifying subprocess calls and removing try except continue statements

doc/changelog.d/3474.added.md

@@ -0,0 +1 @@
+refactor: modifying ``subprocess`` calls and removing ``try except continue`` statements

minimum_requirements.txt

@@ -3,4 +3,4 @@ importlib-metadata==8.5.0
 numpy==2.1.2
 platformdirs==4.3.6
 psutil==6.1.0
-pyansys-tools-versioning==0.6.0
+pyansys-tools-versioning==0.6.0

src/ansys/mapdl/core/examples/downloads.py

@@ -25,13 +25,13 @@
 from functools import wraps
 import os
 import shutil
-import urllib.request
 import zipfile
 
 try:
     import requests
 
     _HAS_REQUESTS = True
+
 except ModuleNotFoundError:
     _HAS_REQUESTS = False
 
@@ -83,14 +83,8 @@ def _get_file_url(filename, directory=None):
 
 
 def _check_url_exist(url):
-    if not _HAS_REQUESTS:
-        raise ModuleNotFoundError("Examples module requires request module")
-
-    response = requests.get(url)
-    if response.status_code == 200:
-        return [True]
-    else:
-        return [False]
+    response = requests.get(url, timeout=10)  # 10 seconds timeout
+    return response.status_code == 200
 
 
 @check_directory_exist(pymapdl.EXAMPLES_PATH)
@@ -103,24 +97,28 @@ def _retrieve_file(url, filename, _test=False):
     local_path = os.path.join(pymapdl.EXAMPLES_PATH, os.path.basename(filename))
     local_path_no_zip = local_path.replace(".zip", "")
     if os.path.isfile(local_path_no_zip) or os.path.isdir(local_path_no_zip):
-        return local_path_no_zip, None
+        return local_path_no_zip
 
     # Perform download
-    saved_file, resp = urllib.request.urlretrieve(url)
-    shutil.move(saved_file, local_path)
+    requested_file = requests.get(url, timeout=10)
+    requested_file.raise_for_status()
+
+    with open(local_path, "wb") as f:
+        f.write(requested_file.content)
+
     if get_ext(local_path) in [".zip"]:
         _decompress(local_path)
         local_path = local_path[:-4]
-    return local_path, resp
+    return local_path
 
 
 def _download_file(filename, directory=None, _test=False):
     url = _get_file_url(filename, directory)
     try:
         return _retrieve_file(url, filename, _test)
-    except Exception as e:  # Genering exception
-        raise RuntimeError(
-            "For the reason mentioned below, retrieving the file from internet failed.\n"
+    except requests.exceptions.HTTPError as e:
+        raise requests.exceptions.HTTPError(
+            "Retrieving the file from internet failed.\n"
             "You can download this file from:\n"
             f"{url}\n"
             "\n"
@@ -140,27 +138,27 @@ def download_bracket():
     '/home/user/.local/share/ansys_mapdl_core/examples/bracket.iges'
 
     """
-    return _download_file("bracket.iges", "geometry")[0]
+    return _download_file("bracket.iges", "geometry")
 
 
 def download_tech_demo_data(example, filename):
     """Download Tech Demos external data."""
     example = "tech_demos/" + example
-    return _download_file(filename=filename, directory=example)[0]
+    return _download_file(filename=filename, directory=example)
 
 
 def download_vtk_rotor():
     """Download rotor vtk file."""
-    return _download_file("rotor.vtk", "geometry")[0]
+    return _download_file("rotor.vtk", "geometry")
 
 
 def _download_rotor_tech_demo_vtk():
     """Download the rotor surface VTK file."""
-    return _download_file("rotor2.vtk", "geometry")[0]
+    return _download_file("rotor2.vtk", "geometry")
 
 
 def download_example_data(filename, directory=None):
-    return _download_file(filename, directory=directory)[0]
+    return _download_file(filename, directory=directory)
 
 
 def download_manifold_example_data() -> dict:
@@ -188,10 +186,10 @@ def download_manifold_example_data() -> dict:
     return {
         "geometry": _download_file(
             filename="manifold_geometry.anf", directory=files_dir
-        )[0],
+        ),
         "mapping_data": _download_file(
             filename="manifold_cht-final_temp.csv", directory=files_dir
-        )[0],
+        ),
     }
 
 
@@ -219,6 +217,6 @@ def download_cfx_mapping_example_data() -> dict:
     return {
         "data": _download_file(
             filename="11_blades_mode_1_ND_0.csv", directory=files_dir
-        )[0],
-        "model": _download_file(filename="ExampleMapping.db", directory=files_dir)[0],
+        ),
+        "model": _download_file(filename="ExampleMapping.db", directory=files_dir),
     }

src/ansys/mapdl/core/launcher.py

@@ -166,8 +166,7 @@ def _is_ubuntu() -> bool:
         return False
 
     proc = subprocess.Popen(
-        "awk -F= '/^NAME/{print $2}' /etc/os-release",
-        shell=True,
+        ["awk", "-F=", "/^NAME/{print $2}", "/etc/os-release"],
         stdout=subprocess.PIPE,
     )
     if "ubuntu" in proc.stdout.read().decode().lower():
@@ -309,7 +308,7 @@ def generate_mapdl_launch_command(
     ram: Optional[int] = None,
     port: int = MAPDL_DEFAULT_PORT,
     additional_switches: str = "",
-) -> str:
+) -> list[str]:
     """Generate the command line to start MAPDL in gRPC mode.
 
     Parameters
@@ -346,7 +345,7 @@ def generate_mapdl_launch_command(
 
     Returns
     -------
-    str
+    list[str]
         Command
 
     """
@@ -369,10 +368,10 @@ def generate_mapdl_launch_command(
 
     # Windows will spawn a new window, special treatment
     if os.name == "nt":
+        exec_file = f'"{exec_file}"'
         # must start in batch mode on windows to hide APDL window
         tmp_inp = ".__tmp__.inp"
         command_parm = [
-            '"%s"' % exec_file,
             job_sw,
             cpu_sw,
             ram_sw,
@@ -388,7 +387,6 @@ def generate_mapdl_launch_command(
 
     else:  # linux
         command_parm = [
-            '"%s"' % exec_file,
             job_sw,
             cpu_sw,
             ram_sw,
@@ -398,16 +396,19 @@ def generate_mapdl_launch_command(
         ]
 
     command_parm = [
-        each for each in command_parm if command_parm
+        each for each in command_parm if each.strip()
     ]  # cleaning empty args.
-    command = " ".join(command_parm)
 
-    LOG.debug(f"Generated command: {command}")
-    return command
+    # removing spaces in cells
+    command_parm = " ".join(command_parm).split(" ")
+    command_parm.insert(0, f"{exec_file}")
+
+    LOG.debug(f"Generated command: {' '.join(command_parm)}")
+    return command_parm
 
 
 def launch_grpc(
-    cmd: str,
+    cmd: list[str],
     run_location: str = None,
     env_vars: Optional[Dict[str, str]] = None,
 ) -> subprocess.Popen:
@@ -442,15 +443,14 @@ def launch_grpc(
 
     if os.name == "nt":
         # getting tmp file name
-        tmp_inp = cmd.split()[cmd.split().index("-i") + 1]
+        tmp_inp = cmd[cmd.index("-i") + 1]
         with open(os.path.join(run_location, tmp_inp), "w") as f:
             f.write("FINISH\r\n")
             LOG.debug(f"Writing temporary input file: {tmp_inp} with 'FINISH' command.")
 
     LOG.debug("MAPDL starting in background.")
     process = subprocess.Popen(
         cmd,
-        shell=os.name != "nt",
         cwd=run_location,
         stdin=subprocess.DEVNULL,
         stdout=subprocess.PIPE,

src/ansys/mapdl/core/licensing.py

@@ -329,11 +329,10 @@ def _checkout_license(self, lic, host=None, port=2325):
 
         tstart = time.time()
         process = subprocess.Popen(
-            f'"{ansysli_util_path}" -checkout {lic}',
+            [f'"{ansysli_util_path}"', "-checkout", f"{lic}"],
             stdout=subprocess.PIPE,
             stderr=subprocess.STDOUT,
             env=env,
-            shell=True,
         )
         output = process.stdout.read().decode()
 

src/ansys/mapdl/core/mapdl_console.py

@@ -28,6 +28,7 @@
 import re
 import time
 
+from ansys.mapdl.core import LOG
 from ansys.mapdl.core.errors import MapdlExitedError, MapdlRuntimeError
 from ansys.mapdl.core.mapdl import MapdlBase
 from ansys.mapdl.core.misc import requires_package
@@ -284,8 +285,8 @@ def exit(self, close_log=True, timeout=3):
             try:
                 self._process.sendline("FINISH")
                 self._process.sendline("EXIT")
-            except:
-                pass
+            except Exception as e:
+                LOG.warning(f"Unable to exit ANSYS MAPDL: {e}")
 
         if close_log:
             self._close_apdl_log()

src/ansys/mapdl/core/mapdl_core.py

@@ -500,8 +500,10 @@
         while (not self._path and i > 5) or i == 0:
             try:
                 self._path = self.inquire("", "DIRECTORY")
-            except Exception:  # pragma: no cover
-                pass
+            except Exception as e:  # pragma: no cover
+                logger.warning(
+                    f"Failed to get the directory due to the following error: {e}"
+                )
             i += 1
             if not self._path:  # pragma: no cover
                 time.sleep(0.1)
@@ -676,8 +678,8 @@
         """
         try:
             self._jobname = self.inquire("", "JOBNAME")
-        except Exception:
-            pass
+        except Exception as e:
+            logger.warning(f"Failed to get the jobname due to the following error: {e}")
         return self._jobname
 
     @jobname.setter
@@ -1693,11 +1695,21 @@
                 "MAPDL GUI has been opened using 'inplace' kwarg. "
                 f"The changes you make will overwrite the files in {run_dir}."
             )
+        add_sw = add_sw.split()
+        exec_array = [
+            f"{exec_file}",
+            "-g",
+            "-j",
+            f"{name}",
+            "-np",
+            f"{nproc}",
+            *add_sw,
+        ]
 
         call(
-            f'cd "{run_dir}" && "{exec_file}" -g -j {name} -np {nproc} {add_sw}',
-            shell=True,
+            exec_array,
             stdout=DEVNULL,
+            cwd=run_dir,
         )
 
         # Going back
@@ -2298,7 +2310,7 @@
                 try:  # logger might be closed
                     if self._log is not None:
                         self._log.error("exit: %s", str(e))
-                except Exception:
+                except ValueError:
                     pass
 
     def _cleanup_loggers(self):

src/ansys/mapdl/core/mapdl_extended.py

@@ -807,9 +807,11 @@
 
                 for surf in surfs:
                     anum = np.unique(surf["entity_num"])
-                    assert (
-                        len(anum) == 1
-                    ), f"The pv.Unstructured from the entity {anum[0]} contains entities from other entities {anum}"  # Sanity check
+                    if len(anum) != 1:
+                        raise RuntimeError(
+                            f"The pv.Unstructured from the entity {anum[0]} contains entities"
+                            f"from other entities {anum}"  # Sanity check
+                        )
 
                     area = surf.extract_cells(surf["entity_num"] == anum)
                     centers.append(area.center)

src/ansys/mapdl/core/mapdl_grpc.py

@@ -881,8 +881,9 @@
                 self.prep7()
                 success = True
                 break
-            except:
-                pass
+            except MapdlRuntimeError:
+                time.sleep(1)
+                warn("PyMAPDL is taking longer than expected to connect to the server.")
 
         if not success:
             raise MapdlConnectionError("Unable to reconnect to MAPDL")
@@ -1025,7 +1026,7 @@
         return "".join(response)
 
     def _threaded_heartbeat(self):
-        """To be called from a thread to verify mapdl instance is alive"""
+        """To be called from a thread to verify MAPDL instance is alive"""
         self._initialised.set()
         while True:
             if self._exited:
@@ -1038,7 +1039,7 @@
             except ReferenceError:
                 break
             except Exception:
-                continue
+                self._log.debug("Checking if MAPDL instance is still alive.")
 
     @protect_from(ValueError, "I/O operation on closed file.")
     def exit(self, save=False, force=False, **kwargs):

src/ansys/mapdl/core/misc.py

@@ -28,7 +28,6 @@
 import os
 from pathlib import Path
 import platform
-import random
 import socket
 import string
 import tempfile
@@ -378,7 +377,9 @@ def is_float(input_string):
 
 def random_string(stringLength=10, letters=string.ascii_lowercase):
     """Generate a random string of fixed length"""
-    return "".join(random.choice(letters) for i in range(stringLength))
+    import secrets
+
+    return "".join(secrets.choice(letters) for _ in range(stringLength))
 
 
 def _check_has_ansys():