[Feature] Styling adjustments (#6408)

* styling adjustments

* auto completion for main menu commands; cached results style

* move platform settings above

* don't check if obbject.results - we want to have the error message if smt goes wrong with the to_df()

* register only if there are results

* minor style change

---------

Co-authored-by: montezdesousa <79287829+montezdesousa@users.noreply.github.com>
This commit is contained in:
Henrique Joaquim
2024-05-14 14:35:17 +01:00
committed by GitHub
parent 07693784fb
commit c2f5f7a3e6
5 changed files with 170 additions and 140 deletions

View File

@@ -20,8 +20,10 @@ class Registry:
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

View File

@@ -99,17 +99,18 @@ class MenuText:
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"

View File

@@ -479,123 +479,137 @@ class BaseController(metaclass=ABCMeta):
"\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
# 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"
)
routine_file = os.path.join(
f"{current_user.preferences.export_directory}/routines",
title_for_local_storage,
)
# 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"
)
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 "-h" not in other_args and "--help" not in other_args:
global RECORD_SESSION # noqa: PLW0603
global SESSION_RECORDED # noqa: PLW0603
if not RECORD_SESSION:
session.console.print(
f"[green]Your routine has been recorded and saved here: {routine_file}[/green]\n"
"[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 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
# 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"
)
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:
routine_file = os.path.join(
f"{current_user.preferences.export_directory}/routines",
title_for_local_storage,
)
# If file already exists, add a timestamp to the name
if os.path.isfile(routine_file):
i = session.console.input(
"A routine with the same name already exists, "
"do you want to replace it? (y/n): "
"A local routine with the same name already exists, "
"do you want to override 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]")
while i.lower() not in ["y", "yes", "n", "no"]:
i = session.console.input("Select 'y' or 'n' to proceed: ")
session.console.print("")
# Clear session to be recorded again
RECORD_SESSION = False
SESSION_RECORDED = list()
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"
)
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)
session.console.print(
f"[green]Your routine has been recorded and saved here: {routine_file}[/green]\n"
)
# 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."""

View File

@@ -159,8 +159,8 @@ class PlatformController(BaseController):
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]"
@@ -181,7 +181,9 @@ class PlatformController(BaseController):
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)
@@ -326,11 +328,14 @@ class PlatformController(BaseController):
)
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)

View File

@@ -188,8 +188,9 @@ class CLIController(BaseController):
"--input": None,
"-i": "--input",
"--url": None,
"--help": None,
"-h": "--help",
}
choices["record"] = {
"--name": None,
"-n": "--name",
@@ -200,14 +201,36 @@ class CLIController(BaseController):
"--tag1": {c: None for c in constants.SCRIPT_TAGS},
"--tag2": {c: None for c in constants.SCRIPT_TAGS},
"--tag3": {c: None for c in constants.SCRIPT_TAGS},
"--help": None,
"-h": "--help",
}
choices["stop"] = {"--help": None, "-h": "--help"}
choices["results"] = {"--help": None, "-h": "--help"}
self.update_completer(choices)
def print_help(self):
"""Print help."""
mt = MenuText("")
mt.add_info("Configure your own CLI")
mt.add_info("Configure the platform and manage your account")
for router, value in PLATFORM_ROUTERS.items():
if router not in NON_DATA_ROUTERS or router in ["reference", "coverage"]:
continue
if value == "menu":
menu_description = (
obb.reference["routers"] # type: ignore
.get(f"{self.PATH}{router}", {})
.get("description")
) or ""
mt.add_menu(
name=router,
description=menu_description.split(".")[0].lower(),
)
else:
mt.add_cmd(router)
mt.add_info("\nConfigure your CLI")
mt.add_menu(
"settings",
description="enable and disable feature flags, preferences and settings",
@@ -260,32 +283,17 @@ class CLIController(BaseController):
else:
mt.add_cmd(router)
mt.add_info("\nConfigure the platform and manage your account")
for router, value in PLATFORM_ROUTERS.items():
if router not in NON_DATA_ROUTERS or router in ["reference", "coverage"]:
continue
if value == "menu":
menu_description = (
obb.reference["routers"] # type: ignore
.get(f"{self.PATH}{router}", {})
.get("description")
) or ""
mt.add_menu(
name=router,
description=menu_description.split(".")[0].lower(),
)
else:
mt.add_cmd(router)
mt.add_info("\nAccess and manage your cached results")
mt.add_raw("\n")
mt.add_cmd("results")
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())[ # type: ignore
: session.settings.N_TO_DISPLAY_OBBJECT_REGISTRY
]:
mt.add_raw(f"\tOBB{key}: {value['command']}\n") # type: ignore
mt.add_raw(
f"[yellow]OBB{key}[/yellow]: {value['command']}",
left_spacing=True,
)
session.console.print(text=mt.menu_text, menu="Home")
self.update_runtime_choices()