Source code for tksn.adb.call

# -*- coding: utf-8 -*-
"""Implementation of subprocess invocation

:copyright: (c) 2016 by tksn
:license: MIT
"""

from __future__ import unicode_literals
import logging
import os
import subprocess


_logger = logging.getLogger('tksn.adb.call')


def _log_command(command):
    _logger.info(msg=' '.join(command))


class CallError(Exception):
    """Call error

    Thrown when error occurs upon invocation of external commands.
    """


def get_adb_executable():
    """Get path to ADB executable

    Returns:
        str: Path to adb.exe under ANDROID_HOME if ANDROID_HOME exists in environment variables.
            otherwise 'adb'.
    """
    android_home = os.environ.get('ANDROID_HOME')
    if android_home:
        return os.path.join(android_home, os.path.join('platform-tools', 'adb'))
    return 'adb'


def call_sync(command):
    """Call given command as subprocess, and wait until it finishes

    Args:
        command (sequence): sequence of command strings
    Returns:
        tuple: (stdout, stderr) of the subprocess
    Raises:
        CallError: if command invocation failed
    """
    _log_command(command)
    proc = subprocess.Popen(
        command, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
    out_err = proc.communicate()
    if proc.returncode != 0:
        raise CallError('Call {0} failed'.format(' '.join(command)))
    return out_err


def call_async(command, out=None, err=None):
    """Call given command as subprocess, and returns immediately.

    Args:
        command (sequence): list of command strings
        out (file): file object used as stdout of the subprocess.
        err (file): file object used as stderr of the subprocess.
    Returns:
        subprocess.Popen: Popen object
    """
    _log_command(command)
    proc = subprocess.Popen(command, stdout=out, stderr=err)
    return proc