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

refactor: small refactoring of command classes #3345

Merged
merged 2 commits into from
Oct 2, 2024
Merged
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
86 changes: 36 additions & 50 deletions src/ansys/fluent/core/solver/flobject.py
Original file line number Diff line number Diff line change
Expand Up @@ -1596,7 +1596,7 @@ class Map(SettingsBase[DictStateType]):
"""A ``Map`` object representing key-value settings."""


def _get_new_keywords(obj, args, kwds):
def _get_new_keywords(obj, *args, **kwds):
newkwds = {}
argNames = []
argumentNames = []
Expand Down Expand Up @@ -1674,6 +1674,33 @@ def __getattr__(self, name: str):
class BaseCommand(Action):
"""Executes command."""

def _execute_command(self, *args, **kwds):
mkundu1 marked this conversation as resolved.
Show resolved Hide resolved
"""Execute a command with the specified positional and keyword arguments."""
newkwds = _get_new_keywords(self, *args, **kwds)
if self.flproxy.is_interactive_mode():
prompt = self.flproxy.get_command_confirmation_prompt(
self._parent.path, self.obj_name, **newkwds
)
if prompt:
valid_responses = {"y": True, "yes": True, "n": False, "no": False}
while True:
response = input(prompt + ": y[es]/n[o] ").strip().lower()
if response in valid_responses:
if not valid_responses[response]:
return
break
else:
print("Please enter 'y[es]' or 'n[o]'.")
with self._while_executing_command():
ret = self.flproxy.execute_cmd(self._parent.path, self.obj_name, **newkwds)
if os.getenv("PYFLUENT_NO_FIX_PARAMETER_LIST_RETURN") != "1":
if (self._parent.path, self.obj_name) in [
("parameters/input-parameters", "list"),
("parameters/output-parameters", "list"),
]:
ret = _fix_parameter_list_return(ret)
return ret

def execute_command(self, *args, **kwds):
"""Execute command."""
for arg, value in kwds.items():
Expand Down Expand Up @@ -1701,7 +1728,11 @@ def execute_command(self, *args, **kwds):
return ret

def __call__(self, *args, **kwds):
return self.execute_command(*args, **kwds)
try:
return self.execute_command(*args, **kwds)
except KeyboardInterrupt:
self._root._on_interrupt(self)
raise KeyboardInterrupt


# TODO: Remove this after paremater list() method is fixed from Fluent side
Expand All @@ -1728,32 +1759,6 @@ def _fix_parameter_list_return(val):
class Command(BaseCommand):
"""Command object."""

def _execute_command(self, **kwds):
"""Execute a command with the specified keyword arguments."""
newkwds = _get_new_keywords(self, [], kwds)
if self.flproxy.is_interactive_mode():
prompt = self.flproxy.get_command_confirmation_prompt(
self._parent.path, self.obj_name, **newkwds
)
if prompt:
while True:
response = input(prompt + ": y[es]/n[o] ")
if response in ["y", "Y", "n", "N", "yes", "no"]:
break
else:
print("Enter y[es]/n[o]")
if response in ["n", "N", "no"]:
return
with self._while_executing_command():
ret = self.flproxy.execute_cmd(self._parent.path, self.obj_name, **newkwds)
if os.getenv("PYFLUENT_NO_FIX_PARAMETER_LIST_RETURN") != "1":
if (self._parent.path, self.obj_name) in [
("parameters/input-parameters", "list"),
("parameters/output-parameters", "list"),
]:
ret = _fix_parameter_list_return(ret)
return ret

def __call__(self, **kwds):
"""Call a command with the specified keyword arguments."""
try:
Expand All @@ -1764,29 +1769,10 @@ def __call__(self, **kwds):


class CommandWithPositionalArgs(BaseCommand):
"""Command Object."""

def _execute_command(self, *args, **kwds):
"""Execute a command with the specified keyword arguments."""
newkwds = _get_new_keywords(self, args, kwds)
if self.flproxy.is_interactive_mode():
prompt = self.flproxy.get_command_confirmation_prompt(
self._parent.path, self.obj_name, **newkwds
)
if prompt:
while True:
response = input(prompt + ": y[es]/n[o] ")
if response in ["y", "Y", "n", "N", "yes", "no"]:
break
else:
print("Enter y[es]/n[o]")
if response in ["n", "N", "no"]:
return
with self._while_executing_command():
return self.flproxy.execute_cmd(self._parent.path, self.obj_name, **newkwds)
"""Command Object supporting positional arguments."""

def __call__(self, *args, **kwds):
"""Call a command with the specified keyword arguments."""
"""Call a command with the specified positional and keyword arguments."""
try:
return self.execute_command(*args, **kwds)
except KeyboardInterrupt:
Expand All @@ -1799,7 +1785,7 @@ class Query(Action):

def __call__(self, **kwds):
"""Call a query with the specified keyword arguments."""
newkwds = _get_new_keywords(self, [], kwds)
newkwds = _get_new_keywords(self, **kwds)
return self.flproxy.execute_query(self._parent.path, self.obj_name, **newkwds)


Expand Down