Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

feat: Remove alive-progress dependency #3424

Merged
merged 12 commits into from
Oct 25, 2024
1 change: 0 additions & 1 deletion pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -44,7 +44,6 @@ ansys-api-fluent = "^0.3.28"
ansys-platform-instancemanagement = "~=1.0"
ansys-tools-filetransfer = ">=0.1,<0.3"
ansys-units = "^0.3.2"
alive-progress = ">=3.1.5"
beartype = ">=0.17"
docker = ">=7.1.0"
grpcio = "^1.30.0"
Expand Down
78 changes: 46 additions & 32 deletions src/ansys/fluent/core/utils/file_transfer_service.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@
import pathlib
import random
import shutil
import sys
from typing import Any, Callable, List, Protocol # noqa: F401
import warnings

Expand Down Expand Up @@ -369,6 +370,25 @@ def download(self, file_name: list[str] | str, local_directory: str | None = Non
)


def _progress_bar(file_name: str, upload: bool):
progress_bar_size = 40
for file in range(2):
progress = file / 1
hpohekar marked this conversation as resolved.
Show resolved Hide resolved
filled_length = int(progress_bar_size * progress)
progrss_bar = "█" * filled_length + "-" * (progress_bar_size - filled_length)
if upload:
sys.stdout.write(
f"\r|{progrss_bar}| {progress:.1%} {os.path.basename(file_name)} uploaded."
)
else:
sys.stdout.write(
f"\r|{progrss_bar}| {progress:.1%} {os.path.basename(file_name)} downloaded."
)
sys.stdout.flush()

print()


class PimFileTransferService:
"""Provides a file transfer service based on `PyPIM <https://pypim.docs.pyansys.com/version/stable/>`_ and the ``simple_upload_server()`` method.

Expand Down Expand Up @@ -482,23 +502,20 @@ def upload(self, file_name: list[str] | str, remote_file_name: str | None = None
"""
files = [file_name] if isinstance(file_name, str) else file_name
if self.is_configured():
from alive_progress import alive_bar

with alive_bar(len(files), title="Uploading...") as bar:
for file in files:
if os.path.isfile(file):
if not self.file_service.file_exist(os.path.basename(file)):
self.upload_file(
file_name=file, remote_file_name=remote_file_name
)
bar()
else:
warnings.warn(
f"\n{file} with the same name exists at the remote location.\n",
PyFluentUserWarning,
)
elif not self.file_service.file_exist(os.path.basename(file)):
raise FileNotFoundError(f"{file} does not exist.")
for file in files:
if os.path.isfile(file):
if not self.file_service.file_exist(os.path.basename(file)):
self.upload_file(
file_name=file, remote_file_name=remote_file_name
)
_progress_bar(file_name=file, upload=True)
else:
warnings.warn(
f"\n{file} with the same name exists at the remote location.\n",
PyFluentUserWarning,
)
elif not self.file_service.file_exist(os.path.basename(file)):
raise FileNotFoundError(f"{file} does not exist.")

def download_file(self, file_name: str, local_directory: str | None = None):
"""Download a file from the server supported by `PyPIM<https://pypim.docs.pyansys.com/version/stable/>`.
Expand Down Expand Up @@ -537,21 +554,18 @@ def download(self, file_name: list[str] | str, local_directory: str | None = "."
"""
files = [file_name] if isinstance(file_name, str) else file_name
if self.is_configured():
from alive_progress import alive_bar

with alive_bar(len(files), title="Downloading...") as bar:
for file in files:
if os.path.isfile(file):
warnings.warn(
f"\nFile already exists. File path:\n{file}\n",
PyFluentUserWarning,
)
else:
self.download_file(
file_name=os.path.basename(file),
local_directory=local_directory,
)
bar()
for file in files:
if os.path.isfile(file):
warnings.warn(
f"\nFile already exists. File path:\n{file}\n",
PyFluentUserWarning,
)
else:
self.download_file(
file_name=os.path.basename(file),
local_directory=local_directory,
)
_progress_bar(file_name=file, upload=False)

def __call__(self, pim_instance: Any | None = None):
self.pim_instance = pim_instance