#!/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 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): # 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'] = 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') mojo_bindings_path = \ os.path.join(SRC_ROOT, 'mojo/public/dart/bindings.dart') mojo_core_path = os.path.join(SRC_ROOT, 'mojo/public/dart/core.dart') analyzer_args = [ANALYZER_PATH, "--url-mapping=dart:sky,%s" % dart_sky_path, "--url-mapping=dart:sky_builtin,%s" % sky_builtin_path, "--url-mapping=mojo:bindings,%s" % mojo_bindings_path, "--url-mapping=mojo:core,%s" % mojo_core_path, args.app_path ] subprocess.call(analyzer_args) 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(), Analyze()]: 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()