feat: App utilities gRPC services

.pre-commit-config.yaml

@@ -13,7 +13,7 @@ repos:
   hooks:
   - id: black
     args: [
-      --target-version=py39
+      --target-version=py310
     ]
 
 - repo: https://github.com/pycqa/isort

codegen/allapigen.py

@@ -3,7 +3,8 @@
 from time import time
 
 from ansys.fluent.core import CODEGEN_OUTDIR, FluentMode, FluentVersion, launch_fluent
-from ansys.fluent.core.codegen import StaticInfoType, allapigen, print_fluent_version
+from ansys.fluent.core.codegen import StaticInfoType, allapigen
+from ansys.fluent.core.codegen.print_fluent_version import print_fluent_version
 from ansys.fluent.core.search import _generate_api_data
 from ansys.fluent.core.utils.fluent_version import get_version_for_file_name
 
@@ -60,7 +61,7 @@
     t1 = time()
     print(f"Time to fetch static info: {t1 - t0:.2f} seconds")
     CODEGEN_OUTDIR.mkdir(parents=True, exist_ok=True)
-    print_fluent_version.generate(version, solver.scheme_eval.scheme_eval)
+    print_fluent_version(solver._app_utilities)
     solver.exit()
 
     allapigen.generate(version, static_infos)

src/ansys/fluent/core/codegen/print_fluent_version.py

@@ -1,26 +1,21 @@
 """Module to write Fluent version information."""
 
-from ansys.fluent.core import CODEGEN_OUTDIR, launch_fluent
-from ansys.fluent.core.utils.fluent_version import get_version_for_file_name
+from ansys.fluent.core import CODEGEN_OUTDIR, FluentVersion, launch_fluent
 
 
-def print_fluent_version(version: str, scheme_eval):
+def print_fluent_version(app_utilities):
     """Write Fluent version information to file."""
+    version = FluentVersion(app_utilities.get_product_version()).number
+    build_info = app_utilities.get_build_info()
     version_file = (CODEGEN_OUTDIR / f"fluent_version_{version}.py").resolve()
     with open(version_file, "w", encoding="utf8") as f:
         f.write(f'FLUENT_VERSION = "{version}"\n')
-        f.write(f'FLUENT_BUILD_TIME = "{scheme_eval("(inquire-build-time)")}"\n')
-        f.write(f'FLUENT_BUILD_ID = "{scheme_eval("(inquire-build-id)")}"\n')
-        f.write(f'FLUENT_REVISION = "{scheme_eval("(inquire-src-vcs-id)")}"\n')
-        f.write(f'FLUENT_BRANCH = "{scheme_eval("(inquire-src-vcs-branch)")}"\n')
-
-
-def generate(version: str, scheme_eval):
-    """Write Fluent version information."""
-    print_fluent_version(version, scheme_eval)
+        f.write(f'FLUENT_BUILD_TIME = "{build_info["build_time"]}"\n')
+        f.write(f'FLUENT_BUILD_ID = "{build_info["build_id"]}"\n')
+        f.write(f'FLUENT_REVISION = "{build_info["vcs_revision"]}"\n')
+        f.write(f'FLUENT_BRANCH = "{build_info["vcs_branch"]}"\n')
 
 
 if __name__ == "__main__":
     solver = launch_fluent()
-    version = get_version_for_file_name(session=solver)
-    generate(version, solver.scheme_eval.scheme_eval)
+    print_fluent_version(solver._app_utilities)

src/ansys/fluent/core/fluent_connection.py

@@ -21,6 +21,10 @@
 
 import ansys.fluent.core as pyfluent
 from ansys.fluent.core.services import service_creator
+from ansys.fluent.core.services.app_utilities import (
+    AppUtilitiesOld,
+    AppUtilitiesService,
+)
 from ansys.fluent.core.services.scheme_eval import SchemeEvalService
 from ansys.fluent.core.utils.execution import timeout_exec, timeout_loop
 from ansys.fluent.core.utils.file_transfer_service import RemoteFileTransferStrategy
@@ -246,15 +250,24 @@ def __init__(self, create_grpc_service, error_state):
         self.scheme_eval = service_creator("scheme_eval").create(
             self._scheme_eval_service
         )
+        if (
+            pyfluent.FluentVersion(self.scheme_eval.version)
+            < pyfluent.FluentVersion.v252
+        ):
+            self._app_utilities = AppUtilitiesOld(self.scheme_eval)
+        else:
+            self._app_utilities_service = create_grpc_service(
+                AppUtilitiesService, error_state
+            )
+            self._app_utilities = service_creator("app_utilities").create(
+                self._app_utilities_service
+            )
 
     @property
     def product_build_info(self) -> str:
         """Get Fluent build information."""
-        build_time = self.scheme_eval.scheme_eval("(inquire-build-time)")
-        build_id = self.scheme_eval.scheme_eval("(inquire-build-id)")
-        rev = self.scheme_eval.scheme_eval("(inquire-src-vcs-id)")
-        branch = self.scheme_eval.scheme_eval("(inquire-src-vcs-branch)")
-        return f"Build Time: {build_time}  Build Id: {build_id}  Revision: {rev}  Branch: {branch}"
+        build_info = self._app_utilities.get_build_info()
+        return f'Build Time: {build_info["build_time"]}  Build Id: {build_info["build_id"]}  Revision: {build_info["vcs_revision"]}  Branch: {build_info["vcs_branch"]}'
 
     def get_cortex_connection_properties(self):
         """Get connection properties of Fluent."""
@@ -263,10 +276,12 @@ def get_cortex_connection_properties(self):
         try:
             logger.info(self.product_build_info)
             logger.debug("Obtaining Cortex connection properties...")
-            fluent_host_pid = self.scheme_eval.scheme_eval("(cx-client-id)")
-            cortex_host = self.scheme_eval.scheme_eval("(cx-cortex-host)")
-            cortex_pid = self.scheme_eval.scheme_eval("(cx-cortex-id)")
-            cortex_pwd = self.scheme_eval.scheme_eval("(cortex-pwd)")
+            cortex_info = self._app_utilities.get_controller_process_info()
+            solver_info = self._app_utilities.get_solver_process_info()
+            fluent_host_pid = solver_info["process_id"]
+            cortex_host = cortex_info["hostname"]
+            cortex_pid = cortex_info["process_id"]
+            cortex_pwd = cortex_info["working_directory"]
             logger.debug("Cortex connection properties successfully obtained.")
         except _InactiveRpcError:
             logger.warning(
@@ -282,22 +297,11 @@ def get_cortex_connection_properties(self):
 
     def get_mode(self):
         """Get the mode of a running fluent session."""
-        from ansys.fluent.core import FluentMode
-
-        if self.scheme_eval.scheme_eval("(cx-solver-mode?)"):
-            mode_str = self.scheme_eval.scheme_eval('(getenv "PRJAPP_APP")')
-            if mode_str == "flaero_server":
-                return FluentMode.SOLVER_AERO
-            elif mode_str == "flicing":
-                return FluentMode.SOLVER_ICING
-            else:
-                return FluentMode.SOLVER
-        else:
-            return FluentMode.MESHING
+        return self._app_utilities.get_app_mode()
 
     def exit_server(self):
         """Exits the server."""
-        self.scheme_eval.exec(("(exit-server)",))
+        self._app_utilities.exit()
 
 
 def _pid_exists(pid):

src/ansys/fluent/core/journaling.py

@@ -4,14 +4,14 @@
 class Journal:
     """Control the writing of Fluent Python journals."""
 
-    def __init__(self, scheme_eval):
+    def __init__(self, app_utilities):
         """__init__ method of Journal class."""
-        self.scheme_eval = scheme_eval
+        self._app_utilities = app_utilities
 
     def start(self, file_name: str):
         """Start writing a Fluent Python journal at the specified file_name."""
-        self.scheme_eval.exec([f'(api-start-python-journal "{file_name}")'])
+        self._app_utilities.start_python_journal(journal_name=file_name)
 
     def stop(self):
         """Stop writing the Fluent Python journal."""
-        self.scheme_eval.exec(["(api-stop-python-journal)"])
+        self._app_utilities.stop_python_journal()

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

@@ -1,5 +1,6 @@
 """Provides a module to create gRPC services."""
 
+from ansys.fluent.core.services.app_utilities import AppUtilities
 from ansys.fluent.core.services.batch_ops import BatchOpsService
 from ansys.fluent.core.services.datamodel_se import (
     DatamodelService as DatamodelService_SE,
@@ -22,6 +23,7 @@
 from ansys.fluent.core.services.transcript import TranscriptService
 
 _service_cls_by_name = {
+    "app_utilities": AppUtilities,
     "health_check": HealthCheckService,
     "datamodel": DatamodelService_SE,
     "tui": DatamodelService_TUI,

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

@@ -3,7 +3,7 @@
 import os
 from typing import TypeVar
 
-from ansys.fluent.core.services.scheme_eval import SchemeEval
+from ansys.fluent.core.services.app_utilities import AppUtilities
 from ansys.fluent.core.utils.fluent_version import FluentVersion
 
 _TApiUpgradeAdvisor = TypeVar("_TApiUpgradeAdvisor", bound="ApiUpgradeAdvisor")
@@ -12,11 +12,17 @@
 class ApiUpgradeAdvisor:
     """API upgrade advisor."""
 
-    def __init__(self, scheme_eval: SchemeEval, version: str, mode: str) -> None:
+    def __init__(
+        self,
+        app_utilities: AppUtilities,
+        version: str,
+        mode: str,
+    ) -> None:
         """Initialize ApiUpgradeAdvisor."""
-        self._scheme_eval = scheme_eval.scheme_eval
+        self._app_utilities = app_utilities
         self._version = version
         self._mode = mode
+        self._id = None
 
     def _can_advise(self) -> bool:
         return (
@@ -27,16 +33,12 @@ def _can_advise(self) -> bool:
 
     def __enter__(self) -> _TApiUpgradeAdvisor:
         if self._can_advise():
-            self._scheme_eval("(define pyfluent-journal-str-port (open-output-string))")
-            self._scheme_eval("(api-echo-python-port pyfluent-journal-str-port)")
+            self._id = self._app_utilities.start_python_journal()
         return self
 
     def __exit__(self, exc_type, exc_value, exc_tb) -> None:
         if self._can_advise():
-            self._scheme_eval("(api-unecho-python-port pyfluent-journal-str-port)")
-            journal_str = self._scheme_eval(
-                "(close-output-port pyfluent-journal-str-port)"
-            ).strip()
+            journal_str = (self._app_utilities.stop_python_journal(self._id)).strip()
             if (
                 journal_str.startswith("solver.")
                 and not journal_str.startswith("solver.tui")