Session upload and download methods on session objects.

src/ansys/fluent/core/session.py

@@ -1,6 +1,7 @@
 """Module containing class encapsulating Fluent connection and the Base
 Session."""
 import json
+import os
 from typing import Any
 import warnings
 
@@ -65,6 +66,7 @@ class _BaseSession:
     def __init__(self, fluent_connection: _FluentConnection):
         self.fluent_connection = fluent_connection
         self.scheme_eval = self.fluent_connection.scheme_eval
+        self._uploader = None
 
     @classmethod
     def create_from_server_info_file(
@@ -170,6 +172,18 @@ def __dir__(self):
             )
         )
 
+    def _upload(self, file_path: str, remote_file_name: str = None):
+        """Uploads a file on the server."""
+        if not self._uploader:
+            self._uploader = _Uploader(self.fluent_connection._remote_instance)
+        return self._uploader.upload(file_path, remote_file_name)
+
+    def _download(self, file_name: str, local_file_path: str = None):
+        """Downloads a file from the server."""
+        if not self._uploader:
+            self._uploader = _Uploader(self.fluent_connection._remote_instance)
+        return self._uploader.download(file_name, local_file_path)
+
 
 class Session:
     """Instantiates a Fluent connection. This is a deprecated class. This has
@@ -239,6 +253,8 @@ def __init__(
             self._datamodel_service_tui, self._settings_service
         )
 
+        self._uploader = None
+
     @classmethod
     def create_from_server_info_file(
         cls,
@@ -319,6 +335,18 @@ def __dir__(self):
             )
         )
 
+    def _upload(self, file_path: str, remote_file_name: str = None):
+        """Uploads a file on the server."""
+        if not self._uploader:
+            self._uploader = _Uploader(self.fluent_connection._remote_instance)
+        return self._uploader.upload(file_path, remote_file_name)
+
+    def _download(self, file_name: str, local_file_path: str = None):
+        """Downloads a file from the server."""
+        if not self._uploader:
+            self._uploader = _Uploader(self.fluent_connection._remote_instance)
+        return self._uploader.download(file_name, local_file_path)
+
     class Solver:
         def __init__(
             self, tui_service: DatamodelService_TUI, settings_service: SettingsService
@@ -348,3 +376,61 @@ def root(self):
             if self._settings_root is None:
                 self._settings_root = settings_get_root(flproxy=self._settings_service)
             return self._settings_root
+
+
+class _Uploader:
+    """Instantiates a file uploader and downloader to have a seamless file
+    reading / writing in the cloud particularly in Ansys lab . Here we are
+    exposing upload and download methods on session objects. These would be no-
+    ops if PyPIM is not configured or not authorized with the appropriate
+    service. This will be used for internal purpose only.
+
+    Attributes
+    ----------
+    pim_instance: PIM instance
+        Instance of PIM which supports upload server services.
+
+    file_service: Client instance
+        Instance of Client which supports upload and download methods.
+
+    Methods
+    -------
+    upload(
+        file_path, remote_file_name
+        )
+        Upload a file to the server.
+
+    download(
+        file_name, local_file_path
+        )
+        Download a file from the server.
+    """
+
+    def __init__(self, pim_instance):
+        self.pim_instance = pim_instance
+
+        try:
+            upload_server = self.pim_instance.services["http-simple-upload-server"]
+        except AttributeError:
+            LOG.error("PIM is not installed or not authorized.")
+        except KeyError:
+            self.file_service = None
+        else:
+            from simple_upload_server.client import Client
+
+            self.file_service = Client(
+                token="token", url=upload_server.uri, headers=upload_server.headers
+            )
+
+    def upload(self, file_path: str, remote_file_name: str = None):
+        """Uploads a file on the server."""
+        expanded_file_path = os.path.expandvars(file_path)
+        upload_file_name = remote_file_name or os.path.basename(expanded_file_path)
+        self.file_service.upload_file(expanded_file_path, upload_file_name)
+
+    def download(self, file_name: str, local_file_path: str = None):
+        """Downloads a file from the server."""
+        if self.file_service.file_exist(file_name):
+            self.file_service.download_file(file_name, local_file_path)
+        else:
+            raise FileNotFoundError("Remote file does not exist.")

tests/test_uploader.py

@@ -0,0 +1,36 @@
+import pytest
+from util.meshing_workflow import new_mesh_session  # noqa: F401
+
+import ansys.fluent.core as pyfluent
+
+
+def test_base_session_upload(new_mesh_session):
+    base_session = new_mesh_session
+    with pytest.raises(AttributeError) as e_info:
+        base_session._upload(
+            pyfluent.EXAMPLES_PATH + "/mixing_elbow.py", "test_upload_download.py"
+        )
+    base_session.exit()
+
+
+def test_base_session_download(new_mesh_session):
+    base_session = new_mesh_session
+    with pytest.raises(AttributeError) as e_info:
+        base_session._download("test_upload_download.py", pyfluent.EXAMPLES_PATH)
+    base_session.exit()
+
+
+def test_session_upload(new_mesh_session):
+    session = new_mesh_session
+    with pytest.raises(AttributeError) as e_info:
+        session._upload(
+            pyfluent.EXAMPLES_PATH + "/mixing_elbow.py", "test_upload_download.py"
+        )
+    session.exit()
+
+
+def test_session_download(new_mesh_session):
+    session = new_mesh_session
+    with pytest.raises(AttributeError) as e_info:
+        session._download("test_upload_download.py", pyfluent.EXAMPLES_PATH)
+    session.exit()