Source code for gnomad.resources.import_resources

# noqa: D100

import argparse
import itertools
import textwrap
from inspect import getmembers
from typing import Dict, Optional, Tuple

import gnomad.resources.grch37 as grch37
import gnomad.resources.grch38 as grch38
from gnomad.resources.config import (
    GnomadPublicResourceSource,
    gnomad_public_resource_configuration,
)
from gnomad.resources.resource_utils import BaseResource, BaseVersionedResource


# Generate a dictionary of resource available for import for a given genome build
[docs]def get_module_importable_resources( module, prefix: Optional[str] = None ) -> Dict[str, Tuple[str, BaseResource]]: """ Take a module that was imported and generates a list of all resources in this module that can be imported (i.e. with a path and import_func). The dict produced is as follows: - keys: {prefix}.{resource_name}.{version} (with prefix only present if `prefix` is set, and `version` only present for versioned resources) - values: ({resource_name}[ version {version}], resource) with resource_name set to the variable name in the module and the version present for versioned resources. The following example will generate a dict with all the resources in gnomad.resources.grch37 that can be imported: .. code-block:: python import gnomad.resources.grch37 as grch37 grch37_resources = get_module_importable_resources(grch37, prefix='grch37') :param module: Input module :param prefix: :return: """ _prefix = f"{prefix}." if prefix else "" resources = {} for name, obj in getmembers(module): if isinstance(obj, BaseResource) and obj.path and obj.import_func: resources[f"{_prefix}{name}"] = (name, obj) if isinstance(obj, BaseVersionedResource): for version_name, version_resource in obj.versions.items(): if version_resource.path and version_resource.import_func: resources[f"{_prefix}{name}.{version_name}"] = ( f"{name}.{version_name}", version_resource, ) return resources
[docs]def get_resources_descriptions( resources: Dict[str, Tuple[str, BaseResource]], width: Optional[int] = 100 ) -> str: """ Return a string listing all resources in the input dict along with the path from which they are imported and the path at which they are stored. :param resources: A dict returned from get_module_importable_resources :param width: Maximum width of lines in the returned string """ wrapper = textwrap.TextWrapper( width=width, initial_indent=" " * 2, subsequent_indent=" " * 4 ) return "\n".join( itertools.chain.from_iterable( [ f"{resource_arg}:", wrapper.fill( f"import {getattr(resource, 'import_args', {}).get('path', '???')}" ), wrapper.fill(f"to {resource.path}"), "", ] for resource_arg, (resource_name, resource) in resources.items() ) )
grch37_resources = get_module_importable_resources(grch37, "grch37") grch38_resources = get_module_importable_resources(grch38, "grch38") all_resources = {**grch37_resources, **grch38_resources}
[docs]def main(args): """Import selected resources.""" gnomad_public_resource_configuration.source = GnomadPublicResourceSource.GNOMAD for resource_arg in args.resources: resource_name, resource = all_resources[resource_arg] print(f"Importing {resource_name}...") resource.import_resource(args.overwrite)
if __name__ == "__main__": parser = argparse.ArgumentParser(formatter_class=argparse.RawTextHelpFormatter) parser.add_argument( "resources", choices=list(all_resources.keys()), metavar="resource", nargs="+", help="Resource to import. Choices are:\n\n" + get_resources_descriptions(all_resources), ) parser.add_argument( "--overwrite", help="Overwrites existing files", action="store_true" ) main(parser.parse_args())