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 join(*paths):
 155     """Join paths using '/' regardless of platform"""
 156     return '/'.join(paths)
 157
 158
 159 def pathset(path):
 160     """Return all of the path components for the specified path
 161
 162     >>> pathset('foo/bar/baz') == ['foo', 'foo/bar', 'foo/bar/baz']
 163     True
 164
 165     """
 166     result = []
 167     parts = splitpath(path)
 168     prefix = ''
 169     for part in parts:
 170         result.append(prefix + part)
 171         prefix += part + '/'
 172
 173     return result
 174
 175
 176 def select_directory(paths):
 177     """Return the first directory in a list of paths"""
 178     if not paths:
 179         return core.getcwd()
 180
 181     for path in paths:
 182         if core.isdir(path):
 183             return path
 184
 185     return os.path.dirname(paths[0])
 186
 187
 188 def strip_prefix(prefix, string):
 189     """Return string, without the prefix. Blow up if string doesn't
 190     start with prefix."""
 191     assert string.startswith(prefix)
 192     return string[len(prefix) :]
 193
 194
 195 def sanitize(s):
 196     """Removes shell metacharacters from a string."""
 197     for c in """ \t!@#$%^&*()\\;,<>"'[]{}~|""":
 198         s = s.replace(c, '_')
 199     return s
 200
 201
 202 def tablength(word, tabwidth):
 203     """Return length of a word taking tabs into account
 204
 205     >>> tablength("\\t\\t\\t\\tX", 8)
 206     33
 207
 208     """
 209     return len(word.replace('\t', '')) + word.count('\t') * tabwidth
 210
 211
 212 def _shell_split_py2(s):
 213     """Python2 requires bytes inputs to shlex.split().  Returns [unicode]"""
 214     try:
 215         result = shlex.split(core.encode(s))
 216     except ValueError:
 217         result = core.encode(s).strip().split()
 218     # Decode to unicode strings
 219     return [core.decode(arg) for arg in result]
 220
 221
 222 def _shell_split_py3(s):
 223     """Python3 requires unicode inputs to shlex.split().  Converts to unicode"""
 224     try:
 225         result = shlex.split(s)
 226     except ValueError:
 227         result = core.decode(s).strip().split()
 228     # Already unicode
 229     return result
 230
 231
 232 def shell_split(s):
 233     if compat.PY2:
 234         # Encode before calling split()
 235         values = _shell_split_py2(s)
 236     else:
 237         # Python3 does not need the encode/decode dance
 238         values = _shell_split_py3(s)
 239     return values
 240
 241
 242 def tmp_filename(label, suffix=''):
 243     label = 'git-cola-' + label.replace('/', '-').replace('\\', '-')
 244     fd = tempfile.NamedTemporaryFile(prefix=label + '-', suffix=suffix)
 245     fd.close()
 246     return fd.name
 247
 248
 249 def is_linux():
 250     """Is this a linux machine?"""
 251     return sys.platform.startswith('linux')
 252
 253
 254 def is_debian():
 255     """Is it debian?"""
 256     return os.path.exists('/usr/bin/apt-get')
 257
 258
 259 def is_darwin():
 260     """Return True on OSX."""
 261     return sys.platform == 'darwin'
 262
 263
 264 def is_win32():
 265     """Return True on win32"""
 266     return sys.platform == 'win32' or sys.platform == 'cygwin'
 267
 268
 269 def expandpath(path):
 270     """Expand ~user/ and environment $variables"""
 271     path = os.path.expandvars(path)
 272     if path.startswith('~'):
 273         path = os.path.expanduser(path)
 274     return path
 275
 276
 277 class Group(object):
 278     """Operate on a collection of objects as a single unit"""
 279
 280     def __init__(self, *members):
 281         self._members = members
 282
 283     def __getattr__(self, name):
 284         """Return a function that relays calls to the group"""
 285
 286         def relay(*args, **kwargs):
 287             for member in self._members:
 288                 method = getattr(member, name)
 289                 method(*args, **kwargs)
 290
 291         setattr(self, name, relay)
 292         return relay
 293
 294
 295 class Proxy(object):
 296     """Wrap an object and override attributes"""
 297
 298     def __init__(self, obj, **overrides):
 299         self._obj = obj
 300         for k, v in overrides.items():
 301             setattr(self, k, v)
 302
 303     def __getattr__(self, name):
 304         return getattr(self._obj, name)
 305
 306
 307 def slice_fn(input_items, map_fn):
 308     """Slice input_items and call map_fn over every slice
 309
 310     This exists because of "errno: Argument list too long"
 311
 312     """
 313     # This comment appeared near the top of include/linux/binfmts.h
 314     # in the Linux source tree:
 315     #
 316     # /*
 317     #  * MAX_ARG_PAGES defines the number of pages allocated for arguments
 318     #  * and envelope for the new program. 32 should suffice, this gives
 319     #  * a maximum env+arg of 128kB w/4KB pages!
 320     #  */
 321     # #define MAX_ARG_PAGES 32
 322     #
 323     # 'size' is a heuristic to keep things highly performant by minimizing
 324     # the number of slices.  If we wanted it to run as few commands as
 325     # possible we could call "getconf ARG_MAX" and make a better guess,
 326     # but it's probably not worth the complexity (and the extra call to
 327     # getconf that we can't do on Windows anyways).
 328     #
 329     # In my testing, getconf ARG_MAX on Mac OS X Mountain Lion reported
 330     # 262144 and Debian/Linux-x86_64 reported 2097152.
 331     #
 332     # The hard-coded max_arg_len value is safely below both of these
 333     # real-world values.
 334
 335     # 4K pages x 32 MAX_ARG_PAGES
 336     max_arg_len = (32 * 4096) // 4  # allow plenty of space for the environment
 337     max_filename_len = 256
 338     size = max_arg_len // max_filename_len
 339
 340     status = 0
 341     outs = []
 342     errs = []
 343
 344     items = copy.copy(input_items)
 345     while items:
 346         stat, out, err = map_fn(items[:size])
 347         if stat < 0:
 348             status = min(stat, status)
 349         else:
 350             status = max(stat, status)
 351         outs.append(out)
 352         errs.append(err)
 353         items = items[size:]
 354
 355     return (status, '\n'.join(outs), '\n'.join(errs))
 356
 357
 358 class seq(object):
 359     def __init__(self, sequence):
 360         self.seq = sequence
 361
 362     def index(self, item, default=-1):
 363         try:
 364             idx = self.seq.index(item)
 365         except ValueError:
 366             idx = default
 367         return idx
 368
 369     def __getitem__(self, idx):
 370         return self.seq[idx]