Mercurial > repos > rliterman > csp2

annotate CSP2/CSP2_env/env-d9b9114564458d9d-741b3de822f2aaca6c6caa4325c4afce/lib/python3.8/site-packages/tqdm/cli.py @ 68:5028fdace37b

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
planemo upload commit 2e9511a184a1ca667c7be0c6321a36dc4e3d116d
author jpayne
date Tue, 18 Mar 2025 16:23:26 -0400
parents
children
rev   line source
jpayne@68 1 """
jpayne@68 2 Module version for monitoring CLI pipes (`... | python -m tqdm | ...`).
jpayne@68 3 """
jpayne@68 4 import logging
jpayne@68 5 import re
jpayne@68 6 import sys
jpayne@68 7 from ast import literal_eval as numeric
jpayne@68 8 from textwrap import indent
jpayne@68 9
jpayne@68 10 from .std import TqdmKeyError, TqdmTypeError, tqdm
jpayne@68 11 from .version import __version__
jpayne@68 12
jpayne@68 13 __all__ = ["main"]
jpayne@68 14 log = logging.getLogger(__name__)
jpayne@68 15
jpayne@68 16
jpayne@68 17 def cast(val, typ):
jpayne@68 18 log.debug((val, typ))
jpayne@68 19 if " or " in typ:
jpayne@68 20 for t in typ.split(" or "):
jpayne@68 21 try:
jpayne@68 22 return cast(val, t)
jpayne@68 23 except TqdmTypeError:
jpayne@68 24 pass
jpayne@68 25 raise TqdmTypeError(f"{val} : {typ}")
jpayne@68 26
jpayne@68 27 # sys.stderr.write('\ndebug | `val:type`: `' + val + ':' + typ + '`.\n')
jpayne@68 28 if typ == 'bool':
jpayne@68 29 if (val == 'True') or (val == ''):
jpayne@68 30 return True
jpayne@68 31 if val == 'False':
jpayne@68 32 return False
jpayne@68 33 raise TqdmTypeError(val + ' : ' + typ)
jpayne@68 34 if typ == 'chr':
jpayne@68 35 if len(val) == 1:
jpayne@68 36 return val.encode()
jpayne@68 37 if re.match(r"^\\\w+$", val):
jpayne@68 38 return eval(f'"{val}"').encode()
jpayne@68 39 raise TqdmTypeError(f"{val} : {typ}")
jpayne@68 40 if typ == 'str':
jpayne@68 41 return val
jpayne@68 42 if typ == 'int':
jpayne@68 43 try:
jpayne@68 44 return int(val)
jpayne@68 45 except ValueError as exc:
jpayne@68 46 raise TqdmTypeError(f"{val} : {typ}") from exc
jpayne@68 47 if typ == 'float':
jpayne@68 48 try:
jpayne@68 49 return float(val)
jpayne@68 50 except ValueError as exc:
jpayne@68 51 raise TqdmTypeError(f"{val} : {typ}") from exc
jpayne@68 52 raise TqdmTypeError(f"{val} : {typ}")
jpayne@68 53
jpayne@68 54
jpayne@68 55 def posix_pipe(fin, fout, delim=b'\\n', buf_size=256,
jpayne@68 56 callback=lambda float: None, callback_len=True):
jpayne@68 57 """
jpayne@68 58 Params
jpayne@68 59 ------
jpayne@68 60 fin : binary file with `read(buf_size : int)` method
jpayne@68 61 fout : binary file with `write` (and optionally `flush`) methods.
jpayne@68 62 callback : function(float), e.g.: `tqdm.update`
jpayne@68 63 callback_len : If (default: True) do `callback(len(buffer))`.
jpayne@68 64 Otherwise, do `callback(data) for data in buffer.split(delim)`.
jpayne@68 65 """
jpayne@68 66 fp_write = fout.write
jpayne@68 67
jpayne@68 68 if not delim:
jpayne@68 69 while True:
jpayne@68 70 tmp = fin.read(buf_size)
jpayne@68 71
jpayne@68 72 # flush at EOF
jpayne@68 73 if not tmp:
jpayne@68 74 getattr(fout, 'flush', lambda: None)()
jpayne@68 75 return
jpayne@68 76
jpayne@68 77 fp_write(tmp)
jpayne@68 78 callback(len(tmp))
jpayne@68 79 # return
jpayne@68 80
jpayne@68 81 buf = b''
jpayne@68 82 len_delim = len(delim)
jpayne@68 83 # n = 0
jpayne@68 84 while True:
jpayne@68 85 tmp = fin.read(buf_size)
jpayne@68 86
jpayne@68 87 # flush at EOF
jpayne@68 88 if not tmp:
jpayne@68 89 if buf:
jpayne@68 90 fp_write(buf)
jpayne@68 91 if callback_len:
jpayne@68 92 # n += 1 + buf.count(delim)
jpayne@68 93 callback(1 + buf.count(delim))
jpayne@68 94 else:
jpayne@68 95 for i in buf.split(delim):
jpayne@68 96 callback(i)
jpayne@68 97 getattr(fout, 'flush', lambda: None)()
jpayne@68 98 return # n
jpayne@68 99
jpayne@68 100 while True:
jpayne@68 101 i = tmp.find(delim)
jpayne@68 102 if i < 0:
jpayne@68 103 buf += tmp
jpayne@68 104 break
jpayne@68 105 fp_write(buf + tmp[:i + len(delim)])
jpayne@68 106 # n += 1
jpayne@68 107 callback(1 if callback_len else (buf + tmp[:i]))
jpayne@68 108 buf = b''
jpayne@68 109 tmp = tmp[i + len_delim:]
jpayne@68 110
jpayne@68 111
jpayne@68 112 # ((opt, type), ... )
jpayne@68 113 RE_OPTS = re.compile(r'\n {4}(\S+)\s{2,}:\s*([^,]+)')
jpayne@68 114 # better split method assuming no positional args
jpayne@68 115 RE_SHLEX = re.compile(r'\s*(?<!\S)--?([^\s=]+)(\s+|=|$)')
jpayne@68 116
jpayne@68 117 # TODO: add custom support for some of the following?
jpayne@68 118 UNSUPPORTED_OPTS = ('iterable', 'gui', 'out', 'file')
jpayne@68 119
jpayne@68 120 # The 8 leading spaces are required for consistency
jpayne@68 121 CLI_EXTRA_DOC = r"""
jpayne@68 122 Extra CLI Options
jpayne@68 123 -----------------
jpayne@68 124 name : type, optional
jpayne@68 125 TODO: find out why this is needed.
jpayne@68 126 delim : chr, optional
jpayne@68 127 Delimiting character [default: '\n']. Use '\0' for null.
jpayne@68 128 N.B.: on Windows systems, Python converts '\n' to '\r\n'.
jpayne@68 129 buf_size : int, optional
jpayne@68 130 String buffer size in bytes [default: 256]
jpayne@68 131 used when `delim` is specified.
jpayne@68 132 bytes : bool, optional
jpayne@68 133 If true, will count bytes, ignore `delim`, and default
jpayne@68 134 `unit_scale` to True, `unit_divisor` to 1024, and `unit` to 'B'.
jpayne@68 135 tee : bool, optional
jpayne@68 136 If true, passes `stdin` to both `stderr` and `stdout`.
jpayne@68 137 update : bool, optional
jpayne@68 138 If true, will treat input as newly elapsed iterations,
jpayne@68 139 i.e. numbers to pass to `update()`. Note that this is slow
jpayne@68 140 (~2e5 it/s) since every input must be decoded as a number.
jpayne@68 141 update_to : bool, optional
jpayne@68 142 If true, will treat input as total elapsed iterations,
jpayne@68 143 i.e. numbers to assign to `self.n`. Note that this is slow
jpayne@68 144 (~2e5 it/s) since every input must be decoded as a number.
jpayne@68 145 null : bool, optional
jpayne@68 146 If true, will discard input (no stdout).
jpayne@68 147 manpath : str, optional
jpayne@68 148 Directory in which to install tqdm man pages.
jpayne@68 149 comppath : str, optional
jpayne@68 150 Directory in which to place tqdm completion.
jpayne@68 151 log : str, optional
jpayne@68 152 CRITICAL|FATAL|ERROR|WARN(ING)|[default: 'INFO']|DEBUG|NOTSET.
jpayne@68 153 """
jpayne@68 154
jpayne@68 155
jpayne@68 156 def main(fp=sys.stderr, argv=None):
jpayne@68 157 """
jpayne@68 158 Parameters (internal use only)
jpayne@68 159 ---------
jpayne@68 160 fp : file-like object for tqdm
jpayne@68 161 argv : list (default: sys.argv[1:])
jpayne@68 162 """
jpayne@68 163 if argv is None:
jpayne@68 164 argv = sys.argv[1:]
jpayne@68 165 try:
jpayne@68 166 log_idx = argv.index('--log')
jpayne@68 167 except ValueError:
jpayne@68 168 for i in argv:
jpayne@68 169 if i.startswith('--log='):
jpayne@68 170 logLevel = i[len('--log='):]
jpayne@68 171 break
jpayne@68 172 else:
jpayne@68 173 logLevel = 'INFO'
jpayne@68 174 else:
jpayne@68 175 # argv.pop(log_idx)
jpayne@68 176 # logLevel = argv.pop(log_idx)
jpayne@68 177 logLevel = argv[log_idx + 1]
jpayne@68 178 logging.basicConfig(level=getattr(logging, logLevel),
jpayne@68 179 format="%(levelname)s:%(module)s:%(lineno)d:%(message)s")
jpayne@68 180
jpayne@68 181 # py<3.13 doesn't dedent docstrings
jpayne@68 182 d = (tqdm.__doc__ if sys.version_info < (3, 13)
jpayne@68 183 else indent(tqdm.__doc__, " ")) + CLI_EXTRA_DOC
jpayne@68 184
jpayne@68 185 opt_types = dict(RE_OPTS.findall(d))
jpayne@68 186 # opt_types['delim'] = 'chr'
jpayne@68 187
jpayne@68 188 for o in UNSUPPORTED_OPTS:
jpayne@68 189 opt_types.pop(o)
jpayne@68 190
jpayne@68 191 log.debug(sorted(opt_types.items()))
jpayne@68 192
jpayne@68 193 # d = RE_OPTS.sub(r' --\1=<\1> : \2', d)
jpayne@68 194 split = RE_OPTS.split(d)
jpayne@68 195 opt_types_desc = zip(split[1::3], split[2::3], split[3::3])
jpayne@68 196 d = ''.join(('\n --{0} : {2}{3}' if otd[1] == 'bool' else
jpayne@68 197 '\n --{0}=<{1}> : {2}{3}').format(
jpayne@68 198 otd[0].replace('_', '-'), otd[0], *otd[1:])
jpayne@68 199 for otd in opt_types_desc if otd[0] not in UNSUPPORTED_OPTS)
jpayne@68 200
jpayne@68 201 help_short = "Usage:\n tqdm [--help | options]\n"
jpayne@68 202 d = help_short + """
jpayne@68 203 Options:
jpayne@68 204 -h, --help Print this help and exit.
jpayne@68 205 -v, --version Print version and exit.
jpayne@68 206 """ + d.strip('\n') + '\n'
jpayne@68 207
jpayne@68 208 # opts = docopt(d, version=__version__)
jpayne@68 209 if any(v in argv for v in ('-v', '--version')):
jpayne@68 210 sys.stdout.write(__version__ + '\n')
jpayne@68 211 sys.exit(0)
jpayne@68 212 elif any(v in argv for v in ('-h', '--help')):
jpayne@68 213 sys.stdout.write(d + '\n')
jpayne@68 214 sys.exit(0)
jpayne@68 215 elif argv and argv[0][:2] != '--':
jpayne@68 216 sys.stderr.write(f"Error:Unknown argument:{argv[0]}\n{help_short}")
jpayne@68 217
jpayne@68 218 argv = RE_SHLEX.split(' '.join(["tqdm"] + argv))
jpayne@68 219 opts = dict(zip(argv[1::3], argv[3::3]))
jpayne@68 220
jpayne@68 221 log.debug(opts)
jpayne@68 222 opts.pop('log', True)
jpayne@68 223
jpayne@68 224 tqdm_args = {'file': fp}
jpayne@68 225 try:
jpayne@68 226 for (o, v) in opts.items():
jpayne@68 227 o = o.replace('-', '_')
jpayne@68 228 try:
jpayne@68 229 tqdm_args[o] = cast(v, opt_types[o])
jpayne@68 230 except KeyError as e:
jpayne@68 231 raise TqdmKeyError(str(e))
jpayne@68 232 log.debug('args:' + str(tqdm_args))
jpayne@68 233
jpayne@68 234 delim_per_char = tqdm_args.pop('bytes', False)
jpayne@68 235 update = tqdm_args.pop('update', False)
jpayne@68 236 update_to = tqdm_args.pop('update_to', False)
jpayne@68 237 if sum((delim_per_char, update, update_to)) > 1:
jpayne@68 238 raise TqdmKeyError("Can only have one of --bytes --update --update_to")
jpayne@68 239 except Exception:
jpayne@68 240 fp.write("\nError:\n" + help_short)
jpayne@68 241 stdin, stdout_write = sys.stdin, sys.stdout.write
jpayne@68 242 for i in stdin:
jpayne@68 243 stdout_write(i)
jpayne@68 244 raise
jpayne@68 245 else:
jpayne@68 246 buf_size = tqdm_args.pop('buf_size', 256)
jpayne@68 247 delim = tqdm_args.pop('delim', b'\\n')
jpayne@68 248 tee = tqdm_args.pop('tee', False)
jpayne@68 249 manpath = tqdm_args.pop('manpath', None)
jpayne@68 250 comppath = tqdm_args.pop('comppath', None)
jpayne@68 251 if tqdm_args.pop('null', False):
jpayne@68 252 class stdout(object):
jpayne@68 253 @staticmethod
jpayne@68 254 def write(_):
jpayne@68 255 pass
jpayne@68 256 else:
jpayne@68 257 stdout = sys.stdout
jpayne@68 258 stdout = getattr(stdout, 'buffer', stdout)
jpayne@68 259 stdin = getattr(sys.stdin, 'buffer', sys.stdin)
jpayne@68 260 if manpath or comppath:
jpayne@68 261 try: # py<3.9
jpayne@68 262 import importlib_resources as resources
jpayne@68 263 except ImportError:
jpayne@68 264 from importlib import resources
jpayne@68 265 from pathlib import Path
jpayne@68 266
jpayne@68 267 def cp(name, dst):
jpayne@68 268 """copy resource `name` to `dst`"""
jpayne@68 269 fi = resources.files('tqdm') / name
jpayne@68 270 dst.write_bytes(fi.read_bytes())
jpayne@68 271 log.info("written:%s", dst)
jpayne@68 272 if manpath is not None:
jpayne@68 273 cp('tqdm.1', Path(manpath) / 'tqdm.1')
jpayne@68 274 if comppath is not None:
jpayne@68 275 cp('completion.sh', Path(comppath) / 'tqdm_completion.sh')
jpayne@68 276 sys.exit(0)
jpayne@68 277 if tee:
jpayne@68 278 stdout_write = stdout.write
jpayne@68 279 fp_write = getattr(fp, 'buffer', fp).write
jpayne@68 280
jpayne@68 281 class stdout(object): # pylint: disable=function-redefined
jpayne@68 282 @staticmethod
jpayne@68 283 def write(x):
jpayne@68 284 with tqdm.external_write_mode(file=fp):
jpayne@68 285 fp_write(x)
jpayne@68 286 stdout_write(x)
jpayne@68 287 if delim_per_char:
jpayne@68 288 tqdm_args.setdefault('unit', 'B')
jpayne@68 289 tqdm_args.setdefault('unit_scale', True)
jpayne@68 290 tqdm_args.setdefault('unit_divisor', 1024)
jpayne@68 291 log.debug(tqdm_args)
jpayne@68 292 with tqdm(**tqdm_args) as t:
jpayne@68 293 posix_pipe(stdin, stdout, '', buf_size, t.update)
jpayne@68 294 elif delim == b'\\n':
jpayne@68 295 log.debug(tqdm_args)
jpayne@68 296 write = stdout.write
jpayne@68 297 if update or update_to:
jpayne@68 298 with tqdm(**tqdm_args) as t:
jpayne@68 299 if update:
jpayne@68 300 def callback(i):
jpayne@68 301 t.update(numeric(i.decode()))
jpayne@68 302 else: # update_to
jpayne@68 303 def callback(i):
jpayne@68 304 t.update(numeric(i.decode()) - t.n)
jpayne@68 305 for i in stdin:
jpayne@68 306 write(i)
jpayne@68 307 callback(i)
jpayne@68 308 else:
jpayne@68 309 for i in tqdm(stdin, **tqdm_args):
jpayne@68 310 write(i)
jpayne@68 311 else:
jpayne@68 312 log.debug(tqdm_args)
jpayne@68 313 with tqdm(**tqdm_args) as t:
jpayne@68 314 callback_len = False
jpayne@68 315 if update:
jpayne@68 316 def callback(i):
jpayne@68 317 t.update(numeric(i.decode()))
jpayne@68 318 elif update_to:
jpayne@68 319 def callback(i):
jpayne@68 320 t.update(numeric(i.decode()) - t.n)
jpayne@68 321 else:
jpayne@68 322 callback = t.update
jpayne@68 323 callback_len = True
jpayne@68 324 posix_pipe(stdin, stdout, delim, buf_size, callback, callback_len)