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 signal
import subprocess
import sys
import time
import urlparse
import hashlib
import tempfile

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
OBSERVATORY_PORT = 8181
DEFAULT_URL = "sky://domokit.github.io/sky_home"
APK_NAME = 'SkyDemo.apk'
ADB_PATH = os.path.join(SRC_ROOT,
    'third_party/android_tools/sdk/platform-tools/adb')
ANDROID_PACKAGE = "org.domokit.sky.demo"
SHA1_PATH = '/sdcard/%s/%s.sha1' % (ANDROID_PACKAGE, APK_NAME)

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

_IGNORED_PATTERNS = [
  # Ignored because they're not indicative of specific errors.
  re.compile(r'^$'),
  re.compile(r'^Analyzing \['),
  re.compile(r'^No issues found'),
  re.compile(r'^[0-9]+ errors? and [0-9]+ warnings? found.'),
  re.compile(r'^([0-9]+|No) (error|warning|issue)s? found.'),

  # TODO: Remove once sdk-extensions are in place
  re.compile(r'^\[error\] Native functions can only be declared in'),
  # TODO: Remove this once dev SDK includes Uri.directory constructor.
  re.compile(r'.*The class \'Uri\' does not have a constructor \'directory\''),
  # TODO: Remove this once Sky no longer generates this warning.
  # dartbug.com/22836
  re.compile(r'.*cannot both be unnamed'),
]

# 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 = {}

    def pop(self, key, default=None):
        assert key in self._known_keys, '%s not in allowed_keys' % key
        return self._dict.pop(key, default)

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


def dev_packages_root(build_dir):
    return os.path.join(build_dir, 'gen', 'dart-pkg', 'packages')


def ensure_assets_are_downloaded(build_dir):
    sky_pkg_dir = os.path.join(build_dir, 'gen', 'dart-pkg', 'sky')
    sky_pkg_lib_dir = os.path.join(sky_pkg_dir, 'lib')
    sky_icons_dir = \
        os.path.join(sky_pkg_lib_dir, 'assets', 'material-design-icons')
    if not os.path.isdir(sky_icons_dir):
        logging.info('NOTE: sky/assets/material-design-icons missing, '
                     'Running `download_material_design_icons` for you.')
        subprocess.check_call(
            [os.path.join(sky_pkg_lib_dir, 'download_material_design_icons')])


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.add_argument('--no_install', action="store_false",
            default=True, dest="install",
            help="Don't install SkyDemo.apk before starting")
        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

        ensure_assets_are_downloaded(args.build_dir)

        packages_root = dev_packages_root(args.build_dir)
        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)

        if args.install:
            # We might need to install a new APK, so check SHA1
            source_sha1 = hashlib.sha1(open(apk_path, 'rb').read()).hexdigest()
            dest_sha1 = subprocess.check_output([ADB_PATH, 'shell', 'cat', SHA1_PATH])
            use_existing_apk = False
            if source_sha1 == dest_sha1:
                # Make sure that the APK didn't get uninstalled somehow
                use_existing_apk = subprocess.check_output([
                    ADB_PATH, 'shell', 'pm', 'list', 'packages', ANDROID_PACKAGE
                ])
        else:
            # User is telling us not to bother installing an APK
            use_existing_apk = True

        if use_existing_apk:
            # APK is already on the device, we only need to stop it
            subprocess.check_call([
                ADB_PATH, 'shell', 'am', 'force-stop', ANDROID_PACKAGE
            ])
        else:
            # Slow path, need to upload a new APK to the device
            # -r to replace an existing apk, -d to allow version downgrade.
            subprocess.check_call([ADB_PATH, 'install', '-r', '-d', apk_path])
            # record the SHA1 of the APK we just pushed
            with tempfile.NamedTemporaryFile() as fp:
                fp.write(source_sha1)
                fp.seek(0)
                subprocess.check_call([ADB_PATH, 'push', fp.name, SHA1_PATH])

        # Set up port forwarding for observatory
        port_string = 'tcp:%s' % OBSERVATORY_PORT
        subprocess.check_call([
            ADB_PATH, 'forward', port_string, port_string
        ])

        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 pids:
            port_string = 'tcp:%s' % 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):
        build_dir = pids.get('build_dir')
        if not build_dir:
            logging.fatal("pids file missing build_dir. Try 'start' first.")
            return 2
        ANALYZER_PATH = 'third_party/dart-sdk/dart-sdk/bin/dartanalyzer'

        bindings_path = os.path.join(build_dir, 'gen/sky/bindings')
        sky_builtin_path = \
            os.path.join(SRC_ROOT, 'sky/engine/bindings/builtin.dart')
        sky_internals_path = \
            os.path.join(SRC_ROOT, 'sky/sdk/lib/internals.dart')
        dart_sky_path = os.path.join(bindings_path, 'dart_sky.dart')
        analyzer_args = [ANALYZER_PATH,
            "--url-mapping=dart:sky.internals,%s" % sky_internals_path,
            "--url-mapping=dart:sky,%s" % dart_sky_path,
            "--url-mapping=dart:sky_builtin,%s" % sky_builtin_path,
            "--package-root", dev_packages_root(build_dir),
            "--package-warnings",
            args.app_path
        ]
        try:
          subprocess.check_output(analyzer_args,
                                  shell=False,
                                  stderr=subprocess.STDOUT)
        except subprocess.CalledProcessError as e:
          errors = set(l for l in e.output.split('\n')
                       if not any(p.match(l) for p in _IGNORED_PATTERNS))
          # If we do not have any errors left after filtering, return 0.
          if len(errors) == 0:
            return 0
          # Print errors.
          for error in sorted(errors):
            print >> sys.stderr, error
          # Return analyzer error code.
          return e.returncode
        return 0


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