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

[Feature] Styling adjustments #6408

Merged
merged 12 commits into from
May 14, 2024
Merged
6 changes: 4 additions & 2 deletions cli/openbb_cli/argparse_translator/obbject_registry.py
Original file line number Diff line number Diff line change
Expand Up @@ -20,8 +20,10 @@ def _contains_obbject(uuid: str, obbjects: List[OBBject]) -> bool:

def register(self, obbject: OBBject) -> bool:
"""Designed to add an OBBject instance to the registry."""
if isinstance(obbject, OBBject) and not self._contains_obbject(
obbject.id, self._obbjects
if (
isinstance(obbject, OBBject)
and not self._contains_obbject(obbject.id, self._obbjects)
and obbject.results
):
self._obbjects.append(obbject)
return True
Expand Down
9 changes: 5 additions & 4 deletions cli/openbb_cli/config/menu_text.py
Original file line number Diff line number Diff line change
Expand Up @@ -99,17 +99,18 @@ def _format_cmd_description(
else description
)

def add_raw(self, text: str):
def add_raw(self, text: str, left_spacing: bool = False):
"""Append raw text (without translation)."""
self.menu_text += text
if left_spacing:
self.menu_text += f"{self.SECTION_SPACING * ' '}{text}\n"
else:
self.menu_text += text

def add_section(
self, text: str, description: str = "", leading_new_line: bool = False
):
"""Append raw text (without translation)."""
spacing = (self.CMD_NAME_LENGTH - len(text) + self.SECTION_SPACING) * " "
left_spacing = self.SECTION_SPACING * " "
text = f"{left_spacing}{text}"
if description:
text = f"{text}{spacing}{description}\n"

Expand Down
214 changes: 114 additions & 100 deletions cli/openbb_cli/controllers/base_controller.py
Original file line number Diff line number Diff line change
Expand Up @@ -479,123 +479,137 @@ def call_record(self, other_args) -> None:
"\n[yellow]Remember to run 'stop' command when you are done!\n[/yellow]"
)

def call_stop(self, _) -> None:
def call_stop(self, other_args) -> None:
"""Process stop command."""
global RECORD_SESSION # noqa: PLW0603
global SESSION_RECORDED # noqa: PLW0603
parser = argparse.ArgumentParser(
add_help=False,
formatter_class=argparse.ArgumentDefaultsHelpFormatter,
prog="stop",
description="Stop recording session into .openbb routine file",
)
# This is only for auto-completion purposes
_ = self.parse_simple_args(parser, other_args)

if not RECORD_SESSION:
session.console.print(
"[red]There is no session being recorded. Start one using the command 'record'[/red]\n"
)
elif len(SESSION_RECORDED) < 5:
session.console.print(
"[red]Run at least 4 commands before stopping recording a session.[/red]\n"
)
else:
current_user = session.user
if "-h" not in other_args and "--help" not in other_args:
global RECORD_SESSION # noqa: PLW0603
global SESSION_RECORDED # noqa: PLW0603

# Check if the user just wants to store routine locally
# This works regardless of whether they are logged in or not
if RECORD_SESSION_LOCAL_ONLY:
# Whitespaces are replaced by underscores and an .openbb extension is added
title_for_local_storage = (
SESSION_RECORDED_NAME.replace(" ", "_") + ".openbb"
if not RECORD_SESSION:
session.console.print(
"[red]There is no session being recorded. Start one using the command 'record'[/red]\n"
)

routine_file = os.path.join(
f"{current_user.preferences.export_directory}/routines",
title_for_local_storage,
elif len(SESSION_RECORDED) < 5:
session.console.print(
"[red]Run at least 4 commands before stopping recording a session.[/red]\n"
)
else:
current_user = session.user

# Check if the user just wants to store routine locally
# This works regardless of whether they are logged in or not
if RECORD_SESSION_LOCAL_ONLY:
# Whitespaces are replaced by underscores and an .openbb extension is added
title_for_local_storage = (
SESSION_RECORDED_NAME.replace(" ", "_") + ".openbb"
)

# If file already exists, add a timestamp to the name
if os.path.isfile(routine_file):
i = session.console.input(
"A local routine with the same name already exists, "
"do you want to override it? (y/n): "
routine_file = os.path.join(
f"{current_user.preferences.export_directory}/routines",
title_for_local_storage,
)
session.console.print("")
while i.lower() not in ["y", "yes", "n", "no"]:
i = session.console.input("Select 'y' or 'n' to proceed: ")
session.console.print("")

if i.lower() in ["n", "no"]:
new_name = (
datetime.now().strftime("%Y%m%d_%H%M%S_")
+ title_for_local_storage
)
routine_file = os.path.join(
current_user.preferences.export_directory,
"routines",
new_name,
)
session.console.print(
f"[yellow]The routine name has been updated to '{new_name}'[/yellow]\n"
# If file already exists, add a timestamp to the name
if os.path.isfile(routine_file):
i = session.console.input(
"A local routine with the same name already exists, "
"do you want to override it? (y/n): "
)
session.console.print("")
while i.lower() not in ["y", "yes", "n", "no"]:
i = session.console.input("Select 'y' or 'n' to proceed: ")
session.console.print("")

if i.lower() in ["n", "no"]:
new_name = (
datetime.now().strftime("%Y%m%d_%H%M%S_")
+ title_for_local_storage
)
routine_file = os.path.join(
current_user.preferences.export_directory,
"routines",
new_name,
)
session.console.print(
f"[yellow]The routine name has been updated to '{new_name}'[/yellow]\n"
)

# Writing to file
Path(os.path.dirname(routine_file)).mkdir(parents=True, exist_ok=True)

with open(routine_file, "w") as file1:
lines = ["# OpenBB Platform CLI - Routine", "\n"]

username = getattr(
session.user.profile.hub_session, "username", "local"
# Writing to file
Path(os.path.dirname(routine_file)).mkdir(
parents=True, exist_ok=True
)

lines += [f"# Author: {username}", "\n\n"] if username else ["\n"]
lines += [
f"# Title: {SESSION_RECORDED_NAME}",
"\n",
f"# Tags: {SESSION_RECORDED_TAGS}",
"\n\n",
f"# Description: {SESSION_RECORDED_DESCRIPTION}",
"\n\n",
]
lines += [c + "\n" for c in SESSION_RECORDED[:-1]]
# Writing data to a file
file1.writelines(lines)
with open(routine_file, "w") as file1:
lines = ["# OpenBB Platform CLI - Routine", "\n"]

session.console.print(
f"[green]Your routine has been recorded and saved here: {routine_file}[/green]\n"
)
username = getattr(
session.user.profile.hub_session, "username", "local"
)

# If user doesn't specify they want to store routine locally
# Confirm that the user is logged in
elif not session.is_local():
routine = "\n".join(SESSION_RECORDED[:-1])
hub_session = current_user.profile.hub_session
lines += (
[f"# Author: {username}", "\n\n"] if username else ["\n"]
)
lines += [
f"# Title: {SESSION_RECORDED_NAME}",
"\n",
f"# Tags: {SESSION_RECORDED_TAGS}",
"\n\n",
f"# Description: {SESSION_RECORDED_DESCRIPTION}",
"\n\n",
]
lines += [c + "\n" for c in SESSION_RECORDED[:-1]]
# Writing data to a file
file1.writelines(lines)

if routine is not None:
auth_header = (
f"{hub_session.token_type} {hub_session.access_token.get_secret_value()}"
if hub_session
else None
session.console.print(
f"[green]Your routine has been recorded and saved here: {routine_file}[/green]\n"
)
kwargs = {
"auth_header": auth_header,
"name": SESSION_RECORDED_NAME,
"description": SESSION_RECORDED_DESCRIPTION,
"routine": routine,
"tags": SESSION_RECORDED_TAGS,
"public": SESSION_RECORDED_PUBLIC,
}
response = upload_routine(**kwargs) # type: ignore
if response is not None and response.status_code == 409:
i = session.console.input(
"A routine with the same name already exists, "
"do you want to replace it? (y/n): "
)
session.console.print("")
if i.lower() in ["y", "yes"]:
kwargs["override"] = True # type: ignore
response = upload_routine(**kwargs) # type: ignore
else:
session.console.print("[info]Aborted.[/info]")

# Clear session to be recorded again
RECORD_SESSION = False
SESSION_RECORDED = list()
# If user doesn't specify they want to store routine locally
# Confirm that the user is logged in
elif not session.is_local():
routine = "\n".join(SESSION_RECORDED[:-1])
hub_session = current_user.profile.hub_session

if routine is not None:
auth_header = (
f"{hub_session.token_type} {hub_session.access_token.get_secret_value()}"
if hub_session
else None
)
kwargs = {
"auth_header": auth_header,
"name": SESSION_RECORDED_NAME,
"description": SESSION_RECORDED_DESCRIPTION,
"routine": routine,
"tags": SESSION_RECORDED_TAGS,
"public": SESSION_RECORDED_PUBLIC,
}
response = upload_routine(**kwargs) # type: ignore
if response is not None and response.status_code == 409:
i = session.console.input(
"A routine with the same name already exists, "
"do you want to replace it? (y/n): "
)
session.console.print("")
if i.lower() in ["y", "yes"]:
kwargs["override"] = True # type: ignore
response = upload_routine(**kwargs) # type: ignore
else:
session.console.print("[info]Aborted.[/info]")

# Clear session to be recorded again
RECORD_SESSION = False
SESSION_RECORDED = list()

def call_whoami(self, other_args: List[str]) -> None:
"""Process whoami command."""
Expand Down
15 changes: 10 additions & 5 deletions cli/openbb_cli/controllers/base_platform_controller.py
Original file line number Diff line number Diff line change
Expand Up @@ -159,8 +159,8 @@ def method(self, other_args: List[str], translator=translator):

if obbject:

if isinstance(obbject, OBBject) and obbject.results:
if session.max_obbjects_exceeded():
if isinstance(obbject, OBBject):
if session.max_obbjects_exceeded() and obbject.results:
session.obbject_registry.remove()
session.console.print(
"[yellow]Maximum number of OBBjects reached. The oldest entry was removed.[yellow]"
Expand All @@ -181,7 +181,9 @@ def method(self, other_args: List[str], translator=translator):
session.settings.SHOW_MSG_OBBJECT_REGISTRY
and register_result
):
session.console.print("Added OBBject to registry.")
session.console.print(
"Added `OBBject` to cached results."
)

# making the dataframe available
# either for printing or exporting (or both)
Expand Down Expand Up @@ -326,11 +328,14 @@ def print_help(self):
)

if session.obbject_registry.obbjects:
mt.add_section("Cached Results:\n", leading_new_line=True)
mt.add_info("\nCached Results")
for key, value in list(session.obbject_registry.all.items())[
: session.settings.N_TO_DISPLAY_OBBJECT_REGISTRY
]:
mt.add_raw(f"\tOBB{key}: {value['command']}\n")
mt.add_raw(
f"[yellow]OBB{key}[/yellow]: {value['command']}",
left_spacing=True,
)

session.console.print(text=mt.menu_text, menu=self.PATH)

Expand Down
Loading
Loading