123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313 |
- #
- # SPDX-License-Identifier: GPL-2.0-only
- #
- import logging
- import oe.classutils
- import shlex
- from bb.process import Popen, ExecutionError
- from distutils.version import LooseVersion
- logger = logging.getLogger('BitBake.OE.Terminal')
- class UnsupportedTerminal(Exception):
- pass
- class NoSupportedTerminals(Exception):
- def __init__(self, terms):
- self.terms = terms
- class Registry(oe.classutils.ClassRegistry):
- command = None
- def __init__(cls, name, bases, attrs):
- super(Registry, cls).__init__(name.lower(), bases, attrs)
- @property
- def implemented(cls):
- return bool(cls.command)
- class Terminal(Popen, metaclass=Registry):
- def __init__(self, sh_cmd, title=None, env=None, d=None):
- fmt_sh_cmd = self.format_command(sh_cmd, title)
- try:
- Popen.__init__(self, fmt_sh_cmd, env=env)
- except OSError as exc:
- import errno
- if exc.errno == errno.ENOENT:
- raise UnsupportedTerminal(self.name)
- else:
- raise
- def format_command(self, sh_cmd, title):
- fmt = {'title': title or 'Terminal', 'command': sh_cmd, 'cwd': os.getcwd() }
- if isinstance(self.command, str):
- return shlex.split(self.command.format(**fmt))
- else:
- return [element.format(**fmt) for element in self.command]
- class XTerminal(Terminal):
- def __init__(self, sh_cmd, title=None, env=None, d=None):
- Terminal.__init__(self, sh_cmd, title, env, d)
- if not os.environ.get('DISPLAY'):
- raise UnsupportedTerminal(self.name)
- class Gnome(XTerminal):
- command = 'gnome-terminal -t "{title}" -x {command}'
- priority = 2
- def __init__(self, sh_cmd, title=None, env=None, d=None):
- # Recent versions of gnome-terminal does not support non-UTF8 charset:
- # https://bugzilla.gnome.org/show_bug.cgi?id=732127; as a workaround,
- # clearing the LC_ALL environment variable so it uses the locale.
- # Once fixed on the gnome-terminal project, this should be removed.
- if os.getenv('LC_ALL'): os.putenv('LC_ALL','')
- XTerminal.__init__(self, sh_cmd, title, env, d)
- class Mate(XTerminal):
- command = 'mate-terminal --disable-factory -t "{title}" -x {command}'
- priority = 2
- class Xfce(XTerminal):
- command = 'xfce4-terminal -T "{title}" -e "{command}"'
- priority = 2
- class Terminology(XTerminal):
- command = 'terminology -T="{title}" -e {command}'
- priority = 2
- class Konsole(XTerminal):
- command = 'konsole --separate --workdir . -p tabtitle="{title}" -e {command}'
- priority = 2
- def __init__(self, sh_cmd, title=None, env=None, d=None):
- # Check version
- vernum = check_terminal_version("konsole")
- if vernum and LooseVersion(vernum) < '2.0.0':
- # Konsole from KDE 3.x
- self.command = 'konsole -T "{title}" -e {command}'
- elif vernum and LooseVersion(vernum) < '16.08.1':
- # Konsole pre 16.08.01 Has nofork
- self.command = 'konsole --nofork --workdir . -p tabtitle="{title}" -e {command}'
- XTerminal.__init__(self, sh_cmd, title, env, d)
- class XTerm(XTerminal):
- command = 'xterm -T "{title}" -e {command}'
- priority = 1
- class Rxvt(XTerminal):
- command = 'rxvt -T "{title}" -e {command}'
- priority = 1
- class Screen(Terminal):
- command = 'screen -D -m -t "{title}" -S devshell {command}'
- def __init__(self, sh_cmd, title=None, env=None, d=None):
- s_id = "devshell_%i" % os.getpid()
- self.command = "screen -D -m -t \"{title}\" -S %s {command}" % s_id
- Terminal.__init__(self, sh_cmd, title, env, d)
- msg = 'Screen started. Please connect in another terminal with ' \
- '"screen -r %s"' % s_id
- if (d):
- bb.event.fire(bb.event.LogExecTTY(msg, "screen -r %s" % s_id,
- 0.5, 10), d)
- else:
- logger.warning(msg)
- class TmuxRunning(Terminal):
- """Open a new pane in the current running tmux window"""
- name = 'tmux-running'
- command = 'tmux split-window -c "{cwd}" "{command}"'
- priority = 2.75
- def __init__(self, sh_cmd, title=None, env=None, d=None):
- if not bb.utils.which(os.getenv('PATH'), 'tmux'):
- raise UnsupportedTerminal('tmux is not installed')
- if not os.getenv('TMUX'):
- raise UnsupportedTerminal('tmux is not running')
- if not check_tmux_pane_size('tmux'):
- raise UnsupportedTerminal('tmux pane too small or tmux < 1.9 version is being used')
- Terminal.__init__(self, sh_cmd, title, env, d)
- class TmuxNewWindow(Terminal):
- """Open a new window in the current running tmux session"""
- name = 'tmux-new-window'
- command = 'tmux new-window -c "{cwd}" -n "{title}" "{command}"'
- priority = 2.70
- def __init__(self, sh_cmd, title=None, env=None, d=None):
- if not bb.utils.which(os.getenv('PATH'), 'tmux'):
- raise UnsupportedTerminal('tmux is not installed')
- if not os.getenv('TMUX'):
- raise UnsupportedTerminal('tmux is not running')
- Terminal.__init__(self, sh_cmd, title, env, d)
- class Tmux(Terminal):
- """Start a new tmux session and window"""
- command = 'tmux new -c "{cwd}" -d -s devshell -n devshell "{command}"'
- priority = 0.75
- def __init__(self, sh_cmd, title=None, env=None, d=None):
- if not bb.utils.which(os.getenv('PATH'), 'tmux'):
- raise UnsupportedTerminal('tmux is not installed')
- # TODO: consider using a 'devshell' session shared amongst all
- # devshells, if it's already there, add a new window to it.
- window_name = 'devshell-%i' % os.getpid()
- self.command = 'tmux new -c "{{cwd}}" -d -s {0} -n {0} "{{command}}"'.format(window_name)
- Terminal.__init__(self, sh_cmd, title, env, d)
- attach_cmd = 'tmux att -t {0}'.format(window_name)
- msg = 'Tmux started. Please connect in another terminal with `tmux att -t {0}`'.format(window_name)
- if d:
- bb.event.fire(bb.event.LogExecTTY(msg, attach_cmd, 0.5, 10), d)
- else:
- logger.warning(msg)
- class Custom(Terminal):
- command = 'false' # This is a placeholder
- priority = 3
- def __init__(self, sh_cmd, title=None, env=None, d=None):
- self.command = d and d.getVar('OE_TERMINAL_CUSTOMCMD')
- if self.command:
- if not '{command}' in self.command:
- self.command += ' {command}'
- Terminal.__init__(self, sh_cmd, title, env, d)
- logger.warning('Custom terminal was started.')
- else:
- logger.debug(1, 'No custom terminal (OE_TERMINAL_CUSTOMCMD) set')
- raise UnsupportedTerminal('OE_TERMINAL_CUSTOMCMD not set')
- def prioritized():
- return Registry.prioritized()
- def get_cmd_list():
- terms = Registry.prioritized()
- cmds = []
- for term in terms:
- if term.command:
- cmds.append(term.command)
- return cmds
- def spawn_preferred(sh_cmd, title=None, env=None, d=None):
- """Spawn the first supported terminal, by priority"""
- for terminal in prioritized():
- try:
- spawn(terminal.name, sh_cmd, title, env, d)
- break
- except UnsupportedTerminal:
- continue
- else:
- raise NoSupportedTerminals(get_cmd_list())
- def spawn(name, sh_cmd, title=None, env=None, d=None):
- """Spawn the specified terminal, by name"""
- logger.debug(1, 'Attempting to spawn terminal "%s"', name)
- try:
- terminal = Registry.registry[name]
- except KeyError:
- raise UnsupportedTerminal(name)
- # We need to know when the command completes but some terminals (at least
- # gnome and tmux) gives us no way to do this. We therefore write the pid
- # to a file using a "phonehome" wrapper script, then monitor the pid
- # until it exits.
- import tempfile
- import time
- pidfile = tempfile.NamedTemporaryFile(delete = False).name
- try:
- sh_cmd = bb.utils.which(os.getenv('PATH'), "oe-gnome-terminal-phonehome") + " " + pidfile + " " + sh_cmd
- pipe = terminal(sh_cmd, title, env, d)
- output = pipe.communicate()[0]
- if output:
- output = output.decode("utf-8")
- if pipe.returncode != 0:
- raise ExecutionError(sh_cmd, pipe.returncode, output)
- while os.stat(pidfile).st_size <= 0:
- time.sleep(0.01)
- continue
- with open(pidfile, "r") as f:
- pid = int(f.readline())
- finally:
- os.unlink(pidfile)
- while True:
- try:
- os.kill(pid, 0)
- time.sleep(0.1)
- except OSError:
- return
- def check_tmux_pane_size(tmux):
- import subprocess as sub
- # On older tmux versions (<1.9), return false. The reason
- # is that there is no easy way to get the height of the active panel
- # on current window without nested formats (available from version 1.9)
- vernum = check_terminal_version("tmux")
- if vernum and LooseVersion(vernum) < '1.9':
- return False
- try:
- p = sub.Popen('%s list-panes -F "#{?pane_active,#{pane_height},}"' % tmux,
- shell=True,stdout=sub.PIPE,stderr=sub.PIPE)
- out, err = p.communicate()
- size = int(out.strip())
- except OSError as exc:
- import errno
- if exc.errno == errno.ENOENT:
- return None
- else:
- raise
- return size/2 >= 19
- def check_terminal_version(terminalName):
- import subprocess as sub
- try:
- cmdversion = '%s --version' % terminalName
- if terminalName.startswith('tmux'):
- cmdversion = '%s -V' % terminalName
- newenv = os.environ.copy()
- newenv["LANG"] = "C"
- p = sub.Popen(['sh', '-c', cmdversion], stdout=sub.PIPE, stderr=sub.PIPE, env=newenv)
- out, err = p.communicate()
- ver_info = out.decode().rstrip().split('\n')
- except OSError as exc:
- import errno
- if exc.errno == errno.ENOENT:
- return None
- else:
- raise
- vernum = None
- for ver in ver_info:
- if ver.startswith('Konsole'):
- vernum = ver.split(' ')[-1]
- if ver.startswith('GNOME Terminal'):
- vernum = ver.split(' ')[-1]
- if ver.startswith('MATE Terminal'):
- vernum = ver.split(' ')[-1]
- if ver.startswith('tmux'):
- vernum = ver.split()[-1]
- if ver.startswith('tmux next-'):
- vernum = ver.split()[-1][5:]
- return vernum
- def distro_name():
- try:
- p = Popen(['lsb_release', '-i'])
- out, err = p.communicate()
- distro = out.split(':')[1].strip().lower()
- except:
- distro = "unknown"
- return distro
|