fix: Update LocalFileTransferStrategy (#3353)

* fix: Update LocalFileTransferStrategy

* fix: Update PIMLauncher

* refactoring

src/ansys/fluent/core/utils/file_transfer_service.py

@@ -114,7 +114,7 @@ def upload(
 
         Parameters
         ----------
-        file_name : str
+        file_name : list[str] | str
             File name.
         remote_file_name : str, optional
             Remote file name. The default is ``None``.
@@ -134,17 +134,18 @@ def upload(
         >>> meshing_session.upload(file_name=mesh_file_name, remote_file_name="elbow.msh.h5")
         >>> meshing_session.meshing.File.ReadMesh(FileName="elbow.msh.h5")
         """
-        local_file_name = pathlib.Path(file_name)
-        if local_file_name.exists() and local_file_name.is_file():
-            if remote_file_name:
-                shutil.copyfile(
-                    file_name,
-                    str(self.fluent_cwd / f"{os.path.basename(remote_file_name)}"),
-                )
-            else:
-                shutil.copyfile(
-                    file_name, str(self.fluent_cwd / f"{os.path.basename(file_name)}")
-                )
+        files = _get_files(file_name)
+        for file in files:
+            if file.is_file():
+                if remote_file_name:
+                    shutil.copyfile(
+                        file,
+                        str(self.fluent_cwd / f"{os.path.basename(remote_file_name)}"),
+                    )
+                else:
+                    shutil.copyfile(
+                        file, str(self.fluent_cwd / f"{os.path.basename(file)}")
+                    )
 
     def download(
         self, file_name: list[str] | str, local_directory: str | None = None
@@ -153,7 +154,7 @@ def download(
 
         Parameters
         ----------
-        file_name : str
+        file_name : list[str] | str
             File name.
         local_directory : str, optional
             Local directory. The default is ``None``.
@@ -168,22 +169,21 @@ def download(
         >>> meshing_session.meshing.File.WriteMesh(FileName="write_elbow.msh.h5")
         >>> meshing_session.download(file_name="write_elbow.msh.h5", local_directory="<local_directory_path>")
         """
-        remote_file_name = str(self.fluent_cwd / f"{os.path.basename(file_name)}")
-        local_file_name = None
-        if local_directory:
-            if pathlib.Path(local_directory).is_dir():
-                local_file_name = pathlib.Path(local_directory) / os.path.basename(
-                    file_name
-                )
-            elif not pathlib.Path(local_directory).is_dir():
-                local_file_name = pathlib.Path(local_directory)
-        else:
-            local_file_name = pathlib.Path(self.pyfluent_cwd) / os.path.basename(
-                file_name
-            )
-        if local_file_name.exists() and local_file_name.samefile(remote_file_name):
-            return
-        shutil.copyfile(remote_file_name, str(local_file_name))
+        files = _get_files(file_name)
+        for file in files:
+            remote_file_name = str(self.fluent_cwd / file.name)
+            local_file_name = None
+            if local_directory:
+                local_dir_path = pathlib.Path(local_directory)
+                if local_dir_path.is_dir():
+                    local_file_name = local_dir_path / file.name
+                else:
+                    local_file_name = local_dir_path
+            else:
+                local_file_name = self.pyfluent_cwd / file.name
+            if local_file_name.exists() and local_file_name.samefile(remote_file_name):
+                return
+            shutil.copyfile(remote_file_name, str(local_file_name))
 
 
 def _get_files(
@@ -289,7 +289,7 @@ def upload(self, file_name: list[str] | str, remote_file_name: str | None = None
 
         Parameters
         ----------
-        file_name : str
+        file_name : list[str] | str
             File name.
         remote_file_name : str, optional
             Remote file name. The default is ``None``.
@@ -335,7 +335,7 @@ def download(self, file_name: list[str] | str, local_directory: str | None = Non
 
         Parameters
         ----------
-        file_name : str
+        file_name : list[str] | str
             File name.
         local_directory : str, optional
             Local directory. The default is ``None``.
@@ -470,7 +470,7 @@ def upload(self, file_name: list[str] | str, remote_file_name: str | None = None
 
         Parameters
         ----------
-        file_name : str
+        file_name : list[str] | str
             File name.
         remote_file_name : str, optional
             Remote file name. The default is ``None``.
@@ -530,7 +530,7 @@ def download(self, file_name: list[str] | str, local_directory: str | None = "."
 
         Parameters
         ----------
-        file_name : str
+        file_name : list[str] | str
             File name.
         local_directory : str, optional
             Local directory. The default is the current working directory.

tests/test_file_transfer_service.py

@@ -127,3 +127,39 @@ def test_file_list_in_datamodel(fault_tolerant_workflow_session):
         RemoveEmptyParts=True,
         Route="Native",
     )
+
+
+@pytest.mark.standalone
+def test_local_file_transfer_in_datamodel(monkeypatch):
+    import ansys.fluent.core as pyfluent
+
+    monkeypatch.setattr(pyfluent, "USE_FILE_TRANSFER_SERVICE", True)
+
+    fmd_file = examples.download_file("exhaust_system.fmd", "pyfluent/exhaust_system")
+
+    assert fmd_file
+
+    meshing = pyfluent.launch_fluent(
+        mode=pyfluent.FluentMode.MESHING,
+        file_transfer_service=LocalFileTransferStrategy(),
+    )
+
+    meshing.workflow.InitializeWorkflow(WorkflowType="Fault-tolerant Meshing")
+
+    meshing.PartManagement.AppendFmdFiles(
+        AssemblyParentNode=0,
+        FilePath=[fmd_file],
+        FileUnit="mm",
+        IgnoreSolidNamesAppend=False,
+        JtLOD="1",
+        Options={
+            "Line": False,
+            "Solid": True,
+            "Surface": True,
+        },
+        PartPerBody=False,
+        PrefixParentName=False,
+        RemoveEmptyParts=True,
+        Route="Native",
+    )
+    meshing.exit()