poky/meta/lib/oe/terminal.py

#
# Copyright OpenEmbedded Contributors
#
# SPDX-License-Identifier: GPL-2.0-only
#
import logging
import oe.classutils
import shlex
from bb.process import Popen, ExecutionError

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):
        from subprocess import STDOUT
        fmt_sh_cmd = self.format_command(sh_cmd, title)
        try:
            Popen.__init__(self, fmt_sh_cmd, env=env, stderr=STDOUT)
        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 bb.utils.vercmp_string_op(vernum, "2.0.0", "<"):
            # Konsole from KDE 3.x
            self.command = 'konsole -T "{title}" -e {command}'
        elif vernum and bb.utils.vercmp_string_op(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 bb.utils.vercmp_string_op(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