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 re
import subprocess
import sys
import time
import urlparse

SKY_TOOLS_DIR = os.path.dirname(os.path.abspath(__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 get(self, key, default=None):
        assert key in self._known_keys, '%s not in known_keys' % key
        return self._dict.get(key, default)

    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):
        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):
    parts = urlparse.urlsplit(url)
    parts = parts._replace(scheme='sky')
    return parts.geturl()


# A free function for possible future sharing with a 'load' command.
def _url_from_args(args, pids):
    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 = pids.get('remote_sky_server_port',
        pids['sky_server_port'])
    url = SkyServer.url_for_path(remote_sky_server_port,
        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, packages_root):
        # 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, packages_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

        sdk_root = os.path.join(args.build_dir, 'gen', 'sky_sdk')
        packages_root = os.path.join(sdk_root, 'packages_root')
        sky_tools_directory = os.path.join(SRC_ROOT, 'sky/tools')
        subprocess.check_call([
            os.path.join(sky_tools_directory, 'deploy_sdk.py'),
            '--build-dir', args.build_dir,
            '--non-interactive',
            '--dev-environment',
            '--fake-pub-get-into', packages_root,
            sdk_root,
        ])

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

        pids['build_dir'] = os.path.abspath(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, pids)])


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 Analyze(object):
    def add_subparser(self, subparsers):
        analyze_parser = subparsers.add_parser('analyze',
            help=('run the dart analyzer with sky url mappings'))
        analyze_parser.add_argument('app_path', type=str)
        analyze_parser.set_defaults(func=self.run)

    def run(self, args, pids):
        ANALYZER_PATH = 'third_party/dart-sdk/dart-sdk/bin/dartanalyzer'

        build_dir = os.path.abspath(pids['build_dir'])
        bindings_path = os.path.join(build_dir, 'gen/sky/bindings')
        sky_builtin_path = \
            os.path.join(SRC_ROOT, 'sky/engine/bindings/builtin.dart')
        dart_sky_path = os.path.join(bindings_path, 'dart_sky.dart')
        analyzer_args = [ANALYZER_PATH,
            "--url-mapping=dart:sky,%s" % dart_sky_path,
            "--url-mapping=dart:sky_builtin,%s" % sky_builtin_path,
            args.app_path
        ]
        subprocess.call(analyzer_args)

class StartTracing(object):
    def add_subparser(self, subparsers):
        start_tracing_parser = subparsers.add_parser('start_tracing',
            help=('start tracing a running sky instance'))
        start_tracing_parser.set_defaults(func=self.run)

    def run(self, args, pids):
        subprocess.check_output([ADB_PATH, 'shell',
            'am', 'broadcast',
            '-a', 'org.domokit.sky.demo.TRACING_START'])


TRACE_COMPLETE_REGEXP = re.compile('Trace complete')
TRACE_FILE_REGEXP = re.compile(r'Saving trace to (?P<path>\S+)')

class StopTracing(object):
    def add_subparser(self, subparsers):
        stop_tracing_parser = subparsers.add_parser('stop_tracing',
            help=('stop tracing a running sky instance'))
        stop_tracing_parser.set_defaults(func=self.run)

    def run(self, args, pids):
        subprocess.check_output([ADB_PATH, 'logcat', '-c'])
        subprocess.check_output([ADB_PATH, 'shell',
            'am', 'broadcast',
            '-a', 'org.domokit.sky.demo.TRACING_STOP'])
        device_path = None
        is_complete = False
        while not is_complete:
            time.sleep(0.2)
            log = subprocess.check_output([ADB_PATH, 'logcat', '-d'])
            if device_path is None:
                result = TRACE_FILE_REGEXP.search(log)
                if result:
                    device_path = result.group('path')
            is_complete = TRACE_COMPLETE_REGEXP.search(log) is not None

        print 'Downloading trace %s ...' % os.path.basename(device_path)

        if device_path:
            subprocess.check_output([ADB_PATH, 'pull', device_path])
            subprocess.check_output([ADB_PATH, 'shell', 'rm', device_path])


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')

        commands = [
            StartSky(),
            StopSky(),
            Analyze(),
            StartTracing(),
            StopTracing(),
        ]

        for command in commands:
            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()