pylint: remove bad-option-value from the configuration
[git-cola.git] / cola / utils.py
blob0ea6b31aab96e79d251c3a8436938fed95115661
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
13 from . import core
14 from . import compat
16 random.seed(hash(time.time()))
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
28 def clamp(value, lo, hi):
29 """Clamp a value to the specified range"""
30 return min(hi, max(lo, value))
33 def epoch_millis():
34 return int(time.time() * 1000)
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
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)
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
73 __grep_cache = {}
76 def grep(pattern, items, squash=True):
77 """Greps a list for items that match a pattern
79 :param squash: If only one item matches, return just that item
80 :returns: List of matching items
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)
107 if isdict:
108 result = matchdict
109 elif squash and len(matched) == 1:
110 result = matched[0]
111 else:
112 result = matched
114 return result
117 def basename(path):
119 An os.path.basename() implementation that always uses '/'
121 Avoid os.path.basename because git's output always
122 uses '/' regardless of platform.
125 return path.rsplit('/', 1)[-1]
128 def strip_one(path):
129 """Strip one level of directory"""
130 return path.strip('/').split('/', 1)[-1]
133 def dirname(path, current_dir=''):
135 An os.path.dirname() implementation that always uses '/'
137 Avoid os.path.dirname because git's output always
138 uses '/' regardless of platform.
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]
149 def splitpath(path):
150 """Split paths using '/' regardless of platform"""
151 return path.split('/')
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.
158 This is the same as os.path.split() but only uses '/' as the delimiter.
160 >>> split('a/b/c')
161 ('a/b', 'c')
163 >>> split('xyz')
164 ('', 'xyz')
167 return (dirname(name), basename(name))
170 def join(*paths):
171 """Join paths using '/' regardless of platform
173 >>> join('a', 'b', 'c')
174 'a/b/c'
177 return '/'.join(paths)
180 def normalize_slash(value):
181 """Strip and normalize slashes in a string
183 >>> normalize_slash('///Meow///Cat///')
184 'Meow/Cat'
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
195 def pathjoin(paths):
196 """Join a list of paths using '/' regardless of platform
198 >>> pathjoin(['a', 'b', 'c'])
199 'a/b/c'
202 return join(*paths)
205 def pathset(path):
206 """Return all of the path components for the specified path
208 >>> pathset('foo/bar/baz') == ['foo', 'foo/bar', 'foo/bar/baz']
209 True
212 result = []
213 parts = splitpath(path)
214 prefix = ''
215 for part in parts:
216 result.append(prefix + part)
217 prefix += part + '/'
219 return result
222 def select_directory(paths):
223 """Return the first directory in a list of paths"""
224 if not paths:
225 return core.getcwd()
227 for path in paths:
228 if core.isdir(path):
229 return path
231 return os.path.dirname(paths[0]) or core.getcwd()
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) :]
241 def sanitize(s):
242 """Removes shell metacharacters from a string."""
243 for c in """ \t!@#$%^&*()\\;,<>"'[]{}~|""":
244 s = s.replace(c, '_')
245 return s
248 def tablength(word, tabwidth):
249 """Return length of a word taking tabs into account
251 >>> tablength("\\t\\t\\t\\tX", 8)
255 return len(word.replace('\t', '')) + word.count('\t') * tabwidth
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]
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
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
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
295 def is_linux():
296 """Is this a linux machine?"""
297 return sys.platform.startswith('linux')
300 def is_debian():
301 """Is it debian?"""
302 return os.path.exists('/usr/bin/apt-get')
305 def is_darwin():
306 """Return True on OSX."""
307 return sys.platform == 'darwin'
310 def is_win32():
311 """Return True on win32"""
312 return sys.platform in {'win32', 'cygwin'}
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
323 if is_darwin():
324 launcher = 'open'
325 else:
326 launcher = 'xdg-open'
328 core.fork([launcher] + paths)
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
339 class Group(object):
340 """Operate on a collection of objects as a single unit"""
342 def __init__(self, *members):
343 self._members = members
345 def __getattr__(self, name):
346 """Return a function that relays calls to the group"""
348 def relay(*args, **kwargs):
349 for member in self._members:
350 method = getattr(member, name)
351 method(*args, **kwargs)
353 setattr(self, name, relay)
354 return relay
357 class Proxy(object):
358 """Wrap an object and override attributes"""
360 def __init__(self, obj, **overrides):
361 self._obj = obj
362 for k, v in overrides.items():
363 setattr(self, k, v)
365 def __getattr__(self, name):
366 return getattr(self._obj, name)
369 def slice_fn(input_items, map_fn):
370 """Slice input_items and call map_fn over every slice
372 This exists because of "errno: Argument list too long"
375 # This comment appeared near the top of include/linux/binfmts.h
376 # in the Linux source tree:
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
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).
391 # In my testing, getconf ARG_MAX on Mac OS X Mountain Lion reported
392 # 262144 and Debian/Linux-x86_64 reported 2097152.
394 # The hard-coded max_arg_len value is safely below both of these
395 # real-world values.
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
402 status = 0
403 outs = []
404 errs = []
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:]
417 return (status, '\n'.join(outs), '\n'.join(errs))
420 class seq(object):
421 def __init__(self, sequence):
422 self.seq = sequence
424 def index(self, item, default=-1):
425 try:
426 idx = self.seq.index(item)
427 except ValueError:
428 idx = default
429 return idx
431 def __getitem__(self, idx):
432 return self.seq[idx]
435 def catch_runtime_error(func, *args, **kwargs):
436 """Run the function safely.
438 Catch RuntimeError to avoid tracebacks during application shutdown.
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)