Feat/mwf

codegen/datamodelgen.py

@@ -5,7 +5,7 @@
 from typing import Any, Dict
 
 from ansys.api.fluent.v0 import datamodel_se_pb2 as DataModelProtoModule
-from ansys.fluent.core.session import BaseSession as Session
+from ansys.fluent.core.session import _BaseSession as Session
 
 _THIS_DIR = Path(__file__).parent
 

src/ansys/fluent/core/__init__.py

@@ -13,7 +13,7 @@
     set_ansys_version,
     set_fluent_path,
 )
-from ansys.fluent.core.session import BaseSession as Fluent  # noqa: F401
+from ansys.fluent.core.session import _BaseSession as Fluent  # noqa: F401
 from ansys.fluent.core.utils.logging import LOG
 
 _VERSION_INFO = None

src/ansys/fluent/core/fluent_connection.py

@@ -0,0 +1,273 @@
+from ctypes import c_int, sizeof
+import itertools
+import os
+import threading
+from typing import Callable, List, Optional, Tuple
+import weakref
+
+import grpc
+
+from ansys.fluent.core.services.datamodel_se import (
+    DatamodelService as DatamodelService_SE,
+)
+from ansys.fluent.core.services.datamodel_tui import (
+    DatamodelService as DatamodelService_TUI,
+)
+from ansys.fluent.core.services.events import EventsService
+from ansys.fluent.core.services.field_data import FieldData, FieldDataService, FieldInfo
+from ansys.fluent.core.services.health_check import HealthCheckService
+from ansys.fluent.core.services.monitor import MonitorsService
+from ansys.fluent.core.services.scheme_eval import SchemeEval, SchemeEvalService
+from ansys.fluent.core.services.settings import SettingsService
+from ansys.fluent.core.services.transcript import TranscriptService
+from ansys.fluent.core.solver.events_manager import EventsManager
+from ansys.fluent.core.solver.monitors_manager import MonitorsManager
+
+
+def _get_max_c_int_limit() -> int:
+    """Get the maximum limit of a C int.
+
+    Returns
+    -------
+    int
+        The maximum limit of a C int
+    """
+    return 2 ** (sizeof(c_int) * 8 - 1) - 1
+
+
+class MonitorThread(threading.Thread):
+    """A class used for monitoring a Fluent session.
+
+    Daemon thread which will ensure cleanup of session objects, shutdown of
+    non-deamon threads etc.
+
+    Attributes
+    ----------
+    cbs : List[Callable]
+        Cleanup/shutdown functions
+    """
+
+    def __init__(self):
+        super().__init__(daemon=True)
+        self.cbs: List[Callable] = []
+
+    def run(self) -> None:
+        main_thread = threading.main_thread()
+        main_thread.join()
+        for cb in self.cbs:
+            cb()
+
+
+class _FluentConnection:
+    """Encapsulates a Fluent connection.
+
+    Methods
+    -------
+    start_transcript()
+        Start streaming of Fluent transcript
+
+    stop_transcript()
+        Stop streaming of Fluent transcript
+
+    check_health()
+        Check health of Fluent connection
+
+    exit()
+        Close the Fluent connection and exit Fluent.
+    """
+
+    _on_exit_cbs: List[Callable] = []
+    _id_iter = itertools.count()
+    _monitor_thread: Optional[MonitorThread] = None
+
+    def __init__(
+        self,
+        ip: str = None,
+        port: int = None,
+        password: str = None,
+        channel: grpc.Channel = None,
+        cleanup_on_exit: bool = True,
+        start_transcript: bool = True,
+        remote_instance=None,
+    ):
+        """Instantiate a Session.
+
+        Parameters
+        ----------
+        ip : str, optional
+            IP address to connect to existing Fluent instance. Used only
+            when ``channel`` is ``None``.  Defaults to ``"127.0.0.1"``
+            and can also be set by the environment variable
+            ``PYFLUENT_FLUENT_IP=<ip>``.
+        port : int, optional
+            Port to connect to existing Fluent instance. Used only
+            when ``channel`` is ``None``.  Defaults value can be set by
+            the environment variable ``PYFLUENT_FLUENT_PORT=<port>``.
+        password : str, optional
+            Password to connect to existing Fluent instance.
+        channel : grpc.Channel, optional
+            Grpc channel to use to connect to existing Fluent instance.
+            ip and port arguments will be ignored when channel is
+            specified.
+        cleanup_on_exit : bool, optional
+            When True, the connected Fluent session will be shut down
+            when PyFluent is exited or exit() is called on the session
+            instance, by default True.
+        start_transcript : bool, optional
+            The Fluent transcript is started in the client only when
+            start_transcript is True. It can be started and stopped
+            subsequently via method calls on the Session object.
+        remote_instance : ansys.platform.instancemanagement.Instance
+            The corresponding remote instance when Fluent is launched through
+            PyPIM. This instance will be deleted when calling
+            ``Session.exit()``.
+        """
+        self._channel_str = None
+        if channel is not None:
+            self._channel = channel
+        else:
+            if not ip:
+                ip = os.getenv("PYFLUENT_FLUENT_IP", "127.0.0.1")
+            if not port:
+                port = os.getenv("PYFLUENT_FLUENT_PORT")
+            self._channel_str = f"{ip}:{port}"
+            if not port:
+                raise ValueError(
+                    "The port to connect to Fluent session is not provided."
+                )
+            # Same maximum message length is used in the server
+            max_message_length = _get_max_c_int_limit()
+            self._channel = grpc.insecure_channel(
+                f"{ip}:{port}",
+                options=[
+                    ("grpc.max_send_message_length", max_message_length),
+                    ("grpc.max_receive_message_length", max_message_length),
+                ],
+            )
+        self._metadata: List[Tuple[str, str]] = (
+            [("password", password)] if password else []
+        )
+        self._id = f"session-{next(_FluentConnection._id_iter)}"
+
+        if not _FluentConnection._monitor_thread:
+            _FluentConnection._monitor_thread = MonitorThread()
+            _FluentConnection._monitor_thread.start()
+
+        self._transcript_service = TranscriptService(self._channel, self._metadata)
+        self._transcript_thread: Optional[threading.Thread] = None
+
+        self._events_service = EventsService(self._channel, self._metadata)
+        self.events_manager = EventsManager(self._id, self._events_service)
+
+        self._monitors_service = MonitorsService(self._channel, self._metadata)
+        self.monitors_manager = MonitorsManager(self._id, self._monitors_service)
+
+        self.events_manager.register_callback(
+            "InitializedEvent", self.monitors_manager.refresh
+        )
+        self.events_manager.register_callback(
+            "DataReadEvent", self.monitors_manager.refresh
+        )
+        self.events_manager.start()
+        self.datamodel_service_tui = DatamodelService_TUI(self._channel, self._metadata)
+        self.datamodel_service_se = DatamodelService_SE(self._channel, self._metadata)
+        self.settings_service = SettingsService(self._channel, self._metadata)
+
+        self._field_data_service = FieldDataService(self._channel, self._metadata)
+        self.field_info = FieldInfo(self._field_data_service)
+        self.field_data = FieldData(self._field_data_service, self.field_info)
+
+        self._health_check_service = HealthCheckService(self._channel, self._metadata)
+
+        self._scheme_eval_service = SchemeEvalService(self._channel, self._metadata)
+        self.scheme_eval = SchemeEval(self._scheme_eval_service)
+
+        self._cleanup_on_exit = cleanup_on_exit
+        self._start_transcript = start_transcript
+
+        if start_transcript:
+            self.start_transcript()
+
+        self._remote_instance = remote_instance
+
+        self._finalizer = weakref.finalize(
+            self,
+            _FluentConnection._exit,
+            self._channel,
+            self._cleanup_on_exit,
+            self.scheme_eval,
+            self._transcript_service,
+            self.events_manager,
+            self._remote_instance,
+        )
+        _FluentConnection._monitor_thread.cbs.append(self._finalizer)
+
+    @property
+    def id(self) -> str:
+        """Return the session id."""
+        return self._id
+
+    @staticmethod
+    def _print_transcript(transcript: str):
+        print(transcript)
+
+    @staticmethod
+    def _process_transcript(transcript_service):
+        responses = transcript_service.begin_streaming()
+        transcript = ""
+        while True:
+            try:
+                response = next(responses)
+                transcript += response.transcript
+                if transcript[-1] == "\n":
+                    _FluentConnection._print_transcript(transcript[0:-1])
+                    transcript = ""
+            except StopIteration:
+                break
+
+    def start_transcript(self) -> None:
+        """Start streaming of Fluent transcript."""
+        self._transcript_thread = threading.Thread(
+            target=_FluentConnection._process_transcript,
+            args=(self._transcript_service,),
+        )
+
+        self._transcript_thread.start()
+
+    def stop_transcript(self) -> None:
+        """Stop streaming of Fluent transcript."""
+        self._transcript_service.end_streaming()
+
+    def check_health(self) -> str:
+        """Check health of Fluent connection."""
+        if self._channel:
+            try:
+                return self._health_check_service.check_health()
+            except Exception:
+                return HealthCheckService.Status.NOT_SERVING.name
+        else:
+            return HealthCheckService.Status.NOT_SERVING.name
+
+    def exit(self) -> None:
+        """Close the Fluent connection and exit Fluent."""
+        self._finalizer()
+
+    @staticmethod
+    def _exit(
+        channel,
+        cleanup_on_exit,
+        scheme_eval,
+        transcript_service,
+        events_manager,
+        remote_instance,
+    ) -> None:
+        if channel:
+            if cleanup_on_exit:
+                scheme_eval.exec(("(exit-server)",))
+            transcript_service.end_streaming()
+            events_manager.stop()
+            channel.close()
+            channel = None
+
+        if remote_instance:
+            remote_instance.delete()

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

@@ -14,8 +14,9 @@
 import time
 from typing import Any, Dict, Union
 
+from ansys.fluent.core.fluent_connection import _FluentConnection
 from ansys.fluent.core.launcher.fluent_container import start_fluent_container
-from ansys.fluent.core.session import BaseSession, Session
+from ansys.fluent.core.session import Session, _BaseSession
 from ansys.fluent.core.session_meshing import Meshing
 from ansys.fluent.core.session_pure_meshing import PureMeshing
 from ansys.fluent.core.session_solver import Solver
@@ -239,8 +240,10 @@ def launch_remote_fluent(
     instance.wait_for_ready()
     # nb pymapdl sets max msg len here:
     channel = instance.build_grpc_channel()
-    return BaseSession(
-        channel=channel, cleanup_on_exit=cleanup_on_exit, remote_instance=instance
+    return _BaseSession(
+        fluent_connection=_FluentConnection(
+            channel=channel, cleanup_on_exit=cleanup_on_exit, remote_instance=instance
+        )
     )
 
 
@@ -262,7 +265,7 @@ def launch_fluent(
     case_filepath: str = None,
     meshing_mode: bool = None,
     mode: Union[LaunchModes, str, None] = None,
-) -> Union[BaseSession, Session]:
+) -> Union[_BaseSession, Session]:
     """Launch Fluent locally in server mode or connect to a running Fluent
     server instance.
 
@@ -428,16 +431,20 @@ def launch_fluent(
                 pyfluent.EXAMPLES_PATH, pyfluent.EXAMPLES_PATH, args
             )
             return new_session(
-                port=port,
-                cleanup_on_exit=cleanup_on_exit,
-                start_transcript=start_transcript,
+                _FluentConnection(
+                    port=port,
+                    cleanup_on_exit=cleanup_on_exit,
+                    start_transcript=start_transcript,
+                )
             )
         else:
             ip = argvals.get("ip", None)
             port = argvals.get("port", None)
             return new_session(
-                ip=ip,
-                port=port,
-                cleanup_on_exit=cleanup_on_exit,
-                start_transcript=start_transcript,
+                fluent_connection=_FluentConnection(
+                    ip=ip,
+                    port=port,
+                    cleanup_on_exit=cleanup_on_exit,
+                    start_transcript=start_transcript,
+                )
             )