refactor: list_instances (#4329)

* feat(cli): refactor list_instances for improved process validation and instance retrieval

* feat(cli): refactor list_instances to utilize core functionality for instance retrieval

* fix(cli): optimize process iteration by adding info attribute for better performance

* chore: adding changelog file 4329.miscellaneous.md [dependabot-skip]

* fix(cli): handle AccessDenied exception in get_ansys_process_from_port function

* test(cli): add comprehensive tests for get_ansys_process_from_port function

* refactor: move get_ansys_process_from_port function to core module

---------

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

Author: germa89

doc/changelog.d/4329.miscellaneous.md

@@ -0,0 +1 @@
+List_instances

src/ansys/mapdl/core/cli/__init__.py

@@ -20,12 +20,16 @@
 # OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 # SOFTWARE.
 
-from ansys.mapdl.core import _HAS_CLICK
+try:
+    import click  # noqa: F401
+
+    _HAS_CLICK = True
+except ImportError:  # pragma: no cover
+    _HAS_CLICK = False
 
 if _HAS_CLICK:
     ###################################
     # PyMAPDL CLI
-    import click
 
     @click.group(invoke_without_command=True)
     @click.pass_context
@@ -38,9 +42,9 @@ def main(ctx: click.Context):
     from ansys.mapdl.core.cli.stop import stop as stop_cmd
 
     main.add_command(convert_cmd, name="convert")
+    main.add_command(list_instances, name="list")
     main.add_command(start_cmd, name="start")
     main.add_command(stop_cmd, name="stop")
-    main.add_command(list_instances, name="list")
 
 
 else:

src/ansys/mapdl/core/cli/core.py

@@ -0,0 +1,127 @@
+# Copyright (C) 2016 - 2025 ANSYS, Inc. and/or its affiliates.
+# SPDX-License-Identifier: MIT
+#
+#
+# Permission is hereby granted, free of charge, to any person obtaining a copy
+# of this software and associated documentation files (the "Software"), to deal
+# in the Software without restriction, including without limitation the rights
+# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+# copies of the Software, and to permit persons to whom the Software is
+# furnished to do so, subject to the following conditions:
+#
+# The above copyright notice and this permission notice shall be included in all
+# copies or substantial portions of the Software.
+#
+# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+# SOFTWARE.
+
+"""
+Minimal core functionality for CLI operations.
+This module avoids importing heavy dependencies like pandas, numpy, etc.
+"""
+
+from typing import Any, Dict, List
+
+import psutil
+
+
+def is_valid_ansys_process_name(name: str) -> bool:
+    """Check if process name indicates ANSYS/MAPDL"""
+    return ("ansys" in name.lower()) or ("mapdl" in name.lower())
+
+
+def is_alive_status(status) -> bool:
+    """Check if process status indicates alive"""
+    return status in [
+        psutil.STATUS_RUNNING,
+        psutil.STATUS_IDLE,
+        psutil.STATUS_SLEEPING,
+    ]
+
+
+def get_mapdl_instances() -> List[Dict[str, Any]]:
+    """Get list of MAPDL instances with minimal data"""
+    instances = []
+
+    for proc in psutil.process_iter(attrs=["name"]):
+        name = proc.info["name"]
+        if not is_valid_ansys_process_name(name):
+            continue
+
+        try:
+            status = proc.status()
+            if not is_alive_status(status):
+                continue
+
+            cmdline = proc.cmdline()
+            if "-grpc" not in cmdline:
+                continue
+
+            # Get port from cmdline
+            port = None
+            try:
+                ind_grpc = cmdline.index("-port")
+                port = int(cmdline[ind_grpc + 1])
+            except (ValueError, IndexError):
+                continue
+
+            children = proc.children(recursive=True)
+            is_instance = len(children) >= 2
+
+            cwd = proc.cwd()
+            instances.append(
+                {
+                    "name": name,
+                    "status": status,
+                    "port": port,
+                    "pid": proc.pid,
+                    "cmdline": cmdline,
+                    "is_instance": is_instance,
+                    "cwd": cwd,
+                }
+            )
+
+        except (psutil.NoSuchProcess, psutil.ZombieProcess, psutil.AccessDenied):
+            continue
+
+    return instances
+
+
+def get_ansys_process_from_port(port: int):
+    import socket
+
+    import psutil
+
+    from ansys.mapdl.core.cli.core import is_alive_status, is_valid_ansys_process_name
+
+    # Filter by name first
+    potential_procs = []
+    for proc in psutil.process_iter(attrs=["name"]):
+        name = proc.info["name"]
+        if is_valid_ansys_process_name(name):
+            potential_procs.append(proc)
+
+    for proc in potential_procs:
+        try:
+            status = proc.status()
+            if not is_alive_status(status):
+                continue
+            cmdline = proc.cmdline()
+            if "-grpc" not in cmdline:
+                continue
+            # Check if listening on the port
+            connections = proc.connections()
+            for conn in connections:
+                if (
+                    conn.status == "LISTEN"
+                    and conn.family == socket.AF_INET
+                    and conn.laddr[1] == port
+                ):
+                    return proc
+        except (psutil.NoSuchProcess, psutil.ZombieProcess, psutil.AccessDenied):
+            continue

src/ansys/mapdl/core/cli/list_instances.py

@@ -66,46 +66,16 @@
     help="Print running location info.",
 )
 def list_instances(instances, long, cmd, location) -> None:
-    import psutil
+    return _list_instances(instances, long, cmd, location)
+
+
+def _list_instances(instances, long, cmd, location):
     from tabulate import tabulate
 
+    from ansys.mapdl.core.cli.core import get_mapdl_instances
+
     # Assuming all ansys processes have -grpc flag
-    mapdl_instances = []
-
-    def is_grpc_based(proc):
-        cmdline = proc.cmdline()
-        return "-grpc" in cmdline
-
-    def get_port(proc):
-        cmdline = proc.cmdline()
-        ind_grpc = cmdline.index("-port")
-        return cmdline[ind_grpc + 1]
-
-    def is_valid_process(proc):
-        valid_status = proc.status() in [
-            psutil.STATUS_RUNNING,
-            psutil.STATUS_IDLE,
-            psutil.STATUS_SLEEPING,
-        ]
-        valid_ansys_process = ("ansys" in proc.name().lower()) or (
-            "mapdl" in proc.name().lower()
-        )
-        return valid_status and valid_ansys_process and is_grpc_based(proc)
-
-    for proc in psutil.process_iter():
-        # Check if the process is running and not suspended
-        try:
-            if is_valid_process(proc):
-                # Checking the number of children we infer if the process is the main process,
-                # or one of the main process thread.
-                if len(proc.children(recursive=True)) < 2:
-                    proc.ansys_instance = False
-                else:
-                    proc.ansys_instance = True
-                mapdl_instances.append(proc)
-
-        except (psutil.NoSuchProcess, psutil.ZombieProcess) as e:
-            continue
+    mapdl_instances = get_mapdl_instances()
 
     # printing
     if long:
@@ -124,23 +94,23 @@ def is_valid_process(proc):
 
     table = []
     for each_p in mapdl_instances:
-        if instances and not each_p.ansys_instance:
+        if instances and not each_p.get("is_instance", False):
             # Skip child processes if only printing instances
             continue
 
         proc_line = []
-        proc_line.append(each_p.name())
+        proc_line.append(each_p["name"])
 
         if not instances:
-            proc_line.append(each_p.ansys_instance)
+            proc_line.append(each_p.get("is_instance", False))
 
-        proc_line.extend([each_p.status(), get_port(each_p), each_p.pid])
+        proc_line.extend([each_p["status"], each_p["port"], each_p["pid"]])
 
         if cmd:
-            proc_line.append(" ".join(each_p.cmdline()))
+            proc_line.append(" ".join(each_p["cmdline"]))
 
         if location:
-            proc_line.append(each_p.cwd())
+            proc_line.append(each_p["cwd"])
 
         table.append(proc_line)