feat: adding muted context manager

doc/changelog.d/3760.documentation.md

@@ -0,0 +1 @@
+feat: adding `muted` context manager

doc/source/api/mapdl.rst

@@ -54,6 +54,7 @@
    Mapdl.logger
    Mapdl.mapdl_on_hpc
    Mapdl.mute
+   Mapdl.muted
    Mapdl.name
    Mapdl.non_interactive
    Mapdl.on_docker

src/ansys/mapdl/core/mapdl_core.py

@@ -838,6 +838,21 @@ def non_interactive(self):
         """
         return self._non_interactive(self)
 
+    @property
+    def muted(self):
+        """Context manager that suppress all output from MAPDL
+
+        Use the `muted` context manager to suppress all the output. Similar to
+        setting `mapdl.mute = True` but only for the context manager.
+
+        Examples
+        --------
+        >>> with mapdl.muted:
+        ...    mapdl.run("/SOLU") # This call is muted
+
+        """
+        return self._muted(self)
+
     @property
     def parameters(self) -> "Parameters":
         """Collection of MAPDL parameters.
@@ -1187,10 +1202,6 @@ def _mesh(self) -> "Archive":
             nblock_filename = os.path.join(self.directory, "nblock.cdb")
 
             # must have all nodes elements are using selected
-            if hasattr(self, "mute"):
-                old_mute = self.mute
-                self.mute = True
-
             self.cm("__NODE__", "NODE", mute=True)
             self.nsle("S", mute=True)
             self.cdwrite("db", arch_filename, mute=True)
@@ -1201,9 +1212,6 @@ def _mesh(self) -> "Archive":
             self.cdwrite("db", nblock_filename, mute=True)
             self.cmsel("S", "__ELEM__", "ELEM", mute=True)
 
-            if hasattr(self, "mute"):
-                self.mute = old_mute
-
             self._archive_cache = Archive(arch_filename, parse_vtk=False, name="Mesh")
             grid = self._archive_cache._parse_vtk(additional_checking=True)
             self._archive_cache._grid = grid
@@ -1530,6 +1538,19 @@ def __exit__(self, *args):
         def _cached_routine(self):
             return self._parent()._cached_routine
 
+    class _muted:
+        def __init__(self, parent):
+            self._parent = weakref.ref(parent)
+            self.old_value = None
+
+        def __enter__(self):
+            self.old_value = self._parent().mute
+            self._parent().mute = True
+
+        def __exit__(self, *args):
+            self._parent().mute = self.old_value
+            self.old_value = None
+
     def run_as_routine(self, routine):
         """
         Runs a command or commands at a routine and then revert to the prior routine.

src/ansys/mapdl/core/mapdl_grpc.py

@@ -438,6 +438,7 @@
         self._channel_state: grpc.ChannelConnectivity = (
             grpc.ChannelConnectivity.CONNECTING
         )
+        self._time_step_stream: Optional[int] = None
 
         if channel is None:
             self._log.debug("Creating channel to %s:%s", ip, port)
@@ -812,6 +813,7 @@
 
     @mute.setter
     def mute(self, value):
+        self._log.debug(f"Mute value has been changed to '{value}'.")
         self._mute = value
 
     def __repr__(self):
@@ -2035,20 +2037,24 @@
         if time_step is None:
             if DEFAULT_TIME_STEP_STREAM is not None:
                 time_step = DEFAULT_TIME_STEP_STREAM
-            elif self.platform == "windows":
-                time_step = DEFAULT_TIME_STEP_STREAM_NT
-            elif self.platform == "linux":
-                time_step = DEFAULT_TIME_STEP_STREAM_POSIX
+            elif self._time_step_stream is not None:
+                time_step = self._time_step_stream
             else:
-                raise ValueError(
-                    f"The MAPDL platform ('{self.platform}') is not recognaised."
-                )
+                if self.platform == "windows":
+                    time_step = DEFAULT_TIME_STEP_STREAM_NT
+                elif self.platform == "linux":
+                    time_step = DEFAULT_TIME_STEP_STREAM_POSIX
+                else:
+                    raise ValueError(
+                        f"The MAPDL platform ('{self.platform}') is not recognaised."
+                    )
 
         else:
             if time_step <= 0:
                 raise ValueError("``time_step`` argument must be greater than 0``")
 
         self.logger.debug(f"The time_step argument is set to: {time_step}")
+        self._time_step_stream = time_step
         return time_step
 
     def _get_file_path(self, fname: str, progress_bar: bool = False) -> str:
@@ -2262,41 +2268,39 @@
 
         self._get_lock = True
         try:
-            getresponse = self._stub.Get(pb_types.GetRequest(getcmd=cmd))
+            response = self._stub.Get(pb_types.GetRequest(getcmd=cmd))
         finally:
             self._get_lock = False
 
-        if getresponse.type == 0:
+        if response.type == 0:
             self._log.debug(
                 "The 'grpc' get method seems to have failed. Trying old implementation for more verbose output."
             )
 
-            out = self.run("*GET,__temp__," + cmd)
-            self._log.debug(f"Default *get output:\n{out}")
+            response = self.run("*GET,__temp__," + cmd, mute=False)
+            self._log.debug(f"Default *get output:\n{response}")
 
-            if out:
+            if response:
                 from ansys.mapdl.core.misc import is_float
 
-                if "VALUE=" in out:
-                    out = out.split("VALUE=")[1].strip()
+                if "VALUE=" in response:
+                    response = response.split("VALUE=")[1].strip()
 
-                if is_float(out):
-                    return float(out)
+                if is_float(response):
+                    return float(response)
                 else:
-                    return out
-            elif out is None:
-                return None
+                    return response
             else:
-                raise MapdlError(f"Error when processing '*get' request output.\n{out}")
+                raise MapdlError(
+                    f"Error when processing '*get' request output.\n{response}"
+                )
 
-        if getresponse.type == 1:
-            return getresponse.dval
-        elif getresponse.type == 2:
-            return getresponse.sval
+        if response.type == 1:
+            return response.dval
+        elif response.type == 2:
+            return response.sval
 
-        raise MapdlRuntimeError(
-            f"Unsupported type {getresponse.type} response from MAPDL"
-        )
+        raise MapdlRuntimeError(f"Unsupported type {response.type} response from MAPDL")
 
     def download_project(
         self,

src/ansys/mapdl/core/parameters.py

@@ -663,11 +663,6 @@ def _set_array_chain(self, name, arr, idim, jdim, kdim):
         elif arr.ndim == 2:
             arr = np.expand_dims(arr, 2)
 
-        # backwards compatibility with CORBA
-        if hasattr(self._mapdl, "mute"):
-            old_mute = self._mapdl.mute
-            self._mapdl.mute = True
-
         with self._mapdl.non_interactive:
             self._mapdl.dim(name, imax=idim, jmax=jdim, kmax=kdim)
             for i in range(idim):
@@ -676,9 +671,6 @@ def _set_array_chain(self, name, arr, idim, jdim, kdim):
                         index = f"{i + 1},{j + 1},{k + 1}"
                         self._mapdl.run(f"{name}({index})={arr[i, j, k]}")
 
-        if hasattr(self._mapdl, "mute"):
-            self._mapdl.mute = old_mute
-
     def _write_numpy_array(self, filename, arr):
         """Write a numpy array to disk"""
         if arr.dtype != np.double: