mirror of
https://github.com/OpenBB-finance/OpenBB.git
synced 2026-10-30 14:02:06 +00:00
adding subcontrollers
This commit is contained in:
94
openbb_terminal/base_platform_controller.py
Normal file
94
openbb_terminal/base_platform_controller.py
Normal file
@@ -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)
|
||||
@@ -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")
|
||||
|
||||
@@ -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."""
|
||||
|
||||
Reference in New Issue
Block a user