feat: App utilities gRPC services

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

@@ -4,23 +4,20 @@
 from ansys.fluent.core.utils.fluent_version import get_version_for_file_name
 
 
-def print_fluent_version(version: str, scheme_eval):
+def print_fluent_version(version: str, app_utilities):
     """Write Fluent version information to file."""
     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')
+        f.write(f'"{app_utilities.get_build_info()}"\n')
 
 
-def generate(version: str, scheme_eval):
+def generate(version: str, app_utilities):
     """Write Fluent version information."""
-    print_fluent_version(version, scheme_eval)
+    print_fluent_version(version, app_utilities)
 
 
 if __name__ == "__main__":
     solver = launch_fluent()
     version = get_version_for_file_name(session=solver)
-    generate(version, solver.scheme_eval.scheme_eval)
+    generate(version, solver._app_utilities)

src/ansys/fluent/core/fluent_connection.py

@@ -21,6 +21,7 @@
 
 import ansys.fluent.core as pyfluent
 from ansys.fluent.core.services import service_creator
+from ansys.fluent.core.services.app_utilities import 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 +247,17 @@ def __init__(self, create_grpc_service, error_state):
         self.scheme_eval = service_creator("scheme_eval").create(
             self._scheme_eval_service
         )
+        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}"
+        self._app_utilities.get_build_info()
 
     def get_cortex_connection_properties(self):
         """Get connection properties of Fluent."""
@@ -263,10 +266,13 @@ 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()
+            fluent_host_pid = self._app_utilities.get_solver_process_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 +288,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/app_utilities.py

@@ -0,0 +1,247 @@
+"""Wrappers over AppUtilities gRPC service of Fluent."""
+
+from typing import Any, List, Tuple
+
+import grpc
+
+from ansys.api.fluent.v0 import app_utilities_pb2 as AppUtilitiesProtoModule
+from ansys.api.fluent.v0 import app_utilities_pb2_grpc as AppUtilitiesGrpcModule
+import ansys.fluent.core as pyfluent
+from ansys.fluent.core.services.interceptors import (
+    BatchInterceptor,
+    ErrorStateInterceptor,
+    GrpcErrorInterceptor,
+    TracingInterceptor,
+)
+from ansys.fluent.core.streaming_services.events_streaming import SolverEvent
+
+
+class AppUtilitiesService:
+    """AppUtilities Service."""
+
+    def __init__(
+        self, channel: grpc.Channel, metadata: List[Tuple[str, str]], fluent_error_state
+    ):
+        """__init__ method of AppUtilities class."""
+        intercept_channel = grpc.intercept_channel(
+            channel,
+            GrpcErrorInterceptor(),
+            ErrorStateInterceptor(fluent_error_state),
+            TracingInterceptor(),
+            BatchInterceptor(),
+        )
+        self._stub = AppUtilitiesGrpcModule.AppUtilitiesStub(intercept_channel)
+        self._metadata = metadata
+
+    def get_product_version(
+        self, request: AppUtilitiesProtoModule.GetProductVersionRequest
+    ) -> AppUtilitiesProtoModule.GetProductVersionResponse:
+        """Get product version RPC of AppUtilities service."""
+        return self._stub.GetProductVersion(request, metadata=self._metadata)
+
+    def get_build_info(
+        self, request: AppUtilitiesProtoModule.GetBuildInfoRequest
+    ) -> AppUtilitiesProtoModule.GetBuildInfoResponse:
+        """Get build info RPC of AppUtilities service."""
+        return self._stub.GetBuildInfo(request, metadata=self._metadata)
+
+    def get_controller_process_info(
+        self, request: AppUtilitiesProtoModule.GetControllerProcessInfoRequest
+    ) -> AppUtilitiesProtoModule.GetControllerProcessInfoResponse:
+        """Get controller process info RPC of AppUtilities service."""
+        return self._stub.GetControllerProcessInfo(request, metadata=self._metadata)
+
+    def get_solver_process_info(
+        self, request: AppUtilitiesProtoModule.GetSolverProcessInfoRequest
+    ) -> AppUtilitiesProtoModule.GetSolverProcessInfoResponse:
+        """Get solver process info RPC of AppUtilities service."""
+        return self._stub.GetSolverProcessInfo(request, metadata=self._metadata)
+
+    def get_app_mode(
+        self, request: AppUtilitiesProtoModule.GetAppModeRequest
+    ) -> AppUtilitiesProtoModule.GetAppModeResponse:
+        """Get app mode RPC of AppUtilities service."""
+        return self._stub.GetAppMode(request, metadata=self._metadata)
+
+    def start_python_journal(
+        self, request: AppUtilitiesProtoModule.StartPythonJournalRequest
+    ) -> AppUtilitiesProtoModule.StartPythonJournalResponse:
+        """Start python journal RPC of AppUtilities service."""
+        return self._stub.StartPythonJournal(request, metadata=self._metadata)
+
+    def stop_python_journal(
+        self, request: AppUtilitiesProtoModule.StopPythonJournalRequest
+    ) -> AppUtilitiesProtoModule.StopPythonJournalResponse:
+        """Stop python journal RPC of AppUtilities service."""
+        return self._stub.StopPythonJournal(request, metadata=self._metadata)
+
+    def is_beta_enabled(
+        self, request: AppUtilitiesProtoModule.IsBetaEnabledRequest
+    ) -> AppUtilitiesProtoModule.IsBetaEnabledResponse:
+        """Is beta enabled RPC of AppUtilities service."""
+        return self._stub.IsBetaEnabled(request, metadata=self._metadata)
+
+    def is_wildcard(
+        self, request: AppUtilitiesProtoModule.IsWildcardRequest
+    ) -> AppUtilitiesProtoModule.IsWildcardResponse:
+        """Is wildcard RPC of AppUtilities service."""
+        return self._stub.IsWildcard(request, metadata=self._metadata)
+
+    def is_solution_data_available(
+        self, request: AppUtilitiesProtoModule.IsSolutionDataAvailableRequest
+    ) -> AppUtilitiesProtoModule.IsSolutionDataAvailableResponse:
+        """Is solution data available RPC of AppUtilities service."""
+        return self._stub.IsSolutionDataAvailable(request, metadata=self._metadata)
+
+    def register_pause_on_solution_events(
+        self, request: AppUtilitiesProtoModule.RegisterPauseOnSolutionEventsRequest
+    ) -> AppUtilitiesProtoModule.RegisterPauseOnSolutionEventsResponse:
+        """Register on pause solution events RPC of AppUtilities service."""
+        return self._stub.RegisterPauseOnSolutionEvents(
+            request, metadata=self._metadata
+        )
+
+    def resume_on_solution_event(
+        self, request: AppUtilitiesProtoModule.ResumeOnSolutionEventRequest
+    ) -> AppUtilitiesProtoModule.ResumeOnSolutionEventResponse:
+        """Resume on solution event RPC of AppUtilities service."""
+        return self._stub.ResumeOnSolutionEvent(request, metadata=self._metadata)
+
+    def unregister_pause_on_solution_events(
+        self, request: AppUtilitiesProtoModule.UnregisterPauseOnSolutionEventsRequest
+    ) -> AppUtilitiesProtoModule.UnregisterPauseOnSolutionEventsResponse:
+        """Unregister on pause solution events RPC of AppUtilities service."""
+        return self._stub.UnregisterPauseOnSolutionEvents(
+            request, metadata=self._metadata
+        )
+
+    def exit(
+        self, request: AppUtilitiesProtoModule.ExitRequest
+    ) -> AppUtilitiesProtoModule.ExitResponse:
+        """Exit RPC of AppUtilities service."""
+        return self._stub.Exit(request, metadata=self._metadata)
+
+
+class AppUtilities:
+    """AppUtilities."""
+
+    def __init__(self, service: AppUtilitiesService):
+        """__init__ method of AppUtilities class."""
+        self.service = service
+
+    def get_product_version(self) -> Any:
+        """Get product version."""
+        request = AppUtilitiesProtoModule.GetProductVersionRequest()
+        response = self.service.get_product_version(request)
+        return f"{response.major}.{response.minor}.{response.patch}"
+
+    def get_build_info(self) -> Any:
+        """Get build info."""
+        request = AppUtilitiesProtoModule.GetBuildInfoRequest()
+        response = self.service.get_build_info(request)
+        return f"Build Time: {response.build_time} Build ID: {response.build_id} Revision: {response.vcs_revision} Branch: {response.vcs_branch}"
+
+    def get_controller_process_info(self) -> Any:
+        """Get controller process info."""
+        request = AppUtilitiesProtoModule.GetControllerProcessInfoRequest()
+        response = self.service.get_controller_process_info(request)
+        return {
+            "hostname": response.hostname,
+            "process_id": response.process_id,
+            "working_directory": response.working_directory,
+        }
+
+    def get_solver_process_info(self) -> Any:
+        """Get solver process info."""
+        request = AppUtilitiesProtoModule.GetSolverProcessInfoRequest()
+        response = self.service.get_solver_process_info(request)
+        return {
+            "hostname": response.hostname,
+            "process_id": response.process_id,
+            "working_directory": response.working_directory,
+        }
+
+    def get_app_mode(self) -> Any:
+        """Get app mode.
+
+        Raises
+        ------
+        ValueError
+            If app mode is unknown.
+        """
+        request = AppUtilitiesProtoModule.GetAppModeRequest()
+        response = self.service.get_app_mode(request)
+        if response.app_mode == AppUtilitiesProtoModule.APP_MODE_UNKNOWN:
+            raise ValueError("Unknown app mode.")
+        elif response.app_mode == AppUtilitiesProtoModule.APP_MODE_MESHING:
+            return pyfluent.FluentMode.MESHING
+        elif response.app_mode == AppUtilitiesProtoModule.APP_MODE_SOLVER:
+            return pyfluent.FluentMode.SOLVER
+        elif response.app_mode == AppUtilitiesProtoModule.APP_MODE_SOLVER_ICING:
+            return pyfluent.FluentMode.SOLVER_ICING
+        elif response.app_mode == AppUtilitiesProtoModule.APP_MODE_SOLVER_AERO:
+            return pyfluent.FluentMode.SOLVER_AERO
+
+    def start_python_journal(self, journal_name: str | None = None) -> Any:
+        """Start python journal."""
+        request = AppUtilitiesProtoModule.StartPythonJournalRequest()
+        request.journal_name = journal_name
+        response = self.service.start_python_journal(request)
+        return response
+
+    def stop_python_journal(self) -> Any:
+        """Stop python journal."""
+        request = AppUtilitiesProtoModule.StopPythonJournalRequest()
+        response = self.service.stop_python_journal(request)
+        return response
+
+    def is_beta_enabled(self) -> Any:
+        """Is beta enabled."""
+        request = AppUtilitiesProtoModule.IsBetaEnabledRequest()
+        response = self.service.is_beta_enabled(request)
+        return response.is_beta_enabled
+
+    def is_wildcard(self, input: str | None = None) -> Any:
+        """Is wildcard."""
+        request = AppUtilitiesProtoModule.IsWildcardRequest()
+        request.input = input
+        response = self.service.is_wildcard(request)
+        return response.is_wildcard
+
+    def is_solution_data_available(self) -> Any:
+        """Is solution data available."""
+        request = AppUtilitiesProtoModule.IsSolutionDataAvailableRequest()
+        response = self.service.is_solution_data_available(request)
+        return response.is_solution_data_available
+
+    def register_pause_on_solution_events(self, solution_event: SolverEvent) -> Any:
+        """Register pause on solution events."""
+        request = AppUtilitiesProtoModule.RegisterPauseOnSolutionEventsRequest()
+        if solution_event == SolverEvent.TIMESTEP_ENDED:
+            request.solution_event = 2
+        elif solution_event == SolverEvent.ITERATION_ENDED:
+            request.solution_event = 1
+        else:
+            request.solution_event = 0
+        response = self.service.register_pause_on_solution_events(request)
+        return response.registration_id
+
+    def resume_on_solution_event(self, registration_id: str) -> Any:
+        """Resume on solution event."""
+        request = AppUtilitiesProtoModule.ResumeOnSolutionEventRequest()
+        request.registration_id = registration_id
+        response = self.service.resume_on_solution_event(request)
+        return response
+
+    def unregister_pause_on_solution_events(self, registration_id: str) -> Any:
+        """Unregister pause on solution events."""
+        request = AppUtilitiesProtoModule.UnregisterPauseOnSolutionEventsRequest()
+        request.registration_id = registration_id
+        response = self.service.unregister_pause_on_solution_events(request)
+        return response
+
+    def exit(self) -> Any:
+        """Exit."""
+        request = AppUtilitiesProtoModule.ExitRequest()
+        response = self.service.exit(request)
+        return response