Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

PERF: refactor atef.bin cli entrypoint to defer imports #256

Closed
wants to merge 3 commits into from
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
61 changes: 0 additions & 61 deletions atef/bin/check.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,6 @@
"""
from __future__ import annotations

import argparse
import asyncio
import enum
import itertools
Expand Down Expand Up @@ -79,66 +78,6 @@ def set_or_clear(verbosity: cls, name: str, value: bool) -> cls:
return verbosity


def build_arg_parser(argparser=None):
if argparser is None:
argparser = argparse.ArgumentParser()

argparser.description = DESCRIPTION
argparser.formatter_class = argparse.RawTextHelpFormatter

argparser.add_argument(
"filename",
type=str,
help="Configuration filename",
)

for setting in VerbositySetting:
flag_name = setting.name.replace("_", "-")
if setting == VerbositySetting.default:
continue

help_text = setting.name.replace("_", " ").capitalize()

argparser.add_argument(
f"--{flag_name}",
dest=setting.name,
help=help_text,
action="store_true",
default=setting in VerbositySetting.default,
)

if flag_name.startswith("show-"):
hide_flag_name = flag_name.replace("show-", "hide-")
help_text = help_text.replace("Show ", "Hide ")
argparser.add_argument(
f"--{hide_flag_name}",
dest=setting.name,
help=help_text,
action="store_false",
)

# argparser.add_argument(
# "--filter",
# type=str,
# nargs="*",
# dest="name_filter",
# help="Limit checkout to the named device(s) or identifiers",
# )

argparser.add_argument(
"-p", "--parallel",
action="store_true",
help="Acquire data for comparisons in parallel",
)

argparser.add_argument(
"-r", "--report-path",
help="Path to the report save path, if provided"
)

return argparser


default_severity_to_rich = {
Severity.success: "[bold green]:heavy_check_mark:",
Severity.warning: "[bold yellow]:heavy_check_mark:",
Expand Down
65 changes: 1 addition & 64 deletions atef/bin/config.py
Original file line number Diff line number Diff line change
@@ -1,82 +1,19 @@
"""
`atef config` opens up a graphical config file editor.
"""
import argparse
import logging
import sys
from typing import List, Optional

from pydm import exception
from qtpy.QtWidgets import QApplication, QStyleFactory
from qtpy.QtWidgets import QApplication

from ..type_hints import AnyPath
from ..widgets.config.window import Window

logger = logging.getLogger(__name__)


def build_arg_parser(argparser=None):
if argparser is None:
argparser = argparse.ArgumentParser()

# Arguments that need to be passed through to Qt
qt_args = {
'--qmljsdebugger': 1,
'--reverse': '?',
'--stylesheet': 1,
'--widgetcount': '?',
'--platform': 1,
'--platformpluginpath': 1,
'--platformtheme': 1,
'--plugin': 1,
'--qwindowgeometry': 1,
'--qwindowicon': 1,
'--qwindowtitle': 1,
'--session': 1,
'--display': 1,
'--geometry': 1
}

for name in qt_args:
argparser.add_argument(
name,
type=str,
nargs=qt_args[name]
)

argparser.add_argument(
'--style',
type=str,
choices=QStyleFactory.keys(),
default='fusion',
help='Qt style to use for the application'
)

argparser.description = """
Runs the atef configuration GUI, optionally with an existing configuration.
Qt arguments are also supported. For a full list, see the Qt docs:
https://doc.qt.io/qt-5/qapplication.html#QApplication
https://doc.qt.io/qt-5/qguiapplication.html#supported-command-line-options
"""
argparser.add_argument(
"--cache-size",
metavar="cache_size",
type=int,
default=5,
help="Page widget cache size",
)

argparser.add_argument(
"filenames",
metavar="filename",
type=str,
nargs="*",
help="Configuration filename",
)

return argparser


def main(cache_size: int, filenames: Optional[List[AnyPath]] = None, **kwargs):
app = QApplication(sys.argv)
main_window = Window(cache_size=cache_size, show_welcome=not filenames)
Expand Down
68 changes: 22 additions & 46 deletions atef/bin/main.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,62 +7,34 @@

import argparse
import asyncio
import importlib
import logging
from inspect import iscoroutinefunction

import atef
from atef.bin.subparsers import SUBCOMMANDS

DESCRIPTION = __doc__


COMMAND_TO_MODULE = {
"check": "check",
"config": "config",
"scripts": "scripts",
}


def _try_import(module_name):
return importlib.import_module(f".{module_name}", 'atef.bin')


def _build_commands():
global DESCRIPTION
result = {}
unavailable = []

for command, module_name in sorted(COMMAND_TO_MODULE.items()):
try:
module = _try_import(module_name)
except Exception as ex:
unavailable.append((command, ex))
else:
result[module_name] = (module.build_arg_parser, module.main)
DESCRIPTION += f'\n $ atef {command} --help'

if unavailable:
DESCRIPTION += '\n\n'

for command, ex in unavailable:
DESCRIPTION += (
f'\nWARNING: "atef {command}" is unavailable due to:'
f'\n\t{ex.__class__.__name__}: {ex}'
)

return result


COMMANDS = _build_commands()
def main():
"""
Create the top-level parser for atef. Gathers subparsers from
atef.bin.subparsers, which have been separated to avoid pre-mature imports

Expects SUBCOMMANDS to be a dictionary mapping subcommand name to a tuple of:
- sub-parser builder function: Callable[[], argparse.ArgumentParser]
- function returning the main function for the sub command:
Callable[[], Callable[**subcommand_kwargs]]

def main():
Have fun "parsing" this ;D
"""
top_parser = argparse.ArgumentParser(
prog='atef',
description=DESCRIPTION,
formatter_class=argparse.RawTextHelpFormatter
)

desc = DESCRIPTION

top_parser.add_argument(
'--version', '-V',
action='version',
Expand All @@ -78,11 +50,14 @@ def main():
)

subparsers = top_parser.add_subparsers(help='Possible subcommands')
for command_name, (build_func, main) in COMMANDS.items():
for command_name, (build_func, main) in SUBCOMMANDS.items():
desc += f'\n $ atef {command_name} --help'
sub = subparsers.add_parser(command_name)
build_func(sub)
sub.set_defaults(func=main)

top_parser.description = desc

args = top_parser.parse_args()
kwargs = vars(args)
log_level = kwargs.pop('log_level')
Expand All @@ -93,11 +68,12 @@ def main():

if hasattr(args, 'func'):
func = kwargs.pop('func')
logger.debug('%s(**%r)', func.__name__, kwargs)
if iscoroutinefunction(func):
asyncio.run(func(**kwargs))
logger.debug('main(**%r)', kwargs)
main_fn = func()
if iscoroutinefunction(main_fn):
asyncio.run(main_fn(**kwargs))
else:
func(**kwargs)
main_fn(**kwargs)
else:
top_parser.print_help()

Expand Down
60 changes: 1 addition & 59 deletions atef/bin/scripts.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,67 +2,9 @@
`atef scripts` runs helper scripts. Scripts may be added over time.
"""

import argparse
import importlib
import logging
from pkgutil import iter_modules
from typing import Callable, Dict, Tuple

logger = logging.getLogger(__name__)

DESCRIPTION = __doc__


def gather_scripts() -> Dict[str, Tuple[Callable, Callable]]:
"""Gather scripts, one main function from each submodule"""
# similar to main's _build_commands
global DESCRIPTION
DESCRIPTION += "\nTry:\n"
results = {}
unavailable = []

scripts_module = importlib.import_module("atef.scripts")
for sub_module in iter_modules(scripts_module.__path__):
module_name = sub_module.name
try:
module = importlib.import_module(f".{module_name}", "atef.scripts")
except Exception as ex:
unavailable.append((module_name, ex))
else:
results[module_name] = (module.build_arg_parser, module.main)
DESCRIPTION += f'\n $ atef scripts {module_name} --help'

if unavailable:
DESCRIPTION += '\n\n'

for command, ex in unavailable:
DESCRIPTION += (
f'\nWARNING: "atef scripts {command}" is unavailable due to:'
f'\n\t{ex.__class__.__name__}: {ex}'
)

return results


SCRIPTS = gather_scripts()


def build_arg_parser(argparser=None):
if argparser is None:
argparser = argparse.ArgumentParser()

argparser.description = """
Runs atef related scripts. Pick a subcommand to run its script
"""

sub_parsers = argparser.add_subparsers(help='available script subcommands')
for script_name, (build_parser_func, script_main) in SCRIPTS.items():
sub = sub_parsers.add_parser(script_name)
build_parser_func(sub)
sub.set_defaults(func=script_main)

return argparser


def main():
"""Here as a formality, this is itself a subcommand"""
print(DESCRIPTION)
Loading
Loading