Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
parser = logmuse.add_logging_options(build_argparser())
args, remaining_args = parser.parse_known_args()
global _LOGGER
_LOGGER = logmuse.logger_via_cli(args, make_root=True)
_LOGGER.debug("refgenie {}".format(__version__))
_LOGGER.debug("Args: {}".format(args))
if not args.command:
parser.print_help()
_LOGGER.error("No command given")
sys.exit(1)
gencfg = refgenconf.select_genome_config(filename=args.genome_config, check_exist=not args.command == INIT_CMD,
on_missing=lambda fp: fp, strict_env=True)
if gencfg is None:
raise MissingGenomeConfigError(args.genome_config)
_LOGGER.debug("Determined genome config: {}".format(gencfg))
# From user input we want to construct a list of asset dicts, where each
# asset has a genome name, asset name, and tag
if "asset_registry_paths" in args and args.asset_registry_paths:
_LOGGER.debug("Found registry_path: {}".format(args.asset_registry_paths))
asset_list = [parse_registry_path(x) for x in args.asset_registry_paths]
for a in asset_list:
# every asset must have a genome, either provided via registry path
# or the args.genome arg.
if not a["genome"]:
if args.genome:
a["genome"] = args.genome
else:
def __init__(self, conf_file=None):
"""
Create the error message, using optionally an attempt filepath.
:param str conf_file: path attempted to be used as genome config file
"""
msg = "You must provide a config file either as an argument or via an environment variable: {}"\
.format(", ".join(CFG_ENV_VARS))
if conf_file:
msg = "Not a file {} -- {}.".format(conf_file, msg)
super(MissingGenomeConfigError, self).__init__(msg)
def main():
""" main workflow """
parser = build_argparser()
args, remaining_args = parser.parse_known_args()
cfg = select_config(args.config, refgenconf.CFG_ENV_VARS, check_exist=True, strict_env=True)
if not cfg:
raise MissingGenomeConfigError(args.config)
rgc = refgenconf.RefGenConf(filepath=cfg, writable=True)
pths = [args.path, mkabs(args.path, rgc.genome_folder)]
if not untar_or_copy(pths[0], os.path.join(rgc.genome_folder, args.genome)) \
and not untar_or_copy(pths[1], os.path.join(rgc.genome_folder, args.genome)):
rgc.unlock()
raise OSError("Path '{}' does not exist. Tried: {}".format(args.path, " and ".join(pths)))
path_components = [rgc.genome_folder] + [args.genome] + ["*"] * 3 + ["Sequence"]
assets_paths = glob(os.path.join(*path_components))
assert len(assets_paths) > 0, OSError("Your iGenomes directory is corrupted, more than one directory matched by {}."
"\nMatched dirs: {}".format(os.path.join(*path_components),
", ".join(assets_paths)))
assets_path = assets_paths[0]
asset_names = [d for d in os.listdir(assets_path) if os.path.isdir(assets_path)]
processed = []
for a in asset_names:
asset_dict = {"genome": args.genome, "asset": a, "tag": None, "seek_key": None}