mirrors/xonsh
1
0
Fork 0
mirror of https://github.com/xonsh/xonsh.git synced 2025-03-05 00:41:00 +01:00
Code Issues Projects Releases Packages Wiki Activity Actions

Merge pull request #1724 from laerus/gitstatus

Browse source 
vc_branch prompt
This commit is contained in:
Anthony Scopatz 2016-09-24 17:42:16 -04:00 • committed by GitHub
commit 75eada8412
4 changed files with 116 additions and 100 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

16
news/vc_branch.rst Normal file
View file

@ -0,0 +1,16 @@
**Added:** None
**Changed:**
* ``xonsh.prompt.gitstatus.gitstatus`` now returns a namedtuple
* implementation of ``xonsh.prompt.vc_branch.get_git_branch`` and
``xonsh.prompt.vc_branch.git_dirty_working_directory`` to be faster
**Deprecated:** None
**Removed:** None
**Fixed:** None
**Security:** None

34
xonsh/prompt/base.py
View file

@ -24,22 +24,24 @@ from xonsh.prompt.vc_branch import (
from xonsh.prompt.gitstatus import gitstatus_prompt
FORMATTER_DICT = xl.LazyObject(lambda: dict(
user=os.environ.get('USERNAME' if xp.ON_WINDOWS else 'USER', '<user>'),
prompt_end='#' if xt.is_superuser() else '$',
hostname=socket.gethostname().split('.', 1)[0],
cwd=_dynamically_collapsed_pwd,
cwd_dir=lambda: os.path.dirname(_replace_home_cwd()),
cwd_base=lambda: os.path.basename(_replace_home_cwd()),
short_cwd=_collapsed_pwd,
curr_branch=current_branch,
branch_color=branch_color,
branch_bg_color=branch_bg_color,
current_job=_current_job,
env_name=env_name,
vte_new_tab_cwd=vte_new_tab_cwd,
gitstatus=gitstatus_prompt,
), globals(), 'FORMATTER_DICT')
@xl.lazyobject
def FORMATTER_DICT():
return dict(
user=os.environ.get('USERNAME' if xp.ON_WINDOWS else 'USER', '<user>'),
prompt_end='#' if xt.is_superuser() else '$',
hostname=socket.gethostname().split('.', 1)[0],
cwd=_dynamically_collapsed_pwd,
cwd_dir=lambda: os.path.dirname(_replace_home_cwd()),
cwd_base=lambda: os.path.basename(_replace_home_cwd()),
short_cwd=_collapsed_pwd,
curr_branch=current_branch,
branch_color=branch_color,
branch_bg_color=branch_bg_color,
current_job=_current_job,
env_name=env_name,
vte_new_tab_cwd=vte_new_tab_cwd,
gitstatus=gitstatus_prompt,
)
@xl.lazyobject

57
xonsh/prompt/gitstatus.py
View file

@ -1,13 +1,19 @@
# -*- coding: utf-8 -*-
"""Informative git status prompt formatter"""
import os
import builtins
import collections
import os
import subprocess
import xonsh.lazyasd as xl
GitStatus = collections.namedtuple('GitStatus',
['branch', 'num_ahead', 'num_behind',
'untracked', 'changed', 'conflicts',
'staged', 'stashed', 'operations'])
def _check_output(*args, **kwargs):
kwargs.update(dict(env=builtins.__xonsh_env__.detype(),
stderr=subprocess.DEVNULL,
@ -70,9 +76,10 @@ def _gitoperation(gitdir):
def gitstatus():
"""Return (branch name, number of ahead commit, number of behind commit,
untracked number, changed number, conflicts number,
staged number, stashed number, operation)"""
"""Return namedtuple with fields:
branch name, number of ahead commit, number of behind commit,
untracked number, changed number, conflicts number,
staged number, stashed number, operation."""
status = _check_output(['git', 'status', '--porcelain', '--branch'])
branch = ''
num_ahead, num_behind = 0, 0
@ -111,7 +118,7 @@ def gitstatus():
stashed = _get_stash(gitdir)
operations = _gitoperation(gitdir)
return (branch, num_ahead, num_behind,
return GitStatus(branch, num_ahead, num_behind,
untracked, changed, conflicts, staged, stashed,
operations)
@ -119,31 +126,29 @@ def gitstatus():
def gitstatus_prompt():
"""Return str `BRANCH|OPERATOR|numbers`"""
try:
(branch, num_ahead, num_behind,
untracked, changed, conflicts, staged, stashed,
operations) = gitstatus()
s = gitstatus()
except subprocess.SubprocessError:
return None
ret = _get_def('BRANCH') + branch
if num_ahead > 0:
ret += _get_def('AHEAD') + str(num_ahead)
if num_behind > 0:
ret += _get_def('BEHIND') + str(num_behind)
if operations:
ret += _get_def('OPERATION') + '|' + '|'.join(operations)
ret = _get_def('BRANCH') + s.branch
if s.num_ahead > 0:
ret += _get_def('AHEAD') + str(s.num_ahead)
if s.num_behind > 0:
ret += _get_def('BEHIND') + str(s.num_behind)
if s.operations:
ret += _get_def('OPERATION') + '|' + '|'.join(s.operations)
ret += '|'
if staged > 0:
ret += _get_def('STAGED') + str(staged) + '{NO_COLOR}'
if conflicts > 0:
ret += _get_def('CONFLICTS') + str(conflicts) + '{NO_COLOR}'
if changed > 0:
ret += _get_def('CHANGED') + str(changed) + '{NO_COLOR}'
if untracked > 0:
ret += _get_def('UNTRACKED') + str(untracked) + '{NO_COLOR}'
if stashed > 0:
ret += _get_def('STASHED') + str(stashed) + '{NO_COLOR}'
if staged + conflicts + changed + untracked + stashed == 0:
if s.staged > 0:
ret += _get_def('STAGED') + str(s.staged) + '{NO_COLOR}'
if s.conflicts > 0:
ret += _get_def('CONFLICTS') + str(s.conflicts) + '{NO_COLOR}'
if s.changed > 0:
ret += _get_def('CHANGED') + str(s.changed) + '{NO_COLOR}'
if s.untracked > 0:
ret += _get_def('UNTRACKED') + str(s.untracked) + '{NO_COLOR}'
if s.stashed > 0:
ret += _get_def('STASHED') + str(s.stashed) + '{NO_COLOR}'
if s.staged + s.conflicts + s.changed + s.untracked + s.stashed == 0:
ret += _get_def('CLEAN') + '{NO_COLOR}'
ret += '{NO_COLOR}'

109
xonsh/prompt/vc_branch.py
View file

@ -9,49 +9,40 @@ import warnings
import subprocess
import xonsh.platform as xp
import xonsh.prompt
import threading, queue
def _get_git_branch(q):
try:
status = subprocess.check_output(['git', 'status'],
stderr=subprocess.DEVNULL)
except (subprocess.CalledProcessError, OSError):
q.put(None)
else:
status = status.decode().split()
if status[2] == 'at':
q.put(status[3])
else:
q.put(status[2])
def get_git_branch():
"""Attempts to find the current git branch. If no branch is found, then
an empty string is returned. If a timeout occured, the timeout exception
(subprocess.TimeoutExpired) is returned.
"""Attempts to find the current git branch. If this could not
be determined (timeout, not in a git repo, etc.) then this returns None.
"""
branch = None
env = builtins.__xonsh_env__
cwd = env['PWD']
denv = env.detype()
vcbt = env['VC_BRANCH_TIMEOUT']
if not xp.ON_WINDOWS:
prompt_scripts = ['/usr/lib/git-core/git-sh-prompt',
'/usr/local/etc/bash_completion.d/git-prompt.sh']
for script in prompt_scripts:
# note that this is about 10x faster than bash -i "__git_ps1"
inp = 'source {}; __git_ps1 "${{1:-%s}}"'.format(script)
try:
branch = subprocess.check_output(['bash'], cwd=cwd, input=inp,
stderr=subprocess.PIPE, timeout=vcbt, env=denv,
universal_newlines=True)
break
except subprocess.TimeoutExpired as e:
branch = e
break
except (subprocess.CalledProcessError, FileNotFoundError):
continue
# fall back to using the git binary if the above failed
if branch is None:
cmd = ['git', 'rev-parse', '--abbrev-ref', 'HEAD']
try:
s = subprocess.check_output(cmd, cwd=cwd, timeout=vcbt, env=denv,
stderr=subprocess.PIPE, universal_newlines=True)
if xp.ON_WINDOWS and len(s) == 0:
# Workaround for a bug in ConEMU/cmder, retry without redirection
s = subprocess.check_output(cmd, cwd=cwd, timeout=vcbt,
env=denv, universal_newlines=True)
branch = s.strip()
except subprocess.TimeoutExpired as e:
branch = e
except (subprocess.CalledProcessError, FileNotFoundError):
branch = None
timeout = builtins.__xonsh_env__.get('VC_BRANCH_TIMEOUT')
q = queue.Queue()
t = threading.Thread(target=_get_git_branch, args=(q,))
t.start()
t.join(timeout=timeout)
try:
branch = q.get_nowait()
except queue.Empty:
branch = None
return branch
@ -137,30 +128,32 @@ def current_branch(pad=NotImplemented):
return branch or None
def git_dirty_working_directory(cwd=None, include_untracked=False):
def _git_dirty_working_directory(q):
status = None
try:
status = subprocess.check_output(['git', 'status'],
stderr=subprocess.DEVNULL)
except (subprocess.CalledProcessError, OSError):
q.put(None)
if status is not None:
if b'nothing to commit' in status:
return q.put(False)
else:
return q.put(True)
def git_dirty_working_directory():
"""Returns whether or not the git directory is dirty. If this could not
be determined (timeout, file not sound, etc.) then this returns None.
"""
cmd = ['git', 'status', '--porcelain']
if include_untracked:
cmd.append('--untracked-files=normal')
else:
cmd.append('--untracked-files=no')
env = builtins.__xonsh_env__
cwd = env['PWD']
denv = env.detype()
vcbt = env['VC_BRANCH_TIMEOUT']
timeout = builtins.__xonsh_env__.get("VC_BRANCH_TIMEOUT")
q = queue.Queue()
t = threading.Thread(target=_git_dirty_working_directory, args=(q,))
t.start()
t.join(timeout=timeout)
try:
s = subprocess.check_output(cmd, stderr=subprocess.PIPE, cwd=cwd,
timeout=vcbt, universal_newlines=True,
env=denv)
if xp.ON_WINDOWS and len(s) == 0:
# Workaround for a bug in ConEMU/cmder, retry without redirection
s = subprocess.check_output(cmd, cwd=cwd, timeout=vcbt,
env=denv, universal_newlines=True)
return bool(s)
except (subprocess.CalledProcessError, subprocess.TimeoutExpired,
FileNotFoundError):
return q.get_nowait()
except queue.Empty:
return None