Source code for datalad.cmdline.helpers

# emacs: -*- mode: python; py-indent-offset: 4; tab-width: 4; indent-tabs-mode: nil -*-
# ex: set sts=4 ts=4 sw=4 noet:
# ## ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ##
#
#   See COPYING file distributed along with the datalad package for the
#   copyright and license terms.
#
# ## ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ##
"""
"""

__docformat__ = 'restructuredtext'

import argparse
import os
import re
import sys
import gzip
from tempfile import NamedTemporaryFile
from textwrap import wrap

from ..cmd import WitlessRunner as Runner
from ..interface.common_opts import eval_defaults
from ..log import is_interactive
from ..utils import (
    ensure_unicode,
    getpwd,
    unlink,
)
from ..version import __version__
from ..dochelpers import exc_str

from logging import getLogger
lgr = getLogger('datalad.cmdline')


[docs]class HelpAction(argparse.Action): def __call__(self, parser, namespace, values, option_string=None): # Lets use the manpage on mature systems but only for subcommands -- # --help should behave similar to how git does it: # regular --help for "git" but man pages for specific commands. # It is important since we do discover all subcommands from entry # points at run time and thus any static manpage would like be out of # date interactive = is_interactive() if interactive \ and option_string == '--help' \ and ' ' in parser.prog: # subcommand try: import subprocess # get the datalad manpage to use manfile = os.environ.get('MANPATH', '/usr/share/man') \ + '/man1/{0}.1.gz'.format(parser.prog.replace(' ', '-')) # extract version field from the manpage if not os.path.exists(manfile): raise IOError("manfile is not found") with gzip.open(manfile) as f: man_th = [line for line in f if line.startswith(b".TH")][0] man_version = man_th.split(b' ')[-1].strip(b" '\"\t\n").decode('utf-8') # don't show manpage if man_version not equal to current datalad_version if __version__ != man_version: raise ValueError subprocess.check_call( 'man %s 2> /dev/null' % manfile, shell=True) sys.exit(0) except (subprocess.CalledProcessError, IOError, OSError, IndexError, ValueError) as e: lgr.debug("Did not use manpage since %s", exc_str(e)) if option_string == '-h': usage = parser.format_usage() ucomps = re.match( r'(?P<pre>.*){(?P<cmds>.*)}(?P<post>....*)', usage, re.DOTALL) if ucomps: ucomps = ucomps.groupdict() indent_level = len(ucomps['post']) - len(ucomps['post'].lstrip()) usage = '{pre}{{{cmds}}}{post}'.format( pre=ucomps['pre'], cmds='\n'.join(wrap( ', '.join(sorted(c.strip() for c in ucomps['cmds'].split(','))), break_on_hyphens=False, subsequent_indent=' ' * indent_level)), post=ucomps['post'], ) helpstr = "%s\n%s" % ( usage, "Use '--help' to get more comprehensive information.") else: helpstr = parser.format_help() # better for help2man # for main command -- should be different sections. And since we are in # heavy output massaging mode... if "commands for dataset operations" in helpstr.lower(): opt_args_str = '*Global options*' pos_args_str = '*Commands*' # tune up usage -- default one is way too heavy helpstr = re.sub(r'^[uU]sage: .*?\n\s*\n', 'Usage: datalad [global-opts] command [command-opts]\n\n', helpstr, flags=re.MULTILINE | re.DOTALL) # and altogether remove sections with long list of commands helpstr = re.sub(r'positional arguments:\s*\n\s*{.*}\n', '', helpstr) else: opt_args_str = "*Options*" pos_args_str = "*Arguments*" helpstr = re.sub(r'optional arguments:', opt_args_str, helpstr) helpstr = re.sub(r'positional arguments:', pos_args_str, helpstr) # usage is on the same line helpstr = re.sub(r'^usage:', 'Usage:', helpstr) if interactive and option_string == '--help': import pydoc pydoc.pager(helpstr) else: print(helpstr) sys.exit(0)
[docs]class LogLevelAction(argparse.Action): def __call__(self, parser, namespace, values, option_string=None): from ..log import LoggerHelper LoggerHelper().set_level(level=values)
# MIH: Disabled. Non-functional, untested. #class PBSAction(argparse.Action): # """Action to schedule actual command execution via PBS (e.g. Condor)""" # def __call__(self, parser, namespace, values, option_string=None): # pbs = values[0] # import pdb; pdb.set_trace() # i = 1
[docs]def parser_add_common_opt(parser, opt, names=None, **kwargs): from . import common_args opt_tmpl = getattr(common_args, opt) opt_kwargs = opt_tmpl[2].copy() opt_kwargs.update(kwargs) if names is None: parser.add_argument(*opt_tmpl[1], **opt_kwargs) else: parser.add_argument(*names, **opt_kwargs)
[docs]def parser_add_common_options(parser, version): parser_add_common_opt(parser, 'log_level') parser_add_common_opt(parser, 'pbs_runner') parser_add_common_opt(parser, 'change_path') parser_add_common_opt( parser, 'version', version=f'datalad {version}\n') if __debug__: parser.add_argument( '--dbg', action='store_true', dest='common_debug', help="enter Python debugger when uncaught exception happens") parser.add_argument( '--idbg', action='store_true', dest='common_idebug', help="enter IPython debugger when uncaught exception happens") parser.add_argument( '-c', action='append', dest='cfg_overrides', metavar='KEY=VALUE', help="""configuration variable setting. Overrides any configuration read from a file, but is potentially overridden itself by configuration variables in the process environment.""") # CLI analog of eval_params.result_renderer but with `<template>` handling # and a different default: in Python API we have None as default and do not render # the results but return them. In CLI we default to "default" renderer parser.add_argument( '-f', '--output-format', dest='common_output_format', default='default', type=ensure_unicode, metavar="{default,json,json_pp,tailored,'<template>'}", help="""select format for returned command results. 'default' give one line per result reporting action, status, path and an optional message; 'json' renders a JSON object with all properties for each result (one per line); 'json_pp' pretty-prints JSON spanning multiple lines; 'tailored' enables a command-specific rendering style that is typically tailored to human consumption (no result output otherwise), '<template>' reports any value(s) of any result properties in any format indicated by the template (e.g. '{path}'; compare with JSON output for all key-value choices). The template syntax follows the Python "format() language". It is possible to report individual dictionary values, e.g. '{metadata[name]}'. If a 2nd-level key contains a colon, e.g. 'music:Genre', ':' must be substituted by '#' in the template, like so: '{metadata[music#Genre]}'. [Default: '%(default)s']""") parser.add_argument( '--report-status', dest='common_report_status', choices=['success', 'failure', 'ok', 'notneeded', 'impossible', 'error'], help="""constrain command result report to records matching the given status. 'success' is a synonym for 'ok' OR 'notneeded', 'failure' stands for 'impossible' OR 'error'.""") parser.add_argument( '--report-type', dest='common_report_type', choices=['dataset', 'file'], action='append', help="""constrain command result report to records matching the given type. Can be given more than once to match multiple types.""") # CLI analog of eval_params.on_failure. TODO: dedup parser.add_argument( '--on-failure', dest='common_on_failure', default=eval_defaults['on_failure'], choices=['ignore', 'continue', 'stop'], help="""when an operation fails: 'ignore' and continue with remaining operations, the error is logged but does not lead to a non-zero exit code of the command; 'continue' works like 'ignore', but an error causes a non-zero exit code; 'stop' halts on first failure and yields non-zero exit code. A failure is any result with status 'impossible' or 'error'. [Default: '%(default)s']""") parser.add_argument( '--cmd', dest='_', action='store_true', help="""syntactical helper that can be used to end the list of global command line options before the subcommand label. Options taking an arbitrary number of arguments may require to be followed by a single --cmd in order to enable identification of the subcommand.""")
[docs]def strip_arg_from_argv(args, value, opt_names): """Strip an originally listed option (with its value) from the list cmdline args """ # Yarik doesn't know better if args is None: args = sys.argv # remove present pbs-runner option args_clean = [] skip = 0 for i, arg in enumerate(args): if skip: # we skip only one as instructed skip -= 1 continue if not (arg in opt_names and i < len(args) - 1 and args[i + 1] == value): args_clean.append(arg) else: # we need to skip this one and next one skip = 1 return args_clean
[docs]def run_via_pbs(args, pbs): assert(pbs in ('condor',)) # for now # TODO: RF to support multiple backends, parameters, etc, for now -- just condor, no options f = NamedTemporaryFile('w', prefix='datalad-%s-' % pbs, suffix='.submit', delete=False) try: pwd = getpwd() logs = f.name.replace('.submit', '.log') exe = args[0] # TODO: we might need better way to join them, escaping spaces etc. There must be a stock helper #exe_args = ' '.join(map(repr, args[1:])) if len(args) > 1 else '' exe_args = ' '.join(args[1:]) if len(args) > 1 else '' f.write("""\ Executable = %(exe)s Initialdir = %(pwd)s Output = %(logs)s Error = %(logs)s getenv = True arguments = %(exe_args)s queue """ % locals()) f.close() Runner().run(['condor_submit', f.name]) lgr.info("Scheduled execution via %s. Logs will be stored under %s", pbs, logs) finally: unlink(f.name)
# TODO: useful also outside of cmdline, move to support/ from os import curdir
[docs]def get_repo_instance(path=curdir, class_=None): """Returns an instance of appropriate datalad repository for path. Check whether a certain path is inside a known type of repository and returns an instance representing it. May also check for a certain type instead of detecting the type of repository. Parameters ---------- path: str path to check; default: current working directory class_: class if given, check whether path is inside a repository, that can be represented as an instance of the passed class. Raises ------ RuntimeError, in case cwd is not inside a known repository. """ from os.path import ismount, exists, normpath, isabs from datalad.support.exceptions import InvalidGitRepositoryError from ..utils import expandpath from ..support.gitrepo import GitRepo from ..support.annexrepo import AnnexRepo dir_ = expandpath(path) abspath_ = path if isabs(path) else dir_ if class_ is not None: if class_ == AnnexRepo: type_ = "annex" elif class_ == GitRepo: type_ = "git" else: raise RuntimeError("Unknown class %s." % str(class_)) while not ismount(dir_): # TODO: always correct termination? if exists(opj(dir_, '.git')): # found git dir if class_ is None: # detect repo type: try: return AnnexRepo(dir_, create=False) except RuntimeError as e: pass try: return GitRepo(dir_, create=False) except InvalidGitRepositoryError as e: raise RuntimeError("No datalad repository found in %s" % abspath_) else: try: return class_(dir_, create=False) except (RuntimeError, InvalidGitRepositoryError) as e: raise RuntimeError("No %s repository found in %s." % (type_, abspath_)) else: dir_ = normpath(opj(dir_, "..")) if class_ is not None: raise RuntimeError("No %s repository found in %s" % (type_, abspath_)) else: raise RuntimeError("No datalad repository found in %s" % abspath_)
from appdirs import AppDirs from os.path import join as opj dirs = AppDirs("datalad", "datalad.org")