Source code for core.terminal

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""Handles terminal detection on Linux and terminal command lines."""

import os
import shlex
import shutil
import subprocess
import sys
import tempfile
import time

from . import log
from .lnp import lnp


def get_terminal_command(cmd, force_custom=False):
    """
    Returns a command to launch <cmd> in a new terminal.
    On Linux, if force_custom is set, the custom terminal command will be used.
    """
    log.d("Preparing terminal command for command line %s", cmd)
    if not isinstance(cmd, list):
        cmd = [cmd, ]
    if sys.platform == 'darwin':
        return ['open', '-a', 'Terminal.app'] + cmd
    if sys.platform.startswith('linux'):
        if force_custom:
            term = CustomTerminal.get_command_line()
            log.d("Using custom terminal: %s", term)
        else:
            term = get_configured_terminal().get_command_line()
            log.d(
                "Using configured terminal: %s, command line %s", term,
                get_configured_terminal().name)
        if not term:
            raise Exception(
                'No terminal configured! Use File > Configure Terminal.')
        if "$" in term:
            c = []
            for s in term:
                if s == '$':
                    c += cmd
                else:
                    c.append(s)
            return c
        return term + cmd
    raise Exception('No terminal launcher for platform: ' + sys.platform)


def get_configured_terminal():
    """Retrieves the configured terminal command."""
    s = lnp.userconfig.get_string('terminal_type')
    terminals = get_valid_terminals()
    for t in terminals:
        if s == t.name:
            return t
    return CustomTerminal


def terminal_configured():
    """Returns True if a terminal has been set up."""
    return lnp.userconfig.get('terminal_type') is not None


def get_custom_terminal_cmd():
    """Returns the command used by the custom terminal."""
    return lnp.userconfig.get_string('terminal')


def get_valid_terminals():
    """Gets the terminals that are available on this system."""
    result = []
    terminals = _get_terminals()
    for t in terminals:
        log.d("Checking for terminal %s", t.name)
        if t.detect():
            log.d("Found terminal %s", t.name)
            result.append(t)
    return result


def _get_terminals():
    return LinuxTerminal.__subclasses__()


def configure_terminal(termname):
    """Configures the terminal class used to launch a terminal on Linux."""
    lnp.userconfig['terminal_type'] = termname
    lnp.userconfig.save_data()


def configure_custom_terminal(new_path):
    """Configures the custom command used to launch a terminal on Linux."""
    lnp.userconfig['terminal'] = new_path
    lnp.userconfig.save_data()


class LinuxTerminal(object):
    """
    Class for detecting and launching using a dedicated terminal on Linux.
    """

    # Set this in subclasses to provide a label for the terminal.
    name = "????"

    @staticmethod
    def detect():
        """Detects if this terminal is available."""

    @staticmethod
    def get_command_line():
        """
        Returns a subprocess-compatible command to launch a command with
        this terminal.
        If the command to be launched should go somewhere other than the end
        of the command line, use $ to indicate the correct place.
        """


# Desktop environment-specific terminals
class KDETerminal(LinuxTerminal):
    """Handles terminals on KDE (e.g. Konsole)."""
    name = "KDE"

    @staticmethod
    def detect():
        return os.environ.get('KDE_FULL_SESSION', '') == 'true'

    @staticmethod
    def get_command_line():
        kreadconfig_path = shutil.which('kreadconfig5') or shutil.which('kreadconfig')
        s = subprocess.check_output([
            kreadconfig_path, '--file', 'kdeglobals', '--group', 'General',
            '--key', 'TerminalApplication', '--default', 'konsole'],
            universal_newlines=True).replace('\n', '')
        return ['nohup', s, '-e']


class GNOMETerminal(LinuxTerminal):
    """Handles terminals on GNOME and Cinnamon (e.g. gnome-terminal)."""
    name = "GNOME/Cinnamon"

    @staticmethod
    def detect():
        if os.environ.get('GNOME_DESKTOP_SESSION_ID', ''):
            return True
        if os.environ.get('CINNAMON_VERSION', ''):
            return True
        with open(os.devnull, 'w', encoding="utf-8") as FNULL:
            try:
                return subprocess.call(
                    [
                        'dbus-send', '--print-reply',
                        '--dest=org.freedesktop.DBus',
                        '/org/freedesktop/DBus',
                        'org.freedesktop.DBus.GetNameOwner',
                        'string:org.gnome.SessionManager'
                    ], stdout=FNULL, stderr=FNULL) == 0
            except Exception:
                return False
            try:
                GNOMETerminal.get_command_line()  # Attempt to get the command line
                return True
            except Exception:
                return False

    @staticmethod
    def get_command_line():
        try:  # Try gsettings first (e.g. Ubuntu 17.04)
            term = subprocess.check_output([
                'gsettings', 'get',
                'org.gnome.desktop.default-applications.terminal', 'exec'
            ], universal_newlines=True).replace('\n', '').replace("'", '')
            term_arg = subprocess.check_output([
                'gsettings', 'get',
                'org.gnome.desktop.default-applications.terminal', 'exec-arg'
            ], universal_newlines=True).replace('\n', '').replace("'", '')
            return ['nohup', term, term_arg]
        except Exception:  # fallback to older gconf
            pass
        try:
            term = subprocess.check_output([
                'gconftool-2', '--get',
                '/desktop/gnome/applications/terminal/exec'
            ], universal_newlines=True).replace('\n', '')
            term_arg = subprocess.check_output([
                'gconftool-2', '--get',
                '/desktop/gnome/applications/terminal/exec_arg'
            ], universal_newlines=True).replace('\n', '')
            return ['nohup', term, term_arg]
        except Exception as exc:
            raise Exception("Unable to determine terminal command.") from exc


class XfceTerminal(LinuxTerminal):
    """Handles terminals in the Xfce desktop environment."""
    name = "Xfce"

    @staticmethod
    def detect():
        try:
            s = subprocess.check_output(
                ['ps', '-eo', 'comm='], stderr=subprocess.STDOUT,
                universal_newlines=True)
            return 'xfce' in s
        except Exception:
            return False

    @staticmethod
    def get_command_line():
        return ['nohup', 'exo-open', '--launch', 'TerminalEmulator']


class LXDETerminal(LinuxTerminal):
    """Handles terminals in LXDE."""
    name = "LXDE"

    @staticmethod
    def detect():
        if os.environ.get('DESKTOP_SESSION', '') != 'LXDE':
            return False
        with open(os.devnull, 'w', encoding="utf-8") as FNULL:
            try:
                return subprocess.call(
                    ['which', 'lxterminal'], stdout=FNULL, stderr=FNULL,
                    close_fds=True) == 0
            except Exception:
                return False

    @staticmethod
    def get_command_line():
        return ['nohup', 'lxterminal', '-e']


class MateTerminal(LinuxTerminal):
    """Handles the Mate desktop environment using mate-terminal."""
    name = "Mate"

    @staticmethod
    def detect():
        if os.environ.get('MATE_DESKTOP_SESSION_ID', ''):
            return True
        with open(os.devnull, 'w', encoding="utf-8") as FNULL:
            try:
                return subprocess.call(
                    [
                        'dbus-send', '--print-reply',
                        '--dest=org.freedesktop.DBus',
                        '/org/freedesktop/DBus',
                        'org.freedesktop.DBus.GetNameOwner',
                        'string:org.mate.SessionManager'
                    ], stdout=FNULL, stderr=FNULL) == 0
            except Exception:
                return False

    @staticmethod
    def get_command_line():
        return ['nohup', 'mate-terminal', '-x']


class i3Terminal(LinuxTerminal):
    """Handles terminals in the i3 desktop environment."""
    name = "i3"

    @staticmethod
    def detect():
        return os.environ.get('DESKTOP_STARTUP_ID', '').startswith('i3/')

    @staticmethod
    def get_command_line():
        return ['nohup', 'i3-sensible-terminal', '-e']


# Generic terminals (rxvt, xterm, etc.)
class rxvtTerminal(LinuxTerminal):
    """Handles rxvt and urxvt terminals. urxvt is used if both are available."""
    name = "(u)rxvt"

    @staticmethod
    def detect():
        with open(os.devnull, 'w', encoding="utf-8") as FNULL:
            try:
                if subprocess.call(
                        ['which', 'urxvt'], stdout=FNULL, stderr=FNULL) == 0:
                    rxvtTerminal.exe = 'urxvt'
                    return True
                if subprocess.call(
                        ['which', 'rxvt'], stdout=FNULL, stderr=FNULL) == 0:
                    rxvtTerminal.exe = 'rxvt'
                    return True
            except Exception:
                return False
        return None

    @staticmethod
    def get_command_line():
        return ['nohup', rxvtTerminal.exe, '-e']


class xtermTerminal(LinuxTerminal):
    """Handles the xterm terminal."""
    name = "xterm"

    @staticmethod
    def detect():
        with open(os.devnull, 'w', encoding="utf-8") as FNULL:
            try:
                return subprocess.call(
                    ['which', 'xterm'], stdout=FNULL, stderr=FNULL,
                    close_fds=True) == 0
            except Exception:
                return False

    @staticmethod
    def get_command_line():
        return ['nohup', 'xterm', '-e']


class CustomTerminal(LinuxTerminal):
    """Allows custom terminal commands to handle missing cases."""
    name = "Custom command"

    @staticmethod
    def detect():
        # Custom commands are always an option
        return True

    @staticmethod
    def get_command_line():
        cmd = get_custom_terminal_cmd()
        if cmd:
            return shlex.split(cmd)
        return []

# Terminal testing algorithm:
#     Main app, in thread:
#         Generate temporary file name for synchronization.
#         Write 0 to file.
#         Start parent w/ file name.
#         Wait for parent to terminate.
#         Report success if file contains 4 within 10 seconds, else report error.
#         Delete temporary file.
#
#     Parent:
#         If child process stops running, terminate.
#         Start child process.
#         Write 1 to file.
#         Ensure file contains 2 within 10 seconds.
#         Write 3 to file. Terminate self.
#
#     Child:
#         Ensure file contains 1 within 10 seconds.
#         Write 2 to file.
#         Ensure file contains 3 within 10 seconds.
#         Wait 3 seconds for parent to terminate.
#         Write 4 to file. Terminate self.


def _terminal_test_wait(fn, value):
    """
    Waits for file named <fn> to contain <value>.
    Returns True on success, False on timeout.
    """
    value = str(value)
    timer = 0
    interval = 0.5
    while timer < 10:
        time.sleep(interval)
        try:
            with open(fn, 'r', encoding="utf-8") as f:
                if value == f.read().strip():
                    return True
        except Exception:
            pass
        timer += interval
    return False


def _terminal_test_report(fn, value):
    """Writes a status value <value> to file named <fn>."""
    while True:
        try:
            with open(fn, 'w+b') as f:
                f.write(str(value).encode('utf-8'))
            return
        except IOError:
            pass


def terminal_test_run(status_callback=None):
    """
    Starts and manages the test of terminal launching.
    Intermittent status messages may be received on function <status_callback>.
    Return value is (a, b), where a is a boolean marking the success of the test
    and b is a status text describing the result.
    """
    progress = {
        '0': 'Spawning parent process...',
        '1': 'Spawning child process...',
        '2': 'Waiting for parent to terminate...',
        '3': 'Waiting for child to terminate...',
        '4': 'Done!'
    }
    endmsg = {
        '0': 'Failed to start test. Check that the command is typed correctly.',
        '1': 'Parent process was blocked by child.',
        '2': 'Parent process died before child reported back.',
        '3': 'Child process was terminated along with parent.',
        '4': 'Success!'
    }

    log.d("Starting terminal test.")
    with tempfile.NamedTemporaryFile(delete=False) as f:
        t = f.name
    cmd = sys.argv[:] + ['--terminal-test-parent', t]
    if sys.executable not in cmd:
        cmd.insert(0, sys.executable)
    cmd = get_terminal_command(cmd, True)
    with subprocess.Popen(cmd) as p:
        while p.poll() is None:
            with open(t, 'r', encoding="utf-8") as f:
                value = f.read()
            try:
                log.d("Test status: %s" % progress[value])
                status_callback(progress[value])
            except Exception:
                pass
    timer = 0
    interval = 0.5
    while timer < 10:
        time.sleep(interval)
        try:
            with open(t, 'r', encoding="utf-8") as f:
                value = f.read().strip()
                log.d("Test status: %s" % progress[value])
                if value == '4':
                    break
        except Exception:
            pass
        timer += interval
    os.remove(t)
    log.d("Test complete: %s" % endmsg.get(value, 'Unknown error.'))
    return (value == '4', endmsg.get(value, 'Unknown error.'))


def terminal_test_parent(t):
    """Tests the parent side of terminal launching."""
    print("Terminal successfully started! Test will begin in 3 seconds.")
    time.sleep(3)
    cmd = sys.argv[:-2] + ['--terminal-test-child', t]
    if sys.executable not in cmd:
        cmd.insert(0, sys.executable)
    cmd = get_terminal_command(cmd, True)
    print("Launching child process...")
    with subprocess.Popen(cmd, stdin=subprocess.PIPE,
                          stdout=subprocess.PIPE) as p:
        _terminal_test_report(t, 1)
        if p.poll():
            return 1
        print("Waiting for child process to start...")
        if not _terminal_test_wait(t, 2):
            p.kill()
            return 2
        if p.poll():
            return 1
    print("Terminating parent process...")
    _terminal_test_report(t, 3)
    return 0


def terminal_test_child(t):
    """Tests the child side of terminal launching."""
    t = sys.argv[-1]
    print("Waiting for parent process to continue...")
    if not _terminal_test_wait(t, 1):
        return 1
    _terminal_test_report(t, 2)
    print("Waiting for parent process to terminate...")
    if not _terminal_test_wait(t, 3):
        return 1
    print("Test complete. Will terminate in 3 seconds.")
    time.sleep(3)
    _terminal_test_report(t, 4)
    return 0