from .std import tqdm, TqdmTypeError, TqdmKeyError from ._version import __version__ # NOQA import sys import re import logging __all__ = ["main"] def cast(val, typ): log = logging.getLogger(__name__) log.debug((val, typ)) if " or " in typ: for t in typ.split(" or "): try: return cast(val, t) except TqdmTypeError: pass raise TqdmTypeError(val + ' : ' + typ) # sys.stderr.write('\ndebug | `val:type`: `' + val + ':' + typ + '`.\n') if typ == 'bool': if (val == 'True') or (val == ''): return True elif val == 'False': return False else: raise TqdmTypeError(val + ' : ' + typ) try: return eval(typ + '("' + val + '")') except: if typ == 'chr': return chr(ord(eval('"' + val + '"'))) else: raise TqdmTypeError(val + ' : ' + typ) def posix_pipe(fin, fout, delim='\n', buf_size=256, callback=lambda int: None # pragma: no cover ): """ Params ------ fin : file with `read(buf_size : int)` method fout : file with `write` (and optionally `flush`) methods. callback : function(int), e.g.: `tqdm.update` """ fp_write = fout.write # tmp = '' if not delim: while True: tmp = fin.read(buf_size) # flush at EOF if not tmp: getattr(fout, 'flush', lambda: None)() # pragma: no cover return fp_write(tmp) callback(len(tmp)) # return buf = '' # n = 0 while True: tmp = fin.read(buf_size) # flush at EOF if not tmp: if buf: fp_write(buf) callback(1 + buf.count(delim)) # n += 1 + buf.count(delim) getattr(fout, 'flush', lambda: None)() # pragma: no cover return # n while True: try: i = tmp.index(delim) except ValueError: buf += tmp break else: fp_write(buf + tmp[:i + len(delim)]) callback(1) # n += 1 buf = '' tmp = tmp[i + len(delim):] # ((opt, type), ... ) RE_OPTS = re.compile(r'\n {8}(\S+)\s{2,}:\s*([^,]+)') # better split method assuming no positional args RE_SHLEX = re.compile(r'\s*(? : \2', d) split = RE_OPTS.split(d) opt_types_desc = zip(split[1::3], split[2::3], split[3::3]) d = ''.join('\n --{0}=<{0}> : {1}{2}'.format(*otd) for otd in opt_types_desc if otd[0] not in UNSUPPORTED_OPTS) d = """Usage: tqdm [--help | options] Options: -h, --help Print this help and exit -v, --version Print version and exit """ + d.strip('\n') + '\n' # opts = docopt(d, version=__version__) if any(v in argv for v in ('-v', '--version')): sys.stdout.write(__version__ + '\n') sys.exit(0) elif any(v in argv for v in ('-h', '--help')): sys.stdout.write(d + '\n') sys.exit(0) argv = RE_SHLEX.split(' '.join(["tqdm"] + argv)) opts = dict(zip(argv[1::3], argv[3::3])) log.debug(opts) opts.pop('log', True) tqdm_args = {'file': fp} try: for (o, v) in opts.items(): try: tqdm_args[o] = cast(v, opt_types[o]) except KeyError as e: raise TqdmKeyError(str(e)) log.debug('args:' + str(tqdm_args)) except: fp.write('\nError:\nUsage:\n tqdm [--help | options]\n') for i in sys.stdin: sys.stdout.write(i) raise else: buf_size = tqdm_args.pop('buf_size', 256) delim = tqdm_args.pop('delim', '\n') delim_per_char = tqdm_args.pop('bytes', False) manpath = tqdm_args.pop('manpath', None) comppath = tqdm_args.pop('comppath', None) stdin = getattr(sys.stdin, 'buffer', sys.stdin) stdout = getattr(sys.stdout, 'buffer', sys.stdout) if manpath or comppath: from os import path from shutil import copyfile from pkg_resources import resource_filename, Requirement def cp(src, dst): """copies from src path to dst""" copyfile(src, dst) log.info("written:" + dst) if manpath is not None: cp(resource_filename(Requirement.parse('tqdm'), 'tqdm/tqdm.1'), path.join(manpath, 'tqdm.1')) if comppath is not None: cp(resource_filename(Requirement.parse('tqdm'), 'tqdm/completion.sh'), path.join(comppath, 'tqdm_completion.sh')) sys.exit(0) if delim_per_char: tqdm_args.setdefault('unit', 'B') tqdm_args.setdefault('unit_scale', True) tqdm_args.setdefault('unit_divisor', 1024) log.debug(tqdm_args) with tqdm(**tqdm_args) as t: posix_pipe(stdin, stdout, '', buf_size, t.update) elif delim == '\n': log.debug(tqdm_args) for i in tqdm(stdin, **tqdm_args): stdout.write(i) else: log.debug(tqdm_args) with tqdm(**tqdm_args) as t: posix_pipe(stdin, stdout, delim, buf_size, t.update)