feat: Raise an exception if license is inaccessible

src/ansys/fluent/core/__init__.py

@@ -82,6 +82,9 @@ def version_info() -> str:
     platformdirs.user_documents_dir(), "ansys_fluent_core_examples"
 )
 
+# Working directory for the Fluent client
+CWD = None
+
 # Host path which is mounted to the container
 CONTAINER_MOUNT_SOURCE = None
 

src/ansys/fluent/core/launcher/error_handler.py

@@ -1,10 +1,21 @@
 """Provides a module for customized error handling."""
 
+import glob
+import os
+
+import ansys.fluent.core as pyfluent
 from ansys.fluent.core.exceptions import InvalidArgument
 from ansys.fluent.core.launcher import launcher_utils
 from ansys.fluent.core.utils.fluent_version import FluentVersion
 
 
+class InaccessibleAnsysLicense(RuntimeError):
+    """Raised when license is inaccessible."""
+
+    def __init__(self):
+        super().__init__("Ansys license is inaccessible.")
+
+
 class InvalidPassword(ValueError):
     """Raised when password is invalid."""
 
@@ -80,3 +91,21 @@ def _process_kwargs(kwargs):
             raise UnexpectedKeywordArgument(
                 f"launch_fluent() got an unexpected keyword argument {next(iter(kwargs))}"
             )
+
+
+def _check_license():
+    if pyfluent.CWD:
+        folder = pyfluent.CWD
+    elif pyfluent.CONTAINER_MOUNT_SOURCE:
+        folder = pyfluent.CONTAINER_MOUNT_SOURCE
+    else:
+        folder = os.getcwd()
+    file_pattern = os.path.join(folder, "*.trn")
+    files = glob.glob(file_pattern)
+    if files:
+        latest_file = max(files, key=os.path.getctime)
+    with open(latest_file, "r") as f:
+        lines = f.readlines()
+        for line in lines:
+            if "Unexpected license problem" in line:
+                raise InaccessibleAnsysLicense()

src/ansys/fluent/core/launcher/launcher.py

@@ -264,6 +264,9 @@ def launch_fluent(
     if env is None:
         env = {}
 
+    if cwd:
+        pyfluent.CWD = cwd
+
     def _mode_to_launcher_type(fluent_launch_mode: LaunchMode):
         launcher_mode_type = {
             LaunchMode.CONTAINER: DockerLauncher,

src/ansys/fluent/core/services/interceptors.py

@@ -8,6 +8,7 @@
 from google.protobuf.message import Message
 import grpc
 
+from ansys.fluent.core.launcher.error_handler import _check_license
 from ansys.fluent.core.services.batch_ops import BatchOps
 
 network_logger: logging.Logger = logging.getLogger("pyfluent.networking")
@@ -109,7 +110,8 @@ def _intercept_call(
             grpc_ex = response.exception()
             ex = RuntimeError(grpc_ex.details())
             ex.__context__ = grpc_ex
-            raise ex from None
+            if not _check_license():
+                raise ex from None
         return response
 
     def intercept_unary_unary(