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])
 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 == 'win32' or sys.platform == 'cygwin'
 313
 314
 315 def expandpath(path):
 316     """Expand ~user/ and environment $variables"""
 317     path = os.path.expandvars(path)
 318     if path.startswith('~'):
 319         path = os.path.expanduser(path)
 320     return path
 321
 322
 323 class Group(object):
 324     """Operate on a collection of objects as a single unit"""
 325
 326     def __init__(self, *members):
 327         self._members = members
 328
 329     def __getattr__(self, name):
 330         """Return a function that relays calls to the group"""
 331
 332         def relay(*args, **kwargs):
 333             for member in self._members:
 334                 method = getattr(member, name)
 335                 method(*args, **kwargs)
 336
 337         setattr(self, name, relay)
 338         return relay
 339
 340
 341 class Proxy(object):
 342     """Wrap an object and override attributes"""
 343
 344     def __init__(self, obj, **overrides):
 345         self._obj = obj
 346         for k, v in overrides.items():
 347             setattr(self, k, v)
 348
 349     def __getattr__(self, name):
 350         return getattr(self._obj, name)
 351
 352
 353 def slice_fn(input_items, map_fn):
 354     """Slice input_items and call map_fn over every slice
 355
 356     This exists because of "errno: Argument list too long"
 357
 358     """
 359     # This comment appeared near the top of include/linux/binfmts.h
 360     # in the Linux source tree:
 361     #
 362     # /*
 363     #  * MAX_ARG_PAGES defines the number of pages allocated for arguments
 364     #  * and envelope for the new program. 32 should suffice, this gives
 365     #  * a maximum env+arg of 128kB w/4KB pages!
 366     #  */
 367     # #define MAX_ARG_PAGES 32
 368     #
 369     # 'size' is a heuristic to keep things highly performant by minimizing
 370     # the number of slices.  If we wanted it to run as few commands as
 371     # possible we could call "getconf ARG_MAX" and make a better guess,
 372     # but it's probably not worth the complexity (and the extra call to
 373     # getconf that we can't do on Windows anyways).
 374     #
 375     # In my testing, getconf ARG_MAX on Mac OS X Mountain Lion reported
 376     # 262144 and Debian/Linux-x86_64 reported 2097152.
 377     #
 378     # The hard-coded max_arg_len value is safely below both of these
 379     # real-world values.
 380
 381     # 4K pages x 32 MAX_ARG_PAGES
 382     max_arg_len = (32 * 4096) // 4  # allow plenty of space for the environment
 383     max_filename_len = 256
 384     size = max_arg_len // max_filename_len
 385
 386     status = 0
 387     outs = []
 388     errs = []
 389
 390     items = copy.copy(input_items)
 391     while items:
 392         stat, out, err = map_fn(items[:size])
 393         if stat < 0:
 394             status = min(stat, status)
 395         else:
 396             status = max(stat, status)
 397         outs.append(out)
 398         errs.append(err)
 399         items = items[size:]
 400
 401     return (status, '\n'.join(outs), '\n'.join(errs))
 402
 403
 404 class seq(object):
 405     def __init__(self, sequence):
 406         self.seq = sequence
 407
 408     def index(self, item, default=-1):
 409         try:
 410             idx = self.seq.index(item)
 411         except ValueError:
 412             idx = default
 413         return idx
 414
 415     def __getitem__(self, idx):
 416         return self.seq[idx]