cola/utils.py

   1 # Copyright (c) 2008 David Aguilar
   2 """This module provides miscellaneous utility functions."""
   3
   4 import errno
   5 import mimetypes
   6 import os
   7 import platform
   8 import random
   9 import re
  10 import shlex
  11 import subprocess
  12 import sys
  13 import time
  14
  15 from cola import git
  16 from cola import core
  17 from cola import resources
  18 from cola.compat import hashlib
  19 from cola.decorators import memoize
  20
  21 random.seed(hash(time.time()))
  22
  23
  24 KNOWN_FILE_MIME_TYPES = {
  25     'text':      'script.png',
  26     'image':     'image.png',
  27     'python':    'script.png',
  28     'ruby':      'script.png',
  29     'shell':     'script.png',
  30     'perl':      'script.png',
  31     'octet':     'binary.png',
  32 }
  33
  34 KNOWN_FILE_EXTENSION = {
  35     '.java':    'script.png',
  36     '.groovy':  'script.png',
  37     '.cpp':     'script.png',
  38     '.c':       'script.png',
  39     '.h':       'script.png',
  40     '.cxx':     'script.png',
  41 }
  42
  43
  44 def add_parents(path_entry_set):
  45     """Iterate over each item in the set and add its parent directories."""
  46     for path in list(path_entry_set):
  47         while '//' in path:
  48             path = path.replace('//', '/')
  49         if path not in path_entry_set:
  50             path_entry_set.add(path)
  51         if '/' in path:
  52             parent_dir = dirname(path)
  53             while parent_dir and parent_dir not in path_entry_set:
  54                 path_entry_set.add(parent_dir)
  55                 parent_dir = dirname(parent_dir)
  56     return path_entry_set
  57
  58
  59 def run_cmd(command):
  60     """
  61     Run arguments as a command and return output.
  62
  63     >>> run_cmd(["echo", "hello", "world"])
  64     'hello world'
  65
  66     """
  67     return git.Git.execute(command)
  68
  69
  70 def ident_file_type(filename):
  71     """Returns an icon based on the contents of filename."""
  72     if os.path.exists(filename):
  73         filemimetype = mimetypes.guess_type(filename)
  74         if filemimetype[0] != None:
  75             for filetype, iconname in KNOWN_FILE_MIME_TYPES.iteritems():
  76                 if filetype in filemimetype[0].lower():
  77                     return iconname
  78         filename = filename.lower()
  79         for fileext, iconname in KNOWN_FILE_EXTENSION.iteritems():
  80             if filename.endswith(fileext):
  81                 return iconname
  82         return 'generic.png'
  83     else:
  84         return 'removed.png'
  85     # Fallback for modified files of an unknown type
  86     return 'generic.png'
  87
  88
  89 def file_icon(filename):
  90     """
  91     Returns the full path to an icon file corresponding to
  92     filename"s contents.
  93     """
  94     return resources.icon(ident_file_type(filename))
  95
  96
  97 def fork(args):
  98     """Launch a command in the background."""
  99     encoded_args = [core.encode(arg) for arg in args]
 100     return subprocess.Popen(encoded_args).pid
 101
 102
 103 def sublist(a,b):
 104     """Subtracts list b from list a and returns the resulting list."""
 105     # conceptually, c = a - b
 106     c = []
 107     for item in a:
 108         if item not in b:
 109             c.append(item)
 110     return c
 111
 112
 113 __grep_cache = {}
 114 def grep(pattern, items, squash=True):
 115     """Greps a list for items that match a pattern and return a list of
 116     matching items.  If only one item matches, return just that item.
 117     """
 118     isdict = type(items) is dict
 119     if pattern in __grep_cache:
 120         regex = __grep_cache[pattern]
 121     else:
 122         regex = __grep_cache[pattern] = re.compile(pattern)
 123     matched = []
 124     matchdict = {}
 125     for item in items:
 126         match = regex.match(item)
 127         if not match:
 128             continue
 129         groups = match.groups()
 130         if not groups:
 131             subitems = match.group(0)
 132         else:
 133             if len(groups) == 1:
 134                 subitems = groups[0]
 135             else:
 136                 subitems = list(groups)
 137         if isdict:
 138             matchdict[item] = items[item]
 139         else:
 140             matched.append(subitems)
 141
 142     if isdict:
 143         return matchdict
 144     else:
 145         if squash and len(matched) == 1:
 146             return matched[0]
 147         else:
 148             return matched
 149
 150
 151 def basename(path):
 152     """
 153     An os.path.basename() implementation that always uses '/'
 154
 155     Avoid os.path.basename because git's output always
 156     uses '/' regardless of platform.
 157
 158     """
 159     return path.rsplit('/', 1)[-1]
 160
 161
 162 def dirname(path):
 163     """
 164     An os.path.dirname() implementation that always uses '/'
 165
 166     Avoid os.path.dirname because git's output always
 167     uses '/' regardless of platform.
 168
 169     """
 170     while '//' in path:
 171         path = path.replace('//', '/')
 172     path_dirname = path.rsplit('/', 1)[0]
 173     if path_dirname == path:
 174         return ''
 175     return path.rsplit('/', 1)[0]
 176
 177
 178 def slurp(path):
 179     """Slurps a filepath into a string."""
 180     fh = open(core.encode(path))
 181     slushy = core.read_nointr(fh)
 182     fh.close()
 183     return core.decode(slushy)
 184
 185
 186 def write(path, contents):
 187     """Writes a raw string to a file."""
 188     fh = open(core.encode(path), 'wb')
 189     core.write_nointr(fh, core.encode(contents))
 190     fh.close()
 191
 192
 193 def strip_prefix(prefix, string):
 194     """Return string, without the prefix. Blow up if string doesn't
 195     start with prefix."""
 196     assert string.startswith(prefix)
 197     return string[len(prefix):]
 198
 199 def sanitize(s):
 200     """Removes shell metacharacters from a string."""
 201     for c in """ \t!@#$%^&*()\\;,<>"'[]{}~|""":
 202         s = s.replace(c, '_')
 203     return s
 204
 205
 206 def shell_split(s):
 207     """Split string apart into utf-8 encoded words using shell syntax"""
 208     try:
 209         return shlex.split(core.encode(s))
 210     except ValueError:
 211         return [s]
 212
 213
 214 def shell_usplit(s):
 215     """Returns a unicode list instead of encoded strings"""
 216     return [core.decode(arg) for arg in shell_split(s)]
 217
 218
 219 def tmp_dir():
 220     # Allow TMPDIR/TMP with a fallback to /tmp
 221     return os.environ.get('TMP', os.environ.get('TMPDIR', '/tmp'))
 222
 223
 224 def tmp_file_pattern():
 225     return os.path.join(tmp_dir(), 'git-cola-%s-.*' % os.getpid())
 226
 227
 228 def tmp_filename(prefix):
 229     randstr = ''.join([chr(random.randint(ord('a'), ord('z'))) for i in range(7)])
 230     prefix = prefix.replace('/', '-').replace('\\', '-')
 231     basename = 'git-cola-%s-%s-%s' % (os.getpid(), randstr, prefix)
 232     return os.path.join(tmp_dir(), basename)
 233
 234
 235 def is_linux():
 236     """Is this a linux machine?"""
 237     while True:
 238         try:
 239             return platform.system() == 'Linux'
 240         except IOError, e:
 241             if e.errno == errno.EINTR:
 242                 continue
 243             raise e
 244
 245 def is_debian():
 246     """Is it debian?"""
 247     return os.path.exists('/usr/bin/apt-get')
 248
 249
 250 def is_darwin():
 251     """Return True on OSX."""
 252     while True:
 253         try:
 254             p = platform.platform()
 255             break
 256         except IOError, e:
 257             if e.errno == errno.EINTR:
 258                 continue
 259             raise e
 260     p = p.lower()
 261     return 'macintosh' in p or 'darwin' in p
 262
 263
 264 @memoize
 265 def is_win32():
 266     """Return True on win32"""
 267     return os.name in ('nt', 'dos')
 268
 269
 270 def win32_set_binary(fd):
 271     try:
 272         import msvcrt
 273     except ImportError:
 274         return
 275     # When run without console, pipes may expose invalid
 276     # fileno(), usually set to -1.
 277     if hasattr(fd, 'fileno') and fd.fileno() >= 0:
 278         msvcrt.setmode(fd.fileno(), os.O_BINARY)
 279
 280
 281 def posix_set_binary(fd):
 282     """POSIX file descriptors are always binary"""
 283     pass
 284
 285
 286 if is_win32():
 287     set_binary = win32_set_binary
 288 else:
 289     set_binary = posix_set_binary
 290
 291
 292 def checksum(path):
 293     """Return a cheap md5 hexdigest for a path."""
 294     md5 = hashlib.new('md5')
 295     md5.update(slurp(path))
 296     return md5.hexdigest()
 297
 298
 299 def error(msg, *args):
 300     """Print an error message to stderr."""
 301     print >> sys.stderr, "ERROR:", msg % args
 302
 303
 304 def warn(msg, *args):
 305     """Print a warning message to stderr."""
 306     print >> sys.stderr, "warning:", msg % args
 307
 308
 309 def die(msg, *args):
 310     """Print as error message to stderr and exit the program."""
 311     error(msg, *args)
 312     sys.exit(1)
 313
 314
 315 class ProgressIndicator(object):
 316
 317     """Simple progress indicator.
 318
 319     Displayed as a spinning character by default, but can be customized
 320     by passing custom messages that overrides the spinning character.
 321
 322     """
 323
 324     States = ("|", "/", "-", "\\")
 325
 326     def __init__(self, prefix="", f=sys.stdout):
 327         """Create a new ProgressIndicator, bound to the given file object."""
 328         self.n = 0  # Simple progress counter
 329         self.f = f  # Progress is written to this file object
 330         self.prev_len = 0  # Length of previous msg (to be overwritten)
 331         self.prefix = prefix  # Prefix prepended to each progress message
 332         self.prefix_lens = [] # Stack of prefix string lengths
 333
 334     def pushprefix(self, prefix):
 335         """Append the given prefix onto the prefix stack."""
 336         self.prefix_lens.append(len(self.prefix))
 337         self.prefix += prefix
 338
 339     def popprefix(self):
 340         """Remove the last prefix from the prefix stack."""
 341         prev_len = self.prefix_lens.pop()
 342         self.prefix = self.prefix[:prev_len]
 343
 344     def __call__(self, msg = None, lf=False):
 345         """Indicate progress, possibly with a custom message."""
 346         if msg is None:
 347             msg = self.States[self.n % len(self.States)]
 348         msg = self.prefix + msg
 349         print >> self.f, "\r%-*s" % (self.prev_len, msg),
 350         self.prev_len = len(msg.expandtabs())
 351         if lf:
 352             print >> self.f
 353             self.prev_len = 0
 354         self.n += 1
 355
 356     def finish (self, msg="done", noprefix=False):
 357         """Finalize progress indication with the given message."""
 358         if noprefix:
 359             self.prefix = ""
 360         self(msg, True)
 361
 362
 363 def start_command(args, cwd=None, shell=False, add_env=None,
 364                   stdin=subprocess.PIPE, stdout=subprocess.PIPE,
 365                   stderr=subprocess.PIPE):
 366     """Start the given command, and return a subprocess object.
 367
 368     This provides a simpler interface to the subprocess module.
 369
 370     """
 371     env = None
 372     if add_env is not None:
 373         env = os.environ.copy()
 374         env.update(add_env)
 375     return subprocess.Popen(args, bufsize=1, stdin=stdin, stdout=stdout,
 376                             stderr=stderr, cwd=cwd, shell=shell,
 377                             env=env, universal_newlines=True)
 378
 379
 380 def run_command(args, cwd=None, shell=False, add_env=None,
 381                 flag_error=True):
 382     """Run the given command to completion, and return its results.
 383
 384     This provides a simpler interface to the subprocess module.
 385
 386     The results are formatted as a 3-tuple: (exit_code, output, errors)
 387
 388     If flag_error is enabled, Error messages will be produced if the
 389     subprocess terminated with a non-zero exit code and/or stderr
 390     output.
 391
 392     The other arguments are passed on to start_command().
 393
 394     """
 395     process = start_command(args, cwd, shell, add_env)
 396     (output, errors) = process.communicate()
 397     exit_code = process.returncode
 398     if flag_error and errors:
 399         error("'%s' returned errors:\n---\n%s---", " ".join(args), errors)
 400     if flag_error and exit_code:
 401         error("'%s' returned exit code %i", " ".join(args), exit_code)
 402     return (exit_code, output, errors)