mirror of
https://github.com/OpenBB-finance/OpenBB.git
synced 2026-10-27 12:40:35 +00:00
* Adding tests * Added tests to po_controller * Refactored po controller * fix custom_resets * fix load * add disable timeout * pylint Co-authored-by: DidierRLopes <dro.lopes@campus.fct.unl.pt> Co-authored-by: montezdesousa <79287829+montezdesousa@users.noreply.github.com> Co-authored-by: montezdesousa <montezdesousa@gmail.com>
1028 lines
36 KiB
Python
1028 lines
36 KiB
Python
#!/usr/bin/env python
|
|
"""Main Terminal Module"""
|
|
__docformat__ = "numpy"
|
|
|
|
import argparse
|
|
import difflib
|
|
import logging
|
|
import os
|
|
import platform
|
|
import sys
|
|
import webbrowser
|
|
from typing import List
|
|
import dotenv
|
|
|
|
from prompt_toolkit import PromptSession
|
|
from prompt_toolkit.completion import NestedCompleter
|
|
from prompt_toolkit.styles import Style
|
|
from prompt_toolkit.formatted_text import HTML
|
|
import pandas as pd
|
|
|
|
from openbb_terminal import feature_flags as obbff
|
|
|
|
from openbb_terminal.core.config.paths import (
|
|
REPOSITORY_DIRECTORY,
|
|
USER_ENV_FILE,
|
|
REPOSITORY_ENV_FILE,
|
|
HOME_DIRECTORY,
|
|
ROUTINES_DIRECTORY,
|
|
)
|
|
from openbb_terminal.core.log.generation.path_tracking_file_handler import (
|
|
PathTrackingFileHandler,
|
|
)
|
|
|
|
from openbb_terminal.helper_funcs import (
|
|
check_positive,
|
|
get_flair,
|
|
parse_simple_args,
|
|
EXPORT_ONLY_RAW_DATA_ALLOWED,
|
|
)
|
|
from openbb_terminal.loggers import setup_logging
|
|
from openbb_terminal.menu import session
|
|
from openbb_terminal.parent_classes import BaseController
|
|
from openbb_terminal.rich_config import console, MenuText, translate
|
|
from openbb_terminal.terminal_helper import (
|
|
bootup,
|
|
check_for_updates,
|
|
is_reset,
|
|
print_goodbye,
|
|
reset,
|
|
suppress_stdout,
|
|
update_terminal,
|
|
welcome_message,
|
|
)
|
|
from openbb_terminal.helper_funcs import parse_and_split_input
|
|
from openbb_terminal.common import feedparser_view
|
|
|
|
# pylint: disable=too-many-public-methods,import-outside-toplevel,too-many-branches,no-member,C0302
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
env_file = str(USER_ENV_FILE)
|
|
|
|
|
|
class TerminalController(BaseController):
|
|
"""Terminal Controller class"""
|
|
|
|
CHOICES_COMMANDS = [
|
|
"keys",
|
|
"settings",
|
|
"survey",
|
|
"update",
|
|
"featflags",
|
|
"exe",
|
|
"guess",
|
|
"news",
|
|
]
|
|
CHOICES_MENUS = [
|
|
"stocks",
|
|
"economy",
|
|
"crypto",
|
|
"portfolio",
|
|
"forex",
|
|
"etf",
|
|
"reports",
|
|
"dashboards",
|
|
"funds",
|
|
"alternative",
|
|
"econometrics",
|
|
"sources",
|
|
"forecast",
|
|
]
|
|
|
|
PATH = "/"
|
|
|
|
GUESS_TOTAL_TRIES = 0
|
|
GUESS_NUMBER_TRIES_LEFT = 0
|
|
GUESS_SUM_SCORE = 0.0
|
|
GUESS_CORRECTLY = 0
|
|
|
|
def __init__(self, jobs_cmds: List[str] = None):
|
|
"""Constructor"""
|
|
super().__init__(jobs_cmds)
|
|
|
|
self.queue: List[str] = list()
|
|
|
|
if jobs_cmds:
|
|
self.queue = parse_and_split_input(
|
|
an_input=" ".join(jobs_cmds), custom_filters=[]
|
|
)
|
|
|
|
self.update_success = False
|
|
|
|
self.update_runtime_choices()
|
|
|
|
def update_runtime_choices(self):
|
|
"""Update runtime choices"""
|
|
self.ROUTINE_FILES = {
|
|
filepath.name: filepath
|
|
for filepath in (REPOSITORY_DIRECTORY / "routines").rglob("*.openbb")
|
|
}
|
|
self.ROUTINE_FILES.update(
|
|
{
|
|
filepath.name: filepath
|
|
for filepath in ROUTINES_DIRECTORY.rglob("*.openbb")
|
|
}
|
|
)
|
|
self.ROUTINE_CHOICES = {filename: None for filename in self.ROUTINE_FILES}
|
|
if session and obbff.USE_PROMPT_TOOLKIT:
|
|
choices: dict = {c: {} for c in self.controller_choices}
|
|
choices["support"] = self.SUPPORT_CHOICES
|
|
choices["exe"] = self.ROUTINE_CHOICES
|
|
|
|
self.completer = NestedCompleter.from_nested_dict(choices)
|
|
|
|
def print_help(self):
|
|
"""Print help"""
|
|
mt = MenuText("")
|
|
mt.add_info("_home_")
|
|
mt.add_cmd("about")
|
|
mt.add_cmd("support")
|
|
mt.add_cmd("survey")
|
|
mt.add_cmd("update")
|
|
mt.add_cmd("wiki")
|
|
mt.add_cmd("record")
|
|
mt.add_cmd("stop")
|
|
mt.add_raw("\n")
|
|
mt.add_info("_configure_")
|
|
mt.add_menu("keys")
|
|
mt.add_menu("featflags")
|
|
mt.add_menu("sources")
|
|
mt.add_menu("settings")
|
|
mt.add_raw("\n")
|
|
mt.add_cmd("news")
|
|
mt.add_cmd("exe")
|
|
mt.add_raw("\n")
|
|
mt.add_info("_main_menu_")
|
|
mt.add_menu("stocks")
|
|
mt.add_menu("crypto")
|
|
mt.add_menu("etf")
|
|
mt.add_menu("economy")
|
|
mt.add_menu("forex")
|
|
mt.add_menu("funds")
|
|
mt.add_menu("alternative")
|
|
mt.add_raw("\n")
|
|
mt.add_info("_others_")
|
|
mt.add_menu("econometrics")
|
|
mt.add_menu("forecast")
|
|
mt.add_menu("portfolio")
|
|
mt.add_menu("dashboards")
|
|
mt.add_menu("reports")
|
|
mt.add_raw("\n")
|
|
console.print(text=mt.menu_text, menu="Home")
|
|
self.update_runtime_choices()
|
|
|
|
def call_news(self, other_args: List[str]) -> None:
|
|
"""Process news command"""
|
|
parse = argparse.ArgumentParser(
|
|
add_help=False,
|
|
prog="news",
|
|
description=translate("news"),
|
|
)
|
|
parse.add_argument(
|
|
"-t",
|
|
"--term",
|
|
dest="term",
|
|
default=[""],
|
|
nargs="+",
|
|
help="search for a term on the news",
|
|
)
|
|
parse.add_argument(
|
|
"-s",
|
|
"--sources",
|
|
dest="sources",
|
|
default=["bloomberg.com"],
|
|
nargs="+",
|
|
help="sources from where to get news from",
|
|
)
|
|
if other_args and "-" not in other_args[0][0]:
|
|
other_args.insert(0, "-t")
|
|
news_parser = self.parse_known_args_and_warn(
|
|
parse, other_args, EXPORT_ONLY_RAW_DATA_ALLOWED, limit=5
|
|
)
|
|
if news_parser:
|
|
feedparser_view.display_news(
|
|
term=" ".join(news_parser.term),
|
|
sources=" ".join(news_parser.sources),
|
|
limit=news_parser.limit,
|
|
export=news_parser.export,
|
|
)
|
|
|
|
def call_guess(self, other_args: List[str]) -> None:
|
|
"""Process guess command"""
|
|
import time
|
|
import json
|
|
import random
|
|
|
|
if self.GUESS_NUMBER_TRIES_LEFT == 0 and self.GUESS_SUM_SCORE < 0.01:
|
|
parser_exe = argparse.ArgumentParser(
|
|
add_help=False,
|
|
formatter_class=argparse.ArgumentDefaultsHelpFormatter,
|
|
prog="guess",
|
|
description="Guess command to achieve task successfully.",
|
|
)
|
|
parser_exe.add_argument(
|
|
"-l",
|
|
"--limit",
|
|
type=check_positive,
|
|
help="Number of tasks to attempt.",
|
|
dest="limit",
|
|
default=1,
|
|
)
|
|
if other_args and "-" not in other_args[0][0]:
|
|
other_args.insert(0, "-l")
|
|
ns_parser_guess = parse_simple_args(parser_exe, other_args)
|
|
|
|
if self.GUESS_TOTAL_TRIES == 0:
|
|
self.GUESS_NUMBER_TRIES_LEFT = ns_parser_guess.limit
|
|
self.GUESS_SUM_SCORE = 0
|
|
self.GUESS_TOTAL_TRIES = ns_parser_guess.limit
|
|
|
|
try:
|
|
with open(obbff.GUESS_EASTER_EGG_FILE) as f:
|
|
# Load the file as a JSON document
|
|
json_doc = json.load(f)
|
|
|
|
task = random.choice(list(json_doc.keys())) # nosec
|
|
solution = json_doc[task]
|
|
|
|
start = time.time()
|
|
console.print(f"\n[yellow]{task}[/yellow]\n")
|
|
if isinstance(session, PromptSession):
|
|
an_input = session.prompt("GUESS / $ ")
|
|
else:
|
|
an_input = ""
|
|
time_dif = time.time() - start
|
|
|
|
# When there are multiple paths to same solution
|
|
if isinstance(solution, List):
|
|
if an_input.lower() in [s.lower() for s in solution]:
|
|
self.queue = an_input.split("/") + ["home"]
|
|
console.print(
|
|
f"\n[green]You guessed correctly in {round(time_dif, 2)} seconds![green]\n"
|
|
)
|
|
# If we are already counting successes
|
|
if self.GUESS_TOTAL_TRIES > 0:
|
|
self.GUESS_CORRECTLY += 1
|
|
self.GUESS_SUM_SCORE += time_dif
|
|
else:
|
|
solutions_texts = "\n".join(solution)
|
|
console.print(
|
|
f"\n[red]You guessed wrong! The correct paths would have been:\n{solutions_texts}[/red]\n"
|
|
)
|
|
|
|
# When there is a single path to the solution
|
|
else:
|
|
if an_input.lower() == solution.lower():
|
|
self.queue = an_input.split("/") + ["home"]
|
|
console.print(
|
|
f"\n[green]You guessed correctly in {round(time_dif, 2)} seconds![green]\n"
|
|
)
|
|
# If we are already counting successes
|
|
if self.GUESS_TOTAL_TRIES > 0:
|
|
self.GUESS_CORRECTLY += 1
|
|
self.GUESS_SUM_SCORE += time_dif
|
|
else:
|
|
console.print(
|
|
f"\n[red]You guessed wrong! The correct path would have been:\n{solution}[/red]\n"
|
|
)
|
|
|
|
# Compute average score and provide a result if it's the last try
|
|
if self.GUESS_TOTAL_TRIES > 0:
|
|
|
|
self.GUESS_NUMBER_TRIES_LEFT -= 1
|
|
if self.GUESS_NUMBER_TRIES_LEFT == 0 and self.GUESS_TOTAL_TRIES > 1:
|
|
color = (
|
|
"green"
|
|
if self.GUESS_CORRECTLY == self.GUESS_TOTAL_TRIES
|
|
else "red"
|
|
)
|
|
console.print(
|
|
f"[{color}]OUTCOME: You got {int(self.GUESS_CORRECTLY)} out of"
|
|
f" {int(self.GUESS_TOTAL_TRIES)}.[/{color}]\n"
|
|
)
|
|
if self.GUESS_CORRECTLY == self.GUESS_TOTAL_TRIES:
|
|
avg = self.GUESS_SUM_SCORE / self.GUESS_TOTAL_TRIES
|
|
console.print(
|
|
f"[green]Average score: {round(avg, 2)} seconds![/green]\n"
|
|
)
|
|
self.GUESS_TOTAL_TRIES = 0
|
|
self.GUESS_CORRECTLY = 0
|
|
self.GUESS_SUM_SCORE = 0
|
|
else:
|
|
self.queue += ["guess"]
|
|
|
|
except Exception as e:
|
|
console.print(
|
|
f"[red]Failed to load guess game from file: "
|
|
f"{obbff.GUESS_EASTER_EGG_FILE}[/red]"
|
|
)
|
|
console.print(f"[red]{e}[/red]")
|
|
|
|
@staticmethod
|
|
def call_survey(_) -> None:
|
|
"""Process survey command"""
|
|
webbrowser.open("https://openbb.co/survey")
|
|
|
|
def call_update(self, _):
|
|
"""Process update command"""
|
|
if not obbff.PACKAGED_APPLICATION:
|
|
self.update_success = not update_terminal()
|
|
else:
|
|
console.print(
|
|
"Find the most recent release of the OpenBB Terminal here: "
|
|
"https://openbb.co/products/terminal#get-started\n"
|
|
)
|
|
|
|
def call_keys(self, _):
|
|
"""Process keys command"""
|
|
from openbb_terminal.keys_controller import KeysController
|
|
|
|
self.queue = self.load_class(KeysController, self.queue, env_file)
|
|
|
|
def call_settings(self, _):
|
|
"""Process settings command"""
|
|
from openbb_terminal.settings_controller import SettingsController
|
|
|
|
self.queue = self.load_class(SettingsController, self.queue, env_file)
|
|
|
|
def call_featflags(self, _):
|
|
"""Process feature flags command"""
|
|
from openbb_terminal.featflags_controller import FeatureFlagsController
|
|
|
|
self.queue = self.load_class(FeatureFlagsController, self.queue)
|
|
|
|
def call_stocks(self, _):
|
|
"""Process stocks command"""
|
|
from openbb_terminal.stocks.stocks_controller import StocksController
|
|
|
|
self.queue = self.load_class(StocksController, self.queue)
|
|
|
|
def call_crypto(self, _):
|
|
"""Process crypto command"""
|
|
from openbb_terminal.cryptocurrency.crypto_controller import CryptoController
|
|
|
|
self.queue = self.load_class(CryptoController, self.queue)
|
|
|
|
def call_economy(self, _):
|
|
"""Process economy command"""
|
|
from openbb_terminal.economy.economy_controller import EconomyController
|
|
|
|
self.queue = self.load_class(EconomyController, self.queue)
|
|
|
|
def call_etf(self, _):
|
|
"""Process etf command"""
|
|
from openbb_terminal.etf.etf_controller import ETFController
|
|
|
|
self.queue = self.load_class(ETFController, self.queue)
|
|
|
|
def call_funds(self, _):
|
|
"""Process etf command"""
|
|
from openbb_terminal.mutual_funds.mutual_fund_controller import (
|
|
FundController,
|
|
)
|
|
|
|
self.queue = self.load_class(FundController, self.queue)
|
|
|
|
def call_forex(self, _):
|
|
"""Process forex command"""
|
|
from openbb_terminal.forex.forex_controller import ForexController
|
|
|
|
self.queue = self.load_class(ForexController, self.queue)
|
|
|
|
def call_reports(self, _):
|
|
"""Process reports command"""
|
|
from openbb_terminal.reports.reports_controller import (
|
|
ReportController,
|
|
)
|
|
|
|
self.queue = self.load_class(ReportController, self.queue)
|
|
|
|
def call_dashboards(self, _):
|
|
"""Process dashboards command"""
|
|
if not obbff.PACKAGED_APPLICATION:
|
|
from openbb_terminal.dashboards.dashboards_controller import (
|
|
DashboardsController,
|
|
)
|
|
|
|
self.queue = self.load_class(DashboardsController, self.queue)
|
|
else:
|
|
console.print("This feature is coming soon.")
|
|
console.print(
|
|
"Use the source code and an Anaconda environment if you are familiar with Python."
|
|
)
|
|
|
|
def call_alternative(self, _):
|
|
"""Process alternative command"""
|
|
from openbb_terminal.alternative.alt_controller import (
|
|
AlternativeDataController,
|
|
)
|
|
|
|
self.queue = self.load_class(AlternativeDataController, self.queue)
|
|
|
|
def call_econometrics(self, _):
|
|
"""Process econometrics command"""
|
|
from openbb_terminal.econometrics.econometrics_controller import (
|
|
EconometricsController,
|
|
)
|
|
|
|
self.queue = EconometricsController(self.queue).menu()
|
|
|
|
def call_forecast(self, _):
|
|
"""Process forecast command"""
|
|
from openbb_terminal.forecast.forecast_controller import (
|
|
ForecastController,
|
|
)
|
|
|
|
self.queue = self.load_class(ForecastController, "", pd.DataFrame(), self.queue)
|
|
|
|
def call_portfolio(self, _):
|
|
"""Process portfolio command"""
|
|
from openbb_terminal.portfolio.portfolio_controller import (
|
|
PortfolioController,
|
|
)
|
|
|
|
self.queue = self.load_class(PortfolioController, self.queue)
|
|
|
|
def call_sources(self, _):
|
|
"""Process sources command"""
|
|
from openbb_terminal.sources_controller import SourcesController
|
|
|
|
self.queue = self.load_class(SourcesController, self.queue)
|
|
|
|
def call_exe(self, other_args: List[str]):
|
|
"""Process exe command"""
|
|
# Merge rest of string path to other_args and remove queue since it is a dir
|
|
other_args += self.queue
|
|
|
|
if not other_args:
|
|
console.print(
|
|
"[red]Provide a path to the routine you wish to execute.\n[/red]"
|
|
)
|
|
return
|
|
|
|
full_input = " ".join(other_args)
|
|
if " " in full_input:
|
|
other_args_processed = full_input.split(" ")
|
|
else:
|
|
other_args_processed = [full_input]
|
|
self.queue = []
|
|
|
|
path_routine = ""
|
|
args = list()
|
|
for idx, path_dir in enumerate(other_args_processed):
|
|
if path_dir in ("-i", "--input"):
|
|
args = [path_routine[1:]] + other_args_processed[idx:]
|
|
break
|
|
if path_dir not in ("--file"):
|
|
path_routine += f"/{path_dir}"
|
|
|
|
if not args:
|
|
args = [path_routine[1:]]
|
|
|
|
parser_exe = argparse.ArgumentParser(
|
|
add_help=False,
|
|
formatter_class=argparse.ArgumentDefaultsHelpFormatter,
|
|
prog="exe",
|
|
description="Execute automated routine script.",
|
|
)
|
|
parser_exe.add_argument(
|
|
"--file",
|
|
help="The path or .openbb file to run.",
|
|
dest="path",
|
|
default="",
|
|
required="-h" not in args,
|
|
)
|
|
parser_exe.add_argument(
|
|
"-i",
|
|
"--input",
|
|
help="Select multiple inputs to be replaced in the routine and separated by commas. E.g. GME,AMC,BTC-USD",
|
|
dest="routine_args",
|
|
type=lambda s: [str(item) for item in s.split(",")],
|
|
)
|
|
if args and "-" not in args[0][0]:
|
|
args.insert(0, "--file")
|
|
ns_parser_exe = parse_simple_args(parser_exe, args)
|
|
if ns_parser_exe:
|
|
if ns_parser_exe.path:
|
|
if ns_parser_exe.path in self.ROUTINE_CHOICES:
|
|
path = self.ROUTINE_FILES[ns_parser_exe.path]
|
|
else:
|
|
path = ns_parser_exe.path
|
|
|
|
with open(path) as fp:
|
|
raw_lines = [
|
|
x for x in fp if (not is_reset(x)) and ("#" not in x) and x
|
|
]
|
|
raw_lines = [
|
|
raw_line.strip("\n")
|
|
for raw_line in raw_lines
|
|
if raw_line.strip("\n")
|
|
]
|
|
|
|
lines = list()
|
|
for rawline in raw_lines:
|
|
templine = rawline
|
|
|
|
# Check if dynamic parameter exists in script
|
|
if "$ARGV" in rawline:
|
|
# Check if user has provided inputs through -i or --input
|
|
if ns_parser_exe.routine_args:
|
|
for i, arg in enumerate(ns_parser_exe.routine_args):
|
|
# Check what is the location of the ARGV to be replaced
|
|
if f"$ARGV[{i}]" in templine:
|
|
templine = templine.replace(f"$ARGV[{i}]", arg)
|
|
|
|
# Check if all ARGV have been removed, otherwise means that there are less inputs
|
|
# when running the script than the script expects
|
|
if "$ARGV" in templine:
|
|
console.print(
|
|
"[red]Not enough inputs were provided to fill in dynamic variables. "
|
|
"E.g. --input VAR1,VAR2,VAR3[/red]\n"
|
|
)
|
|
return
|
|
|
|
lines.append(templine)
|
|
# The script expects a parameter that the user has not provided
|
|
else:
|
|
console.print(
|
|
"[red]The script expects parameters, "
|
|
"run the script again with --input defined.[/red]\n"
|
|
)
|
|
return
|
|
else:
|
|
lines.append(templine)
|
|
|
|
simulate_argv = f"/{'/'.join([line.rstrip() for line in lines])}"
|
|
file_cmds = simulate_argv.replace("//", "/home/").split()
|
|
file_cmds = (
|
|
insert_start_slash(file_cmds) if file_cmds else file_cmds
|
|
)
|
|
cmds_with_params = " ".join(file_cmds)
|
|
self.queue = [
|
|
val
|
|
for val in parse_and_split_input(
|
|
an_input=cmds_with_params, custom_filters=[]
|
|
)
|
|
if val
|
|
]
|
|
|
|
if "export" in self.queue[0]:
|
|
export_path = self.queue[0].split(" ")[1]
|
|
# If the path selected does not start from the user root, give relative location from root
|
|
if export_path[0] == "~":
|
|
export_path = export_path.replace(
|
|
"~", HOME_DIRECTORY.as_posix()
|
|
)
|
|
elif export_path[0] != "/":
|
|
export_path = os.path.join(
|
|
os.path.dirname(os.path.abspath(__file__)), export_path
|
|
)
|
|
|
|
# Check if the directory exists
|
|
if os.path.isdir(export_path):
|
|
console.print(
|
|
f"Export data to be saved in the selected folder: '{export_path}'"
|
|
)
|
|
else:
|
|
os.makedirs(export_path)
|
|
console.print(
|
|
f"[green]Folder '{export_path}' successfully created.[/green]"
|
|
)
|
|
obbff.EXPORT_FOLDER_PATH = export_path
|
|
self.queue = self.queue[1:]
|
|
|
|
|
|
# pylint: disable=global-statement
|
|
def terminal(jobs_cmds: List[str] = None, appName: str = "gst"):
|
|
"""Terminal Menu"""
|
|
# TODO: HELP WANTED! Refactor the appName setting if a more elegant solution comes up
|
|
if obbff.PACKAGED_APPLICATION:
|
|
appName = "gst_packaged"
|
|
|
|
setup_logging(appName)
|
|
logger.info("START")
|
|
log_settings()
|
|
|
|
if jobs_cmds is not None and jobs_cmds:
|
|
logger.info("INPUT: %s", "/".join(jobs_cmds))
|
|
|
|
export_path = ""
|
|
if jobs_cmds and "export" in jobs_cmds[0]:
|
|
export_path = jobs_cmds[0].split("/")[0].split(" ")[1]
|
|
jobs_cmds = ["/".join(jobs_cmds[0].split("/")[1:])]
|
|
|
|
ret_code = 1
|
|
t_controller = TerminalController(jobs_cmds)
|
|
an_input = ""
|
|
|
|
if export_path:
|
|
# If the path selected does not start from the user root, give relative location from terminal root
|
|
if export_path[0] == "~":
|
|
export_path = export_path.replace("~", HOME_DIRECTORY.as_posix())
|
|
elif export_path[0] != "/":
|
|
export_path = os.path.join(
|
|
os.path.dirname(os.path.abspath(__file__)), export_path
|
|
)
|
|
|
|
# Check if the directory exists
|
|
if os.path.isdir(export_path):
|
|
console.print(
|
|
f"Export data to be saved in the selected folder: '{export_path}'"
|
|
)
|
|
else:
|
|
os.makedirs(export_path)
|
|
console.print(
|
|
f"[green]Folder '{export_path}' successfully created.[/green]"
|
|
)
|
|
obbff.EXPORT_FOLDER_PATH = export_path
|
|
|
|
bootup()
|
|
if not jobs_cmds:
|
|
welcome_message()
|
|
t_controller.print_help()
|
|
check_for_updates()
|
|
|
|
dotenv.load_dotenv(USER_ENV_FILE)
|
|
dotenv.load_dotenv(REPOSITORY_ENV_FILE, override=True)
|
|
|
|
while ret_code:
|
|
if obbff.ENABLE_QUICK_EXIT:
|
|
console.print("Quick exit enabled")
|
|
break
|
|
|
|
# There is a command in the queue
|
|
if t_controller.queue and len(t_controller.queue) > 0:
|
|
# If the command is quitting the menu we want to return in here
|
|
if t_controller.queue[0] in ("q", "..", "quit"):
|
|
print_goodbye()
|
|
break
|
|
|
|
# Consume 1 element from the queue
|
|
an_input = t_controller.queue[0]
|
|
t_controller.queue = t_controller.queue[1:]
|
|
|
|
# Print the current location because this was an instruction and we want user to know what was the action
|
|
if an_input and an_input.split(" ")[0] in t_controller.CHOICES_COMMANDS:
|
|
console.print(f"{get_flair()} / $ {an_input}")
|
|
|
|
# Get input command from user
|
|
else:
|
|
# Get input from user using auto-completion
|
|
if session and obbff.USE_PROMPT_TOOLKIT:
|
|
try:
|
|
if obbff.TOOLBAR_HINT:
|
|
an_input = session.prompt(
|
|
f"{get_flair()} / $ ",
|
|
completer=t_controller.completer,
|
|
search_ignore_case=True,
|
|
bottom_toolbar=HTML(
|
|
'<style bg="ansiblack" fg="ansiwhite">[h]</style> help menu '
|
|
'<style bg="ansiblack" fg="ansiwhite">[q]</style> return to previous menu '
|
|
'<style bg="ansiblack" fg="ansiwhite">[e]</style> exit terminal '
|
|
'<style bg="ansiblack" fg="ansiwhite">[cmd -h]</style> '
|
|
"see usage and available options "
|
|
'<style bg="ansiblack" fg="ansiwhite">[about]</style> Getting Started Documentation'
|
|
),
|
|
style=Style.from_dict(
|
|
{
|
|
"bottom-toolbar": "#ffffff bg:#333333",
|
|
}
|
|
),
|
|
)
|
|
else:
|
|
an_input = session.prompt(
|
|
f"{get_flair()} / $ ",
|
|
completer=t_controller.completer,
|
|
search_ignore_case=True,
|
|
)
|
|
except (KeyboardInterrupt, EOFError):
|
|
print_goodbye()
|
|
break
|
|
# Get input from user without auto-completion
|
|
else:
|
|
an_input = input(f"{get_flair()} / $ ")
|
|
|
|
try:
|
|
# Process the input command
|
|
t_controller.queue = t_controller.switch(an_input)
|
|
if an_input in ("q", "quit", "..", "exit"):
|
|
print_goodbye()
|
|
break
|
|
|
|
# Check if the user wants to reset application
|
|
if an_input in ("r", "reset") or t_controller.update_success:
|
|
ret_code = reset(t_controller.queue if t_controller.queue else [])
|
|
if ret_code != 0:
|
|
print_goodbye()
|
|
break
|
|
|
|
except SystemExit:
|
|
logger.exception(
|
|
"The command '%s' doesn't exist on the / menu.",
|
|
an_input,
|
|
)
|
|
console.print(
|
|
f"\nThe command '{an_input}' doesn't exist on the / menu", end=""
|
|
)
|
|
similar_cmd = difflib.get_close_matches(
|
|
an_input.split(" ")[0] if " " in an_input else an_input,
|
|
t_controller.controller_choices,
|
|
n=1,
|
|
cutoff=0.7,
|
|
)
|
|
if similar_cmd:
|
|
if " " in an_input:
|
|
candidate_input = (
|
|
f"{similar_cmd[0]} {' '.join(an_input.split(' ')[1:])}"
|
|
)
|
|
if candidate_input == an_input:
|
|
an_input = ""
|
|
t_controller.queue = []
|
|
console.print("\n")
|
|
continue
|
|
an_input = candidate_input
|
|
else:
|
|
an_input = similar_cmd[0]
|
|
|
|
console.print(f" Replacing by '{an_input}'.")
|
|
t_controller.queue.insert(0, an_input)
|
|
else:
|
|
console.print("\n")
|
|
|
|
|
|
def insert_start_slash(cmds: List[str]) -> List[str]:
|
|
if not cmds[0].startswith("/"):
|
|
cmds[0] = f"/{cmds[0]}"
|
|
if cmds[0].startswith("/home"):
|
|
cmds[0] = f"/{cmds[0][5:]}"
|
|
return cmds
|
|
|
|
|
|
def do_rollover():
|
|
"""RollOver the log file."""
|
|
for handler in logging.getLogger().handlers:
|
|
if isinstance(handler, PathTrackingFileHandler):
|
|
handler.doRollover()
|
|
|
|
|
|
def log_settings() -> None:
|
|
"""Log settings"""
|
|
settings_dict = {}
|
|
settings_dict["tab"] = "True" if obbff.USE_TABULATE_DF else "False"
|
|
settings_dict["cls"] = "True" if obbff.USE_CLEAR_AFTER_CMD else "False"
|
|
settings_dict["color"] = "True" if obbff.USE_COLOR else "False"
|
|
settings_dict["promptkit"] = "True" if obbff.USE_PROMPT_TOOLKIT else "False"
|
|
settings_dict["thoughts"] = "True" if obbff.ENABLE_THOUGHTS_DAY else "False"
|
|
settings_dict["reporthtml"] = "True" if obbff.OPEN_REPORT_AS_HTML else "False"
|
|
settings_dict["exithelp"] = "True" if obbff.ENABLE_EXIT_AUTO_HELP else "False"
|
|
settings_dict["rcontext"] = "True" if obbff.REMEMBER_CONTEXTS else "False"
|
|
settings_dict["rich"] = "True" if obbff.ENABLE_RICH else "False"
|
|
settings_dict["richpanel"] = "True" if obbff.ENABLE_RICH_PANEL else "False"
|
|
settings_dict["ion"] = "True" if obbff.USE_ION else "False"
|
|
settings_dict["watermark"] = "True" if obbff.USE_WATERMARK else "False"
|
|
settings_dict["autoscaling"] = "True" if obbff.USE_PLOT_AUTOSCALING else "False"
|
|
settings_dict["dt"] = "True" if obbff.USE_DATETIME else "False"
|
|
settings_dict["packaged"] = "True" if obbff.PACKAGED_APPLICATION else "False"
|
|
settings_dict["python"] = str(platform.python_version())
|
|
settings_dict["os"] = str(platform.system())
|
|
|
|
logger.info("SETTINGS: %s ", str(settings_dict))
|
|
|
|
do_rollover()
|
|
|
|
|
|
def run_scripts(
|
|
path: str,
|
|
test_mode: bool = False,
|
|
verbose: bool = False,
|
|
routines_args: List[str] = None,
|
|
):
|
|
"""Runs a given .openbb scripts
|
|
|
|
Parameters
|
|
----------
|
|
path : str
|
|
The location of the .openbb file
|
|
test_mode : bool
|
|
Whether the terminal is in test mode
|
|
verbose : bool
|
|
Whether to run tests in verbose mode
|
|
routines_args : List[str]
|
|
One or multiple inputs to be replaced in the routine and separated by commas.
|
|
E.g. GME,AMC,BTC-USD
|
|
"""
|
|
if os.path.isfile(path):
|
|
with open(path) as fp:
|
|
raw_lines = [x for x in fp if (not is_reset(x)) and ("#" not in x) and x]
|
|
raw_lines = [
|
|
raw_line.strip("\n") for raw_line in raw_lines if raw_line.strip("\n")
|
|
]
|
|
|
|
if routines_args:
|
|
lines = list()
|
|
for rawline in raw_lines:
|
|
templine = rawline
|
|
for i, arg in enumerate(routines_args):
|
|
templine = templine.replace(f"$ARGV[{i}]", arg)
|
|
lines.append(templine)
|
|
else:
|
|
lines = raw_lines
|
|
|
|
if test_mode and "exit" not in lines[-1]:
|
|
lines.append("exit")
|
|
|
|
export_folder = ""
|
|
if "export" in lines[0]:
|
|
export_folder = lines[0].split("export ")[1].rstrip()
|
|
lines = lines[1:]
|
|
|
|
simulate_argv = f"/{'/'.join([line.rstrip() for line in lines])}"
|
|
file_cmds = simulate_argv.replace("//", "/home/").split()
|
|
file_cmds = insert_start_slash(file_cmds) if file_cmds else file_cmds
|
|
if export_folder:
|
|
file_cmds = [f"export {export_folder}{' '.join(file_cmds)}"]
|
|
else:
|
|
file_cmds = [" ".join(file_cmds)]
|
|
|
|
if not test_mode:
|
|
terminal(file_cmds, appName="openbb_script")
|
|
# TODO: Add way to track how many commands are tested
|
|
else:
|
|
if verbose:
|
|
terminal(file_cmds, appName="openbb_script")
|
|
else:
|
|
with suppress_stdout():
|
|
terminal(file_cmds, appName="openbb_script")
|
|
|
|
else:
|
|
console.print(f"File '{path}' doesn't exist. Launching base terminal.\n")
|
|
if not test_mode:
|
|
terminal()
|
|
|
|
|
|
def main(
|
|
debug: bool,
|
|
test: bool,
|
|
filtert: str,
|
|
paths: List[str],
|
|
verbose: bool,
|
|
routines_args: List[str] = None,
|
|
):
|
|
"""
|
|
Runs the terminal with various options
|
|
|
|
Parameters
|
|
----------
|
|
debug : bool
|
|
Whether to run the terminal in debug mode
|
|
test : bool
|
|
Whether to run the terminal in integrated test mode
|
|
filtert : str
|
|
Filter test files with given string in name
|
|
paths : List[str]
|
|
The paths to run for scripts or to test
|
|
verbose : bool
|
|
Whether to show output from tests
|
|
routines_args : List[str]
|
|
One or multiple inputs to be replaced in the routine and separated by commas.
|
|
E.g. GME,AMC,BTC-USD
|
|
"""
|
|
|
|
if test:
|
|
os.environ["DEBUG_MODE"] = "true"
|
|
|
|
if paths == []:
|
|
console.print("Please send a path when using test mode")
|
|
return
|
|
test_files = []
|
|
for path in paths:
|
|
if path.endswith(".openbb"):
|
|
file = os.path.join(os.path.abspath(os.path.dirname(__file__)), path)
|
|
test_files.append(file)
|
|
else:
|
|
folder = os.path.join(os.path.abspath(os.path.dirname(__file__)), path)
|
|
files = [
|
|
f"{folder}/{name}"
|
|
for name in os.listdir(folder)
|
|
if os.path.isfile(os.path.join(folder, name))
|
|
and name.endswith(".openbb")
|
|
and (filtert in f"{folder}/{name}")
|
|
]
|
|
test_files += files
|
|
test_files.sort()
|
|
SUCCESSES = 0
|
|
FAILURES = 0
|
|
fails = {}
|
|
length = len(test_files)
|
|
i = 0
|
|
console.print("[green]OpenBB Terminal Integrated Tests:\n[/green]")
|
|
for file in test_files:
|
|
file = file.replace("//", "/")
|
|
repo_path_position = file.rfind(REPOSITORY_DIRECTORY.name)
|
|
if repo_path_position >= 0:
|
|
file_name = file[repo_path_position:].replace("\\", "/")
|
|
else:
|
|
file_name = file
|
|
console.print(f"{file_name} {((i/length)*100):.1f}%")
|
|
try:
|
|
if not os.path.isfile(file):
|
|
raise ValueError("Given file does not exist")
|
|
run_scripts(file, test_mode=True, verbose=verbose)
|
|
SUCCESSES += 1
|
|
except Exception as e:
|
|
fails[file] = e
|
|
FAILURES += 1
|
|
i += 1
|
|
if fails:
|
|
console.print("\n[red]Failures:[/red]\n")
|
|
for key, value in fails.items():
|
|
repo_path_position = key.rfind(REPOSITORY_DIRECTORY.name)
|
|
if repo_path_position >= 0:
|
|
file_name = key[repo_path_position:].replace("\\", "/")
|
|
else:
|
|
file_name = key
|
|
logger.error("%s: %s failed", file_name, value)
|
|
console.print(f"{file_name}: {value}\n")
|
|
console.print(
|
|
f"Summary: [green]Successes: {SUCCESSES}[/green] [red]Failures: {FAILURES}[/red]"
|
|
)
|
|
return
|
|
|
|
if debug:
|
|
os.environ["DEBUG_MODE"] = "true"
|
|
if isinstance(paths, list) and paths[0].endswith(".openbb"):
|
|
run_scripts(paths[0], routines_args=routines_args)
|
|
elif paths:
|
|
argv_cmds = list([" ".join(paths).replace(" /", "/home/")])
|
|
argv_cmds = insert_start_slash(argv_cmds) if argv_cmds else argv_cmds
|
|
terminal(argv_cmds)
|
|
else:
|
|
terminal()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
|
|
parser = argparse.ArgumentParser(
|
|
formatter_class=argparse.ArgumentDefaultsHelpFormatter,
|
|
prog="terminal",
|
|
description="The OpenBB Terminal.",
|
|
)
|
|
parser.add_argument(
|
|
"-d",
|
|
"--debug",
|
|
dest="debug",
|
|
action="store_true",
|
|
default=False,
|
|
help="Runs the terminal in debug mode.",
|
|
)
|
|
parser.add_argument(
|
|
"--file",
|
|
help="The path or .openbb file to run.",
|
|
dest="path",
|
|
nargs="+",
|
|
default="",
|
|
type=str,
|
|
)
|
|
parser.add_argument(
|
|
"-t",
|
|
"--test",
|
|
dest="test",
|
|
action="store_true",
|
|
default=False,
|
|
help="Whether to run in test mode.",
|
|
)
|
|
parser.add_argument(
|
|
"--filter",
|
|
help="Send a keyword to filter in file name",
|
|
dest="filtert",
|
|
default="",
|
|
type=str,
|
|
)
|
|
parser.add_argument(
|
|
"-v", "--verbose", dest="verbose", action="store_true", default=False
|
|
)
|
|
parser.add_argument(
|
|
"-i",
|
|
"--input",
|
|
help=(
|
|
"Select multiple inputs to be replaced in the routine and separated by commas."
|
|
"E.g. GME,AMC,BTC-USD"
|
|
),
|
|
dest="routine_args",
|
|
type=lambda s: [str(item) for item in s.split(",")],
|
|
default=None,
|
|
)
|
|
|
|
if sys.argv[1:] and "-" not in sys.argv[1][0]:
|
|
sys.argv.insert(1, "--file")
|
|
ns_parser = parser.parse_args()
|
|
main(
|
|
ns_parser.debug,
|
|
ns_parser.test,
|
|
ns_parser.filtert,
|
|
ns_parser.path,
|
|
ns_parser.verbose,
|
|
ns_parser.routine_args,
|
|
)
|