diff --git a/openbb_terminal/base_platform_controller.py b/openbb_terminal/base_platform_controller.py new file mode 100644 index 00000000000..2f5d058cc36 --- /dev/null +++ b/openbb_terminal/base_platform_controller.py @@ -0,0 +1,94 @@ +"""Platform Equity Controller.""" +__docformat__ = "numpy" + +import logging +from functools import partial, update_wrapper +from types import MethodType +from typing import Dict, List, Optional + +from openbb_terminal.core.session.current_user import get_current_user +from openbb_terminal.custom_prompt_toolkit import NestedCompleter +from openbb_terminal.menu import session +from openbb_terminal.parent_classes import BaseController +from openbb_terminal.rich_config import MenuText, console + +logger = logging.getLogger(__name__) + + +class PlatformController(BaseController): + """Platform Controller Base class.""" + + CHOICES_COMMANDS = [] + CHOICES_GENERATION = True + + def __init__( + self, + name: str, + queue: Optional[List[str]] = None, + translators: Optional[Dict] = None, + ): + """Construct Data.""" + self.PATH = f"/{name}/" + super().__init__(queue) + self._name = name + + if translators: + self._generate_commands(translators=translators) + self.CHOICES_COMMANDS = translators.keys() + + if session and get_current_user().preferences.USE_PROMPT_TOOLKIT: + choices: dict = self.choices_default + self.completer = NestedCompleter.from_nested_dict(choices) + + def _generate_commands(self, translators): + for name, translator in translators.items(): + self._generate_command_call(name=name, translator=translator) + + def _generate_command_call(self, name, translator): + def method(self, other_args: List[str], translator=translator): + parser = translator.parser + + if ns_parser := self.parse_known_args_and_warn(parser, other_args): + try: + obbject = translator.execute_func(parsed_args=ns_parser) + console.print(obbject.to_dataframe()) + + except Exception as e: + console.print(f"[red]{e}[/]\n") + return + + # Bind the method to the class + bound_method = MethodType(method, self) + + # Update the wrapper and set the attribute + bound_method = update_wrapper( + partial(bound_method, translator=translator), method + ) + setattr(self, f"call_{name}", bound_method) + + def _generate_controller_call(self, controller, name, translators): + def method(self, _, controller=controller, name=name, translators=translators): + self.queue = self.load_class( + controller, name, self.queue, translators=translators + ) + + # Bind the method to the class + bound_method = MethodType(method, self) + + # Update the wrapper and set the attribute + bound_method = update_wrapper( + partial( + bound_method, controller=controller, name=name, translators=translators + ), + method, + ) + setattr(self, f"call_{name}", bound_method) + + def print_help(self): + """Print help.""" + mt = MenuText(self._name, 80) + + for command in self.CHOICES_COMMANDS: + mt.add_cmd(command) + + console.print(text=mt.menu_text, menu=self._name) diff --git a/openbb_terminal/poc_platform_equity_controller.py b/openbb_terminal/poc_platform_equity_controller.py index 87593e57945..66e5aee9d45 100644 --- a/openbb_terminal/poc_platform_equity_controller.py +++ b/openbb_terminal/poc_platform_equity_controller.py @@ -2,17 +2,15 @@ __docformat__ = "numpy" import logging -from functools import partial, update_wrapper -from types import MethodType from typing import List, Optional from openbb import obb from argparse_translator.argparse_class_processor import ArgparseClassProcessor +from openbb_terminal.base_platform_controller import PlatformController from openbb_terminal.core.session.current_user import get_current_user from openbb_terminal.custom_prompt_toolkit import NestedCompleter from openbb_terminal.menu import session -from openbb_terminal.parent_classes import BaseController from openbb_terminal.rich_config import MenuText, console # pylint: disable=R1710,import-outside-toplevel @@ -22,64 +20,65 @@ logger = logging.getLogger(__name__) equity = ArgparseClassProcessor(target_class=obb.equity) -class PocPlatformEquityController(BaseController): - """Forex Controller class.""" +class PocPlatformEquityController(PlatformController): + """Equity Controller class.""" CHOICES_COMMANDS = list(equity.translators.keys()) + CHOICES_MENUS = [] + + for key, value in equity.paths.items(): + if value == "path": + continue + CHOICES_MENUS.append(key) - PATH = "/pocplat/" # FILE_PATH = os.path.join(os.path.dirname(__file__), "README.md") CHOICES_GENERATION = True - def __init__(self, queue: Optional[List[str]] = None): + def __init__(self, name: str, queue: Optional[List[str]] = None): """Construct Data.""" - super().__init__(queue) + super().__init__(name, queue) - self._generate_commands() + self._slave_controllers = {} + + self._handle_paths(paths=equity.paths) + self._generate_commands(translators=equity.translators) if session and get_current_user().preferences.USE_PROMPT_TOOLKIT: choices: dict = self.choices_default self.completer = NestedCompleter.from_nested_dict(choices) - def _generate_commands(self): - for name, translator in equity.translators.items(): + def _handle_paths(self, paths: dict): + """Handle paths.""" + for path in paths: + if paths[path] == "path": + continue - def method(self, other_args: List[str], translator=translator): - parser = translator.parser + sub_menu_translators = {} + choices_commands = [] - if ns_parser := self.parse_known_args_and_warn(parser, other_args): - try: - obbject = translator.execute_func(parsed_args=ns_parser) - console.print(obbject.to_dataframe()) + for translator_name, translator in equity.translators.items(): + if f"{self._name}_{path}" in translator_name: + new_name = translator_name.replace(f"{self._name}_{path}_", "") + sub_menu_translators[new_name] = translator + choices_commands.append(new_name) - except Exception as e: - console.print(f"[red]{e}[/]\n") - return + PlatformController.CHOICES_COMMANDS.extend(choices_commands) - # Bind the method to the class - bound_method = MethodType(method, self.__class__) - - # Update the wrapper and set the attribute - bound_method = update_wrapper( - partial(bound_method, translator=translator), method + self._generate_controller_call( + controller=PlatformController, + name=path, + translators=sub_menu_translators, ) - setattr(self.__class__, f"call_{name}", bound_method) def print_help(self): """Print help.""" mt = MenuText("poc_plat_equity/", 80) menus = equity.paths - commands = self.CHOICES_COMMANDS.copy() - for path in menus: - if equity.paths[path] == "path": + for key, value in menus.items(): + if value == "path": continue - mt.add_menu(path) - new_commands = [command for command in commands if path in command] - for command in new_commands: - mt.add_cmd(command) - - mt.add_raw("\n") + mt.add_menu(key) console.print(text=mt.menu_text, menu="POC Platform Equity") diff --git a/openbb_terminal/terminal_controller.py b/openbb_terminal/terminal_controller.py index c9c5ec3dbd8..caa6849e1b8 100644 --- a/openbb_terminal/terminal_controller.py +++ b/openbb_terminal/terminal_controller.py @@ -613,7 +613,7 @@ class TerminalController(BaseController): PocPlatformEquityController, ) - self.queue = self.load_class(PocPlatformEquityController, self.queue) + self.queue = self.load_class(PocPlatformEquityController, "equity", self.queue) def call_intro(self, _): """Process intro command."""