flutter_flutter/tools/shelldb

#!/usr/bin/env python
# Copyright 2015 The Chromium Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.

from skypy.skyserver import SkyServer
import argparse
import json
import logging
import os
import subprocess
import sys
import urllib
import urlparse

SKY_TOOLS_DIR = os.path.dirname(__file__)
SKY_ROOT = os.path.dirname(SKY_TOOLS_DIR)
SRC_ROOT = os.path.dirname(SKY_ROOT)

SKY_SERVER_PORT = 9888
DEFAULT_URL = "sky://domokit.github.io/home"
APK_NAME = 'SkyDemo.apk'
ADB_PATH = os.path.join(SRC_ROOT,
    'third_party/android_tools/sdk/platform-tools/adb')


PID_FILE_PATH = "/tmp/skydemo.pids"
PID_FILE_KEYS = frozenset([
    'remote_sky_server_port',
    'sky_server_pid',
    'sky_server_port',
    'sky_server_root',
    'build_dir',
])

# This 'strict dictionary' approach is useful for catching typos.
class Pids(object):
    def __init__(self, known_keys, contents=None):
        self._known_keys = known_keys
        self._dict = contents if contents is not None else {}

    def __len__(self):
        return len(self._dict)

    def __getitem__(self, key):
        assert key in self._known_keys, '%s not in known_keys' % key
        return self._dict[key]

    def __setitem__(self, key, value):
        assert key in self._known_keys, '%s not in known_keys' % key
        self._dict[key] = value

    def __delitem__(self, key):
        assert key in self._known_keys, '%s not in known_keys' % key
        del self._dict[key]

    def __iter__(self):
        return iter(self._dict)

    def __contains__(self, key):
        assert key in self._known_keys, '%s not in allowed_keys' % key
        return key in self._dict

    def clear():
        self._dict = {}

    @classmethod
    def read_from(cls, path, known_keys):
        contents = {}
        try:
            with open(path, 'r') as pid_file:
                contents = json.load(pid_file)
        except:
            if os.path.exists(path):
                logging.warn('Failed to read pid file: %s' % path)
        return cls(known_keys, contents)

    def write_to(self, path):
        try:
            with open(path, 'w') as pid_file:
                json.dump(self._dict, pid_file, indent=2, sort_keys=True)
        except:
            logging.warn('Failed to write pid file: %s' % path)


def _convert_to_sky_url(url):
    result = urllib.parse.urlsplit(url)
    result.scheme = 'sky'
    return urllib.parse.urlunsplit(result)


# A free function for possible future sharing with a 'load' command.
def _url_from_args(args):
    if urlparse.urlparse(args.url_or_path).scheme:
        return args.url_or_path
    # The load happens on the remote device, use the remote port.
    remote_sky_server_port = self.pids.get('remote_sky_server_port',
        self.pids['sky_server_port'])
    url = SkyServer.url_for_path(remote_sky_server_port,
        self.pids['sky_server_root'], args.url_or_path)
    return _convert_to_sky_url(url)


class StartSky(object):
    def add_subparser(self, subparsers):
        start_parser = subparsers.add_parser('start',
            help='launch SKyShell.apk on the device')
        start_parser.add_argument('build_dir', type=str)
        start_parser.add_argument('url_or_path', nargs='?', type=str,
            default=DEFAULT_URL)
        start_parser.set_defaults(func=self.run)

    def _server_root_for_url(self, url_or_path):
        path = os.path.abspath(url_or_path)
        if os.path.commonprefix([path, SRC_ROOT]) == SRC_ROOT:
            server_root = SRC_ROOT
        else:
            server_root = os.path.dirname(path)
            logging.warn(
                '%s is outside of mojo root, using %s as server root' %
                (path, server_root))
        return server_root

    def _sky_server_for_args(self, args):
        # FIXME: This is a hack.  sky_server should just take a build_dir
        # not a magical "configuration" name.
        configuration = os.path.basename(os.path.normpath(args.build_dir))
        server_root = self._server_root_for_url(args.url_or_path)
        sky_server = SkyServer(SKY_SERVER_PORT, configuration, server_root)
        return sky_server

    def run(self, args, pids):
        apk_path = os.path.join(args.build_dir, 'apks', APK_NAME)
        if not os.path.exists(apk_path):
            print "'%s' does not exist?" % apk_path
            return 2

        sky_server = self._sky_server_for_args(args)
        pids['sky_server_pid'] = sky_server.start()
        pids['sky_server_port'] = sky_server.port
        pids['sky_server_root'] = sky_server.root

        pids['build_dir'] = args.build_dir

        subprocess.check_call([ADB_PATH, 'install', '-r', apk_path])

        port_string = 'tcp:%s' % sky_server.port
        subprocess.check_call([
            ADB_PATH, 'reverse', port_string, port_string
        ])
        pids['remote_sky_server_port'] = sky_server.port

        subprocess.check_call([ADB_PATH, 'shell',
            'am', 'start',
            '-a', 'android.intent.action.VIEW',
            '-d', _url_from_args(args)])


class StopSky(object):
    def add_subparser(self, subparsers):
        stop_parser = subparsers.add_parser('stop',
            help=('kill all running SkyShell.apk processes'))
        stop_parser.set_defaults(func=self.run)

    def _kill_if_exists(self, pids, key, name):
        pid = pids.pop(key, None)
        if not pid:
            logging.info('No pid for %s, nothing to do.' % name)
            return
        logging.info('Killing %s (%d).' % (name, pid))
        try:
            os.kill(pid, signal.SIGTERM)
        except OSError:
            logging.info('%s (%d) already gone.' % (name, pid))

    def run(self, args, pids):
        self._kill_if_exists(pids, 'sky_server_pid', 'sky_server')

        if 'remote_sky_server_port' in self.pids:
            port_string = 'tcp:%s' % self.pids['remote_sky_server_port']
            subprocess.call([ADB_PATH, 'reverse', '--remove', port_string])

        subprocess.call([
            ADB_PATH, 'shell', 'am', 'force-stop', ANDROID_PACKAGE])

        pids.clear()


class SkyShellRunner(object):
    def main(self):
        logging.basicConfig(level=logging.WARNING)

        parser = argparse.ArgumentParser(description='Sky Shell Runner')
        subparsers = parser.add_subparsers(help='sub-command help')

        for command in [StartSky(), StopSky()]:
            command.add_subparser(subparsers)

        args = parser.parse_args()
        pids = Pids.read_from(PID_FILE_PATH, PID_FILE_KEYS)
        exit_code = args.func(args, pids)
        # We could do this with an at-exit handler instead?
        pids.write_to(PID_FILE_PATH)
        sys.exit(exit_code)


if __name__ == '__main__':
    SkyShellRunner().main()