diff --git a/openbb_terminal/core/library/__init__.py b/openbb_terminal/core/library/__init__.py deleted file mode 100644 index e69de29bb2d1..000000000000 diff --git a/openbb_terminal/core/library/breadcrumb.py b/openbb_terminal/core/library/breadcrumb.py deleted file mode 100644 index 599b1f9dc83d..000000000000 --- a/openbb_terminal/core/library/breadcrumb.py +++ /dev/null @@ -1,164 +0,0 @@ -from typing import Any, List, Optional - -import openbb_terminal.config_terminal as cfg -from openbb_terminal.core.library.metadata import Metadata -from openbb_terminal.core.library.operation import Operation -from openbb_terminal.core.library.trail_map import TrailMap - -# pylint: disable=import-outside-toplevel - - -class MetadataBuilder: - @staticmethod - def get_option_list(trail: str, trail_map: TrailMap) -> List[str]: - option_list = [] - for key in trail_map.map_dict: - if trail == "": - option = key.split(".")[0] - elif key.startswith(trail) and key[len(trail)] == ".": - option = key[len(trail) + 1 :].split(".")[0] - else: - option = None - - if option: - option_list.append(option) - - return list(set(option_list)) - - @classmethod - def build_dir_list(cls, trail: str, trail_map: TrailMap) -> List[str]: - option_list = cls.get_option_list(trail=trail, trail_map=trail_map) - - option_list_full = [] - for option in option_list: - option_list_full.append(option) - - option_view_trail = f"{trail}.{option}_chart" - if trail_map.get_view_function(trail=option_view_trail): - option_list_full.append(f"{option}_chart") - - return option_list_full - - @staticmethod - def build_docstring(trail: str, dir_list: List[str]) -> str: - if trail == "": - docstring = """This is the OpenBB Terminal SDK. - Use the SDK to get data directly into your jupyter notebook or directly use it in your application. - For more information see the official documentation at: https://openbb-finance.github.io/OpenBBTerminal/SDK/ - """ - else: - docstring = ( - trail.rsplit(".")[-1].upper() - + " Menu\n\nThe SDK commands of the the menu:" - ) - for command in dir_list: - docstring += f"\n\t.{trail}.{command}" - - return docstring - - @classmethod - def build(cls, trail: str, trail_map: TrailMap) -> Metadata: - dir_list = cls.build_dir_list(trail=trail, trail_map=trail_map) - docstring = cls.build_docstring(trail=trail, dir_list=dir_list) - metadata = Metadata( - dir_list=dir_list, - docstring=docstring, - ) - return metadata - - -class Breadcrumb: - __version__ = cfg.VERSION - - def __init__( - self, - metadata: Optional[Metadata] = None, - trail: str = "", - trail_map: Optional[TrailMap] = None, - ) -> None: - """ - Generates a 'trail' that allows accessing OpenBB Terminal SDK methods. - - Example: - openbb.forex.get_currency_list() - Breadcrumb(trail="").Breadcrumb(trail="forex").Operation(trail="forex.get_currency_list")() - - Args: - metadata (Optional[Metadata], optional): - Object to generate Breadcrumb's metadata (__dir__, __doc__). - Defaults to None. - trail (str, optional): - Current trail of the Breadcrumb. - Defaults to "". - trail_map (Optional[TrailMap], optional): - Mapping with all the trails available and matching models and views. - Defaults to None. - """ - trail_map = trail_map or TrailMap() - metadata = metadata or MetadataBuilder.build(trail=trail, trail_map=trail_map) - - self._metadata = metadata - self._trail_map = trail_map - self._trail = trail - - self.__doc__ = metadata.docstring - if trail == "": - BreadcrumbLogger() - - def __dir__(self): - return self._metadata.dir_list - - def __getattr__(self, name: str) -> Any: - trail = self._trail - trail_map = self._trail_map - - trail_next = name if trail == "" else f"{trail}.{name}" - - if trail_map.get_model_function( - trail=trail_next - ) or trail_map.get_view_function(trail=trail_next): - next_crumb: Any = Operation( - trail=trail_next, - trail_map=trail_map, - ) - elif name in self._metadata.dir_list: - next_crumb = Breadcrumb( - metadata=MetadataBuilder.build(trail=trail_next, trail_map=trail_map), - trail=trail_next, - trail_map=trail_map, - ) - else: - raise AttributeError( - f"Module or method '{trail}' has no attribute '{name}'." - ) - - return next_crumb - - def about(self): - import webbrowser - - trail = self._trail - url = "https://docs.openbb.co/sdk/reference/" - url += "/".join(trail.split(".")) - webbrowser.open(url) - - -# pylint: disable=R0903 -class BreadcrumbLogger: - def __init__(self) -> None: - self.__check_initialize_logging() - - def __check_initialize_logging(self): - if not cfg.LOGGING_SUPPRESS: - self.__initialize_logging() - - @staticmethod - def __initialize_logging() -> None: - from openbb_terminal.core.log.generation.settings_logger import ( # pylint: disable=C0415 - log_all_settings, - ) - from openbb_terminal.loggers import setup_logging # pylint: disable=C0415 - - cfg.LOGGING_SUB_APP = "sdk" - setup_logging() - log_all_settings() diff --git a/openbb_terminal/core/library/metadata.py b/openbb_terminal/core/library/metadata.py deleted file mode 100644 index 68e47c1db8d7..000000000000 --- a/openbb_terminal/core/library/metadata.py +++ /dev/null @@ -1,19 +0,0 @@ -from typing import List - - -class Metadata: - @property - def dir_list(self) -> List[str]: - return self.__dir_list - - @property - def docstring(self) -> str: - return self.__docstring - - def __init__( - self, - dir_list: List[str], - docstring: str = "", - ) -> None: - self.__dir_list = dir_list - self.__docstring = docstring diff --git a/openbb_terminal/core/library/operation.py b/openbb_terminal/core/library/operation.py deleted file mode 100644 index 1483be1398cf..000000000000 --- a/openbb_terminal/core/library/operation.py +++ /dev/null @@ -1,291 +0,0 @@ -import json -from importlib import import_module -from inspect import signature -from logging import Logger, getLogger -from typing import Any, Callable, Dict, List, Optional - -import openbb_terminal.config_terminal as cfg -from openbb_terminal.core.library.metadata import Metadata -from openbb_terminal.core.library.trail_map import TrailMap - -# pylint: disable=import-outside-toplevel - - -class MetadataBuilder: - def __init__( - self, - method: Callable, - trail: str, - trail_map: TrailMap, - ) -> None: - self.__method = method - self.__trail = trail - self.__trail_map = trail_map - - @staticmethod - def build_docstring( - method: Callable, - trail: str, - trail_map: TrailMap, - ) -> str: - doc = "" - - view_trail = f"{trail}_chart" - if trail_map.get_view_function(view_trail): - doc += f"Use '{view_trail}' to access the view.\n" - - if method.__doc__: - doc += method.__doc__ - - return doc - - def build(self) -> Metadata: - method = self.__method - trail = self.__trail - trail_map = self.__trail_map - - dir_list: List[str] = [] - docstring = self.build_docstring( - method=method, - trail=trail, - trail_map=trail_map, - ) - metadata = Metadata(dir_list=dir_list, docstring=docstring) - - return metadata - - -class Operation: - @staticmethod - def __get_method(method_path: str) -> Callable: - module_path, function_name = method_path.rsplit(".", 1) - module = import_module(module_path) - func = getattr(module, function_name) - - return func - - def __init__( - self, - trail: str, - trail_map: Optional[TrailMap] = None, - metadata: Optional[Metadata] = None, - ) -> None: - trail_map = trail_map or TrailMap() - - method_path = trail_map.get_model_or_view(trail=trail) - method = self.__get_method(method_path=method_path) - metadata = ( - metadata - or MetadataBuilder(method=method, trail=trail, trail_map=trail_map).build() - ) - - self._trail = trail - self._trail_map = trail_map - self._method = method - self.__doc__ = metadata.docstring - self.__signature__ = signature(method) - - def __call__(self, *args: Any, **kwargs: Any) -> Any: - method = self._method - - operation_logger = OperationLogger( - trail=self._trail, - method_chosen=method, - args=args, - kwargs=kwargs, - ) - - operation_logger.log_before_call() - - method_result = method(*args, **kwargs) - - operation_logger.log_after_call(method_result=method_result) - - return method_result - - def about(self): - import webbrowser - - trail = self._trail - url = "https://docs.openbb.co/sdk/reference/" - url += "/".join(trail.split(".")) - webbrowser.open(url) - - -class OperationLogger: - last_method: Dict[Any, Any] = {} - - def __init__( - self, - trail: str, - method_chosen: Callable, - args: Any, - kwargs: Any, - logger: Optional[Logger] = None, - ) -> None: - self.__trail = trail - self.__method_chosen = method_chosen - self.__logger = logger or getLogger(self.__method_chosen.__module__) - self.__args = args - self.__kwargs = kwargs - - def log_before_call( - self, - ): - if self.__check_logging_conditions(): - logger = self.__logger - self.__log_start(logger=logger, method_chosen=self.__method_chosen) - self.__log_method_info( - logger=logger, - trail=self.__trail, - method_chosen=self.__method_chosen, - args=self.__args, - kwargs=self.__kwargs, - ) - - @staticmethod - def __log_start(logger: Logger, method_chosen: Callable): - logger.info( - "START", - extra={"func_name_override": method_chosen.__name__}, - ) - - def __log_method_info( - self, - logger: Logger, - trail: str, - method_chosen: Callable, - args: Any, - kwargs: Any, - ): - merged_args = self.__merge_function_args(method_chosen, args, kwargs) - merged_args = self.__remove_key_and_log_state( - method_chosen.__module__, merged_args - ) - - logging_info: Dict[str, Any] = {} - logging_info["INPUT"] = { - key: str(value)[:100] for key, value in merged_args.items() - } - logging_info["VIRTUAL_PATH"] = trail - logging_info["CHART"] = "chart" in kwargs and kwargs["chart"] is True - - logger.info( - f"{json.dumps(logging_info)}", - extra={"func_name_override": method_chosen.__name__}, - ) - - @staticmethod - def __merge_function_args(func: Callable, args: tuple, kwargs: dict) -> dict: - """ - Merge user input args and kwargs with signature defaults into a dictionary. - - Parameters - ---------- - - func : Callable - Function to get the args from - args : tuple - Positional args - kwargs : dict - Keyword args - - Returns - ---------- - dict - Merged user args and signature defaults - """ - import inspect # pylint: disable=C0415 - - sig = inspect.signature(func) - sig_args = { - param.name: param.default - for param in sig.parameters.values() - if param.default is not inspect.Parameter.empty - } - # merge args with sig_args - sig_args.update(dict(zip(sig.parameters, args))) - # add kwargs elements to sig_args - sig_args.update(kwargs) - return sig_args - - @staticmethod - def __remove_key_and_log_state(func_module: str, function_args: dict) -> dict: - """ - Remove API key from the function args and log state of keys. - - Parameters - ---------- - func_module : str - Module of the function - function_args : dict - Function args - - Returns - ---------- - dict - Function args with API key removed - """ - - if func_module == "openbb_terminal.keys_model": - from openbb_terminal.core.log.generation.settings_logger import ( # pylint: disable=import-outside-toplevel - log_keys, - ) - - # remove key if defined - function_args.pop("key", None) - log_keys() - return function_args - - def log_after_call( - self, - method_result: Any, - ): - if self.__check_logging_conditions(): - logger = self.__logger - self.__log_exception_if_any( - logger=logger, - method_result=method_result, - method_chosen=self.__method_chosen, - ) - self.__log_end( - logger=logger, - method_chosen=self.__method_chosen, - ) - OperationLogger.last_method = { - f"{self.__method_chosen.__module__}.{self.__method_chosen.__name__}": { - "args": str(self.__args)[:100], - "kwargs": str(self.__kwargs)[:100], - } - } - - @staticmethod - def __log_exception_if_any( - logger: Logger, - method_chosen: Callable, - method_result: Any, - ): - if isinstance(method_result, Exception): - logger.exception( - f"Exception: {method_result}", - extra={"func_name_override": method_chosen.__name__}, - ) - - @staticmethod - def __log_end(logger: Logger, method_chosen: Callable): - logger.info( - "END", - extra={"func_name_override": method_chosen.__name__}, - ) - - def __check_logging_conditions(self) -> bool: - return not cfg.LOGGING_SUPPRESS and not self.__check_last_method() - - def __check_last_method(self) -> bool: - current_method = { - f"{self.__method_chosen.__module__}.{self.__method_chosen.__name__}": { - "args": str(self.__args)[:100], - "kwargs": str(self.__kwargs)[:100], - } - } - return OperationLogger.last_method == current_method diff --git a/openbb_terminal/core/library/trail_map.py b/openbb_terminal/core/library/trail_map.py deleted file mode 100644 index 6d10b1ea385d..000000000000 --- a/openbb_terminal/core/library/trail_map.py +++ /dev/null @@ -1,141 +0,0 @@ -from pathlib import Path -from typing import Dict, Optional - -import dotenv -import pandas as pd - -from openbb_terminal.base_helpers import load_env_vars, strtobool -from openbb_terminal.core.config.paths import MISCELLANEOUS_DIRECTORY, SETTINGS_ENV_FILE -from openbb_terminal.rich_config import console - -DISABLE_FORECASTING_WARNING = load_env_vars( - "OPENBB_DISABLE_FORECASTING_WARNING", strtobool, False -) -try: - import darts # pylint: disable=W0611 # noqa: F401 - from darts import utils # pylint: disable=W0611 # noqa: F401 - - FORECASTING_TOOLKIT_ENABLED = True -except ModuleNotFoundError: - FORECASTING_TOOLKIT_ENABLED = False - if not DISABLE_FORECASTING_WARNING: - dotenv.set_key( - str(SETTINGS_ENV_FILE), "OPENBB_DISABLE_FORECASTING_WARNING", "True" - ) - console.print( - "[yellow]" - "Forecasting Toolkit is disabled. " - "To use the Forecasting features please install the toolkit following the " - "instructions here: https://docs.openbb.co/sdk/quickstart/installation/" - "\n" - "[/yellow]" - ) - -DISABLE_OPTIMIZATION_WARNING = load_env_vars( - "OPENBB_DISABLE_OPTIMIZATION_WARNING", strtobool, False -) -try: - # import riskfolio # pylint: disable=W0611 # noqa: F401 - - OPTIMIZATION_TOOLKIT_ENABLED = True -except ModuleNotFoundError: - OPTIMIZATION_TOOLKIT_ENABLED = False - if not DISABLE_OPTIMIZATION_WARNING: - dotenv.set_key( - str(SETTINGS_ENV_FILE), "OPENBB_DISABLE_OPTIMIZATION_WARNING", "True" - ) - console.print( - "[yellow]" - "Portfolio Optimization Toolkit is disabled. " - "To use the Optimization features please install the toolkit following the " - "instructions here: https://docs.openbb.co/sdk/quickstart/installation/" - "\n" - "[/yellow]" - ) - - -class TrailMap: - MAP_PATH = MISCELLANEOUS_DIRECTORY / "library" / "trail_map.csv" - MAP_FORECASTING_PATH = ( - MISCELLANEOUS_DIRECTORY / "library" / "trail_map_forecasting.csv" - ) - MAP_OPTIMIZATION_PATH = ( - MISCELLANEOUS_DIRECTORY / "library" / "trail_map_optimization.csv" - ) - - @classmethod - def load_csv(cls, path: Optional[Path] = None) -> Dict[str, Dict[str, str]]: - path = path or cls.MAP_PATH - df = pd.read_csv(path, keep_default_na=False) - df.set_index("trail", inplace=True) - return df.to_dict(orient="index") - - @classmethod - def save_csv( - cls, map_dict: Dict[str, Dict[str, str]], path: Optional[Path] = None - ) -> None: - path = path or cls.MAP_PATH - df = pd.DataFrame.from_dict(data=map_dict, orient="index") - df.index.name = "trail" - df.to_csv(path_or_buf=path) - - @property - def map_dict(self) -> Dict[str, Dict[str, str]]: - return self.__map_dict - - def get_view_function(self, trail: str) -> Optional[str]: - """Retrieves the view function from the mapping. - - Views ends with "_chart". - - Args: - trail (str): Trail like "futures.curves_chart" - - Returns: - Optional[str]: View function import path. - """ - - map_dict = self.__map_dict - trail = trail[: -len("_chart")] - - view = None - if trail in map_dict and "view" in map_dict[trail] and map_dict[trail]["view"]: - view = map_dict[trail]["view"] - - return view - - def get_model_function(self, trail: str) -> Optional[str]: - map_dict = self.map_dict - - model = None - if trail in map_dict and "view" in map_dict[trail] and map_dict[trail]["model"]: - model = map_dict[trail]["model"] - - return model - - def get_model_or_view(self, trail: str) -> str: - model_path = self.get_model_function(trail=trail) - view_path = self.get_view_function(trail=trail) - - if model_path: - method_path = model_path - elif view_path: - method_path = view_path - else: - raise AttributeError(f"Unknown method : {trail}") - - return method_path - - def load(self): - map_dict = self.load_csv(path=self.MAP_PATH) - if FORECASTING_TOOLKIT_ENABLED: - map_forecasting_dict = self.load_csv(path=self.MAP_FORECASTING_PATH) - map_dict = {**map_dict, **map_forecasting_dict} - if OPTIMIZATION_TOOLKIT_ENABLED: - map_optimization_dict = self.load_csv(path=self.MAP_OPTIMIZATION_PATH) - map_dict = {**map_dict, **map_optimization_dict} - - self.__map_dict = map_dict - - def __init__(self): - self.load()