cola/utils.py

   1 """Miscellaneous utility functions"""
   2 from __future__ import absolute_import, division, print_function, unicode_literals
   3 import copy
   4 import os
   5 import random
   6 import re
   7 import shlex
   8 import sys
   9 import tempfile
  10 import time
  11 import traceback
  12
  13 from . import core
  14 from . import compat
  15
  16 random.seed(hash(time.time()))
  17
  18
  19 def asint(obj, default=0):
  20     """Make any value into an int, even if the cast fails"""
  21     try:
  22         value = int(obj)
  23     except (TypeError, ValueError):
  24         value = default
  25     return value
  26
  27
  28 def clamp(value, lo, hi):
  29     """Clamp a value to the specified range"""
  30     return min(hi, max(lo, value))
  31
  32
  33 def epoch_millis():
  34     return int(time.time() * 1000)
  35
  36
  37 def add_parents(paths):
  38     """Iterate over each item in the set and add its parent directories."""
  39     all_paths = set()
  40     for path in paths:
  41         while '//' in path:
  42             path = path.replace('//', '/')
  43         all_paths.add(path)
  44         if '/' in path:
  45             parent_dir = dirname(path)
  46             while parent_dir:
  47                 all_paths.add(parent_dir)
  48                 parent_dir = dirname(parent_dir)
  49     return all_paths
  50
  51
  52 def format_exception(e):
  53     exc_type, exc_value, exc_tb = sys.exc_info()
  54     details = traceback.format_exception(exc_type, exc_value, exc_tb)
  55     details = '\n'.join(map(core.decode, details))
  56     if hasattr(e, 'msg'):
  57         msg = e.msg
  58     else:
  59         msg = core.decode(repr(e))
  60     return (msg, details)
  61
  62
  63 def sublist(a, b):
  64     """Subtracts list b from list a and returns the resulting list."""
  65     # conceptually, c = a - b
  66     c = []
  67     for item in a:
  68         if item not in b:
  69             c.append(item)
  70     return c
  71
  72
  73 __grep_cache = {}
  74
  75
  76 def grep(pattern, items, squash=True):
  77     """Greps a list for items that match a pattern
  78
  79     :param squash: If only one item matches, return just that item
  80     :returns: List of matching items
  81
  82     """
  83     isdict = isinstance(items, dict)
  84     if pattern in __grep_cache:
  85         regex = __grep_cache[pattern]
  86     else:
  87         regex = __grep_cache[pattern] = re.compile(pattern)
  88     matched = []
  89     matchdict = {}
  90     for item in items:
  91         match = regex.match(item)
  92         if not match:
  93             continue
  94         groups = match.groups()
  95         if not groups:
  96             subitems = match.group(0)
  97         else:
  98             if len(groups) == 1:
  99                 subitems = groups[0]
 100             else:
 101                 subitems = list(groups)
 102         if isdict:
 103             matchdict[item] = items[item]
 104         else:
 105             matched.append(subitems)
 106
 107     if isdict:
 108         result = matchdict
 109     elif squash and len(matched) == 1:
 110         result = matched[0]
 111     else:
 112         result = matched
 113
 114     return result
 115
 116
 117 def basename(path):
 118     """
 119     An os.path.basename() implementation that always uses '/'
 120
 121     Avoid os.path.basename because git's output always
 122     uses '/' regardless of platform.
 123
 124     """
 125     return path.rsplit('/', 1)[-1]
 126
 127
 128 def strip_one(path):
 129     """Strip one level of directory"""
 130     return path.strip('/').split('/', 1)[-1]
 131
 132
 133 def dirname(path, current_dir=''):
 134     """
 135     An os.path.dirname() implementation that always uses '/'
 136
 137     Avoid os.path.dirname because git's output always
 138     uses '/' regardless of platform.
 139
 140     """
 141     while '//' in path:
 142         path = path.replace('//', '/')
 143     path_dirname = path.rsplit('/', 1)[0]
 144     if path_dirname == path:
 145         return current_dir
 146     return path.rsplit('/', 1)[0]
 147
 148
 149 def splitpath(path):
 150     """Split paths using '/' regardless of platform"""
 151     return path.split('/')
 152
 153
 154 def split(name):
 155     """Split a path-like name. Returns tuple "(head, tail)" where "tail" is
 156     everything after the final slash. The "head" may be empty.
 157
 158     This is the same as os.path.split() but only uses '/' as the delimiter.
 159
 160     >>> split('a/b/c')
 161     ('a/b', 'c')
 162
 163     >>> split('xyz')
 164     ('', 'xyz')
 165
 166     """
 167     return (dirname(name), basename(name))
 168
 169
 170 def join(*paths):
 171     """Join paths using '/' regardless of platform
 172
 173     >>> join('a', 'b', 'c')
 174     'a/b/c'
 175
 176     """
 177     return '/'.join(paths)
 178
 179
 180 def normalize_slash(value):
 181     """Strip and normalize slashes in a string
 182
 183     >>> normalize_slash('///Meow///Cat///')
 184     'Meow/Cat'
 185
 186     """
 187     value = value.strip('/')
 188     new_value = value.replace('//', '/')
 189     while new_value != value:
 190         value = new_value
 191         new_value = value.replace('//', '/')
 192     return value
 193
 194
 195 def pathjoin(paths):
 196     """Join a list of paths using '/' regardless of platform
 197
 198     >>> pathjoin(['a', 'b', 'c'])
 199     'a/b/c'
 200
 201     """
 202     return join(*paths)
 203
 204
 205 def pathset(path):
 206     """Return all of the path components for the specified path
 207
 208     >>> pathset('foo/bar/baz') == ['foo', 'foo/bar', 'foo/bar/baz']
 209     True
 210
 211     """
 212     result = []
 213     parts = splitpath(path)
 214     prefix = ''
 215     for part in parts:
 216         result.append(prefix + part)
 217         prefix += part + '/'
 218
 219     return result
 220
 221
 222 def select_directory(paths):
 223     """Return the first directory in a list of paths"""
 224     if not paths:
 225         return core.getcwd()
 226
 227     for path in paths:
 228         if core.isdir(path):
 229             return path
 230
 231     return os.path.dirname(paths[0]) or core.getcwd()
 232
 233
 234 def strip_prefix(prefix, string):
 235     """Return string, without the prefix. Blow up if string doesn't
 236     start with prefix."""
 237     assert string.startswith(prefix)
 238     return string[len(prefix) :]
 239
 240
 241 def sanitize(s):
 242     """Removes shell metacharacters from a string."""
 243     for c in """ \t!@#$%^&*()\\;,<>"'[]{}~|""":
 244         s = s.replace(c, '_')
 245     return s
 246
 247
 248 def tablength(word, tabwidth):
 249     """Return length of a word taking tabs into account
 250
 251     >>> tablength("\\t\\t\\t\\tX", 8)
 252     33
 253
 254     """
 255     return len(word.replace('\t', '')) + word.count('\t') * tabwidth
 256
 257
 258 def _shell_split_py2(s):
 259     """Python2 requires bytes inputs to shlex.split().  Returns [unicode]"""
 260     try:
 261         result = shlex.split(core.encode(s))
 262     except ValueError:
 263         result = core.encode(s).strip().split()
 264     # Decode to unicode strings
 265     return [core.decode(arg) for arg in result]
 266
 267
 268 def _shell_split_py3(s):
 269     """Python3 requires unicode inputs to shlex.split().  Converts to unicode"""
 270     try:
 271         result = shlex.split(s)
 272     except ValueError:
 273         result = core.decode(s).strip().split()
 274     # Already unicode
 275     return result
 276
 277
 278 def shell_split(s):
 279     if compat.PY2:
 280         # Encode before calling split()
 281         values = _shell_split_py2(s)
 282     else:
 283         # Python3 does not need the encode/decode dance
 284         values = _shell_split_py3(s)
 285     return values
 286
 287
 288 def tmp_filename(label, suffix=''):
 289     label = 'git-cola-' + label.replace('/', '-').replace('\\', '-')
 290     fd = tempfile.NamedTemporaryFile(prefix=label + '-', suffix=suffix, delete=False)
 291     fd.close()
 292     return fd.name
 293
 294
 295 def is_linux():
 296     """Is this a linux machine?"""
 297     return sys.platform.startswith('linux')
 298
 299
 300 def is_debian():
 301     """Is it debian?"""
 302     return os.path.exists('/usr/bin/apt-get')
 303
 304
 305 def is_darwin():
 306     """Return True on OSX."""
 307     return sys.platform == 'darwin'
 308
 309
 310 def is_win32():
 311     """Return True on win32"""
 312     return sys.platform in {'win32', 'cygwin'}
 313
 314
 315 def launch_default_app(paths):
 316     """Execute the default application on the specified paths"""
 317     if is_win32():
 318         for path in paths:
 319             if hasattr(os, 'startfile'):
 320                 os.startfile(path)  # pylint: disable=no-member
 321         return
 322
 323     if is_darwin():
 324         launcher = 'open'
 325     else:
 326         launcher = 'xdg-open'
 327
 328     core.fork([launcher] + paths)
 329
 330
 331 def expandpath(path):
 332     """Expand ~user/ and environment $variables"""
 333     path = os.path.expandvars(path)
 334     if path.startswith('~'):
 335         path = os.path.expanduser(path)
 336     return path
 337
 338
 339 class Group(object):
 340     """Operate on a collection of objects as a single unit"""
 341
 342     def __init__(self, *members):
 343         self._members = members
 344
 345     def __getattr__(self, name):
 346         """Return a function that relays calls to the group"""
 347
 348         def relay(*args, **kwargs):
 349             for member in self._members:
 350                 method = getattr(member, name)
 351                 method(*args, **kwargs)
 352
 353         setattr(self, name, relay)
 354         return relay
 355
 356
 357 class Proxy(object):
 358     """Wrap an object and override attributes"""
 359
 360     def __init__(self, obj, **overrides):
 361         self._obj = obj
 362         for k, v in overrides.items():
 363             setattr(self, k, v)
 364
 365     def __getattr__(self, name):
 366         return getattr(self._obj, name)
 367
 368
 369 def slice_fn(input_items, map_fn):
 370     """Slice input_items and call map_fn over every slice
 371
 372     This exists because of "errno: Argument list too long"
 373
 374     """
 375     # This comment appeared near the top of include/linux/binfmts.h
 376     # in the Linux source tree:
 377     #
 378     # /*
 379     #  * MAX_ARG_PAGES defines the number of pages allocated for arguments
 380     #  * and envelope for the new program. 32 should suffice, this gives
 381     #  * a maximum env+arg of 128kB w/4KB pages!
 382     #  */
 383     # #define MAX_ARG_PAGES 32
 384     #
 385     # 'size' is a heuristic to keep things highly performant by minimizing
 386     # the number of slices.  If we wanted it to run as few commands as
 387     # possible we could call "getconf ARG_MAX" and make a better guess,
 388     # but it's probably not worth the complexity (and the extra call to
 389     # getconf that we can't do on Windows anyways).
 390     #
 391     # In my testing, getconf ARG_MAX on Mac OS X Mountain Lion reported
 392     # 262144 and Debian/Linux-x86_64 reported 2097152.
 393     #
 394     # The hard-coded max_arg_len value is safely below both of these
 395     # real-world values.
 396
 397     # 4K pages x 32 MAX_ARG_PAGES
 398     max_arg_len = (32 * 4096) // 4  # allow plenty of space for the environment
 399     max_filename_len = 256
 400     size = max_arg_len // max_filename_len
 401
 402     status = 0
 403     outs = []
 404     errs = []
 405
 406     items = copy.copy(input_items)
 407     while items:
 408         stat, out, err = map_fn(items[:size])
 409         if stat < 0:
 410             status = min(stat, status)
 411         else:
 412             status = max(stat, status)
 413         outs.append(out)
 414         errs.append(err)
 415         items = items[size:]
 416
 417     return (status, '\n'.join(outs), '\n'.join(errs))
 418
 419
 420 class seq(object):
 421     def __init__(self, sequence):
 422         self.seq = sequence
 423
 424     def index(self, item, default=-1):
 425         try:
 426             idx = self.seq.index(item)
 427         except ValueError:
 428             idx = default
 429         return idx
 430
 431     def __getitem__(self, idx):
 432         return self.seq[idx]
 433
 434
 435 def catch_runtime_error(func, *args, **kwargs):
 436     """Run the function safely.
 437
 438     Catch RuntimeError to avoid tracebacks during application shutdown.
 439
 440     """
 441     # Signals and callbacks can sometimes get triggered during application shutdown.
 442     # This can happen when exiting while background tasks are still processing.
 443     # Guard against this by making this operation a no-op.
 444     try:
 445         valid = True
 446         result = func(*args, **kwargs)
 447     except RuntimeError:
 448         valid = False
 449         result = None
 450     return (valid, result)