chore: Code maintenance (#937)

Co-authored-by: pyansys-ci-bot <[email protected]>

doc/changelog.d/937.maintenance.md

@@ -0,0 +1 @@
+Code maintenance

src/ansys/mechanical/core/embedding/initializer.py

@@ -164,8 +164,12 @@ def initialize(version: int = None):
     __check_for_mechanical_env()  # checks for mechanical-env in linux embedding
 
     global INITIALIZED_VERSION
-    if INITIALIZED_VERSION != None:
-        assert INITIALIZED_VERSION == version
+    if INITIALIZED_VERSION is not None:
+        if INITIALIZED_VERSION != version:
+            raise ValueError(
+                f"Initialized version {INITIALIZED_VERSION} "
+                f"does not match the expected version {version}."
+            )
         return
 
     if version == None:

src/ansys/mechanical/core/embedding/viz/utils.py

@@ -41,7 +41,8 @@ def _reshape_3cols(arr: np.array, name: str = "array"):
     """
     err = f"{name} must be of the form (x0,y0,z0,x1,y1,z1,...,xn,yn,zn).\
         Given {name} are not divisible by 3!"
-    assert arr.size % 3 == 0, err
+    if arr.size % 3 != 0:
+        raise ValueError(err)
     numrows = int(arr.size / 3)
     numcols = 3
     arr = np.reshape(arr, (numrows, numcols))

src/ansys/mechanical/core/mechanical.py

@@ -508,8 +508,8 @@ def name(self):
                     return f"GRPC_{self._channel._channel._channel.target().decode()}"
                 else:
                     return f"GRPC_{self._channel._channel.target().decode()}"
-        except Exception:  # pragma: no cover
-            pass
+        except Exception as e:  # pragma: no cover
+            LOG.error(f"Error getting the Mechanical instance name: {str(e)}")
 
         return f"GRPC_instance_{id(self)}"  # pragma: no cover
 

src/ansys/mechanical/core/pool.py

@@ -583,8 +583,8 @@ def threaded_exit(index, instance_local):
             if instance_local:
                 try:
                     instance_local.exit()
-                except:  # pragma: no cover
-                    pass
+                except Exception as e:  # pragma: no cover
+                    LOG.error(f"Error while exiting instance {str(instance_local)}: {str(e)}")
                 self._instances[index] = None
                 LOG.debug(f"Exited instance: {str(instance_local)}")
 

src/ansys/mechanical/core/run.py

@@ -54,7 +54,8 @@ async def _read_and_display(cmd, env, do_display: bool):
     }
     while tasks:
         done, pending = await asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED)
-        assert done
+        if not done:
+            raise RuntimeError("Subprocess read failed: No tasks completed.")
         for future in done:
             buf, stream, display = tasks.pop(future)
             line = future.result()