| # |
| # 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}" -- {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}}"' |
| if not check_tmux_version('1.9'): |
| # `tmux new-session -c` was added in 1.9; |
| # older versions fail with that flag |
| self.command = 'tmux new -d -s {0} -n {0} "{{command}}"' |
| self.command = self.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('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: |
| pass |
| except: |
| bb.warn("Terminal %s is supported but did not start" % (terminal.name)) |
| # when we've run out of options |
| 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('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_version(desired): |
| vernum = check_terminal_version("tmux") |
| if vernum and LooseVersion(vernum) < desired: |
| return False |
| return vernum |
| |
| 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) |
| if not check_tmux_version('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 |