#!/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. import argparse import atexit import errno import hashlib import json import logging import multiprocessing import os import platform import random import re import signal import socket import subprocess import sys import tempfile import time import urlparse PACKAGES_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) SKY_ENGINE_DIR = os.path.join(PACKAGES_DIR, 'sky_engine') APK_DIR = os.path.join(os.path.realpath(SKY_ENGINE_DIR), os.pardir, 'apks') SKY_SERVER_PORT = 9888 OBSERVATORY_PORT = 8181 ADB_PATH = 'adb' APK_NAME = 'SkyShell.apk' ANDROID_PACKAGE = 'org.domokit.sky.shell' ANDROID_COMPONENT = '%s/%s.SkyActivity' % (ANDROID_PACKAGE, ANDROID_PACKAGE) SHA1_PATH = '/sdcard/%s/%s.sha1' % (ANDROID_PACKAGE, APK_NAME) SKY_SHELL_APP_ID = 'com.google.SkyShell' IOS_APP_NAME = 'SkyShell.app' # FIXME: Do we need to look in $DART_SDK? DART_PATH = 'dart' PUB_PATH = 'pub' PID_FILE_PATH = '/tmp/sky_tool.pids' PID_FILE_KEYS = frozenset([ 'remote_sky_server_port', 'sky_server_pid', 'sky_server_port', 'sky_server_root', ]) IOS_SIM_PATH = [ os.path.join('/Applications', 'iOS Simulator.app', 'Contents', 'MacOS', 'iOS Simulator') ] XCRUN_PATH = [ os.path.join('/usr', 'bin', 'env'), 'xcrun', ] SIMCTL_PATH = XCRUN_PATH + [ 'simctl', ] PLIST_BUDDY_PATH = XCRUN_PATH + [ 'PlistBuddy', ] def _port_in_use(port): sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) return sock.connect_ex(('localhost', port)) == 0 def _start_http_server(port, root): server_command = [ PUB_PATH, 'run', 'sky_tools:sky_server', str(port), ] logging.info(' '.join(server_command)) return subprocess.Popen(server_command, cwd=root).pid # 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 known_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): # These keys are required to write a valid file. if not self._dict.viewkeys() >= { 'sky_server_pid', 'sky_server_port' }: return 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 _url_for_path(port, root, path): relative_path = os.path.relpath(path, root) return 'http://localhost:%s/%s' % (port, relative_path) class SkyLogs(object): def add_subparser(self, subparsers): logs_parser = subparsers.add_parser('logs', help='Show logs for running Sky apps') logs_parser.add_argument('--clear', action='store_true', dest='clear_logs', help='Clear log history before reading from logs (currently only implemented for Android)') logs_parser.set_defaults(func=self.run) def run(self, args, pids): android_log_reader = None ios_dev_log_reader = None ios_sim_log_reader = None android = AndroidDevice() if android.is_connected(): android_log_reader = android.logs(args.clear_logs) if IOSDevice.is_connected(): ios_dev_log_reader = IOSDevice.logs(args.clear_logs) if IOSSimulator.is_connected(): ios_sim_log_reader = IOSSimulator.logs(args.clear_logs) if android_log_reader is not None: try: android_log_reader.join() except KeyboardInterrupt: pass if ios_dev_log_reader is not None: try: ios_dev_log_reader.join() except KeyboardInterrupt: pass if ios_sim_log_reader is not None: try: ios_sim_log_reader.join() except KeyboardInterrupt: pass class InstallSky(object): def add_subparser(self, subparsers): install_parser = subparsers.add_parser('install', help='install SkyShell on Android and iOS devices and simulators') install_parser.set_defaults(func=self.run) def run(self, args, pids): android = AndroidDevice() installed_somewhere = False # Install on connected Android device if android.is_connected() and args.android_build_available: installed_somewhere = installed_somewhere or android.install_apk(android.get_apk_path(args)) # Install on connected iOS device if IOSDevice.is_connected() and args.ios_build_available: installed_somewhere = installed_somewhere or IOSDevice.install_app(IOSDevice.get_app_path(args)) # Install on iOS simulator if it's running if IOSSimulator.is_booted() and args.ios_sim_build_available: installed_somewhere = installed_somewhere or IOSSimulator.fork_install_app(IOSSimulator.get_app_path(args)) if installed_somewhere: return 0 else: return 2 # TODO(iansf): get rid of need for args def needs_install(self, args): return AndroidDevice().needs_install(args) or IOSDevice.needs_install(args) or IOSSimulator.needs_install(args) class StartSky(object): def add_subparser(self, subparsers): start_parser = subparsers.add_parser('start', help='launch %s on the device' % APK_NAME) start_parser.add_argument('--poke', action='store_true') start_parser.add_argument('--checked', action='store_true') start_parser.add_argument('project_or_path', nargs='?', type=str, default='.') start_parser.set_defaults(func=self.run) def run(self, args, pids): started_sky_somewhere = False if not args.poke: StopSky().run(args, pids) # Only install if the user did not specify a poke installer = InstallSky() if installer.needs_install(args): started_sky_somewhere = (installer.run(args, pids) == 0) project_or_path = os.path.abspath(args.project_or_path) if os.path.isdir(project_or_path): sky_server_root = project_or_path main_dart = os.path.join(project_or_path, 'lib', 'main.dart') missing_msg = 'Missing lib/main.dart in project: %s' % project_or_path else: sky_server_root = os.getcwd() main_dart = project_or_path missing_msg = '%s does not exist.' % main_dart if not os.path.isfile(main_dart): logging.error(missing_msg) return 2 package_root = os.path.join(sky_server_root, 'packages') if not os.path.isdir(package_root): logging.error('%s is not a valid packages path.' % package_root) return 2 android = AndroidDevice() # TODO(iansf): fix this so that we don't have to pass sky_server_root, main_dart, pid, and args. started_sky_on_android = android.setup_servers(sky_server_root, main_dart, pids, args) if started_sky_somewhere or started_sky_on_android: return 0 else: return 2 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 _run(self, args): with open('/dev/null', 'w') as dev_null: logging.info(' '.join(args)) subprocess.call(args, stdout=dev_null, stderr=dev_null) def run(self, args, pids): if 'remote_sky_server_port' in pids: port_string = 'tcp:%s' % pids['remote_sky_server_port'] self._run([AndroidDevice().adb_path, 'reverse', '--remove', port_string]) self._run([AndroidDevice().adb_path, 'shell', 'am', 'force-stop', ANDROID_PACKAGE]) try: # Because the server gets forked by dart, pids.get(['sky_server_pid']) returns an invalid pid, # so just force things closed. if platform.system() == 'Darwin': cmd = ['lsof', '-i', ':%s' % SKY_SERVER_PORT, '-t'] logging.info(' '.join(cmd)) pid = subprocess.check_output(cmd) # Killing a pid with a shell command from within python is hard, # so use a library command, but it's still nice to give the # equivalent command when doing verbose logging. logging.info('kill %s' % pid) os.kill(int(pid), signal.SIGTERM) else: # This usage of fuser is not valid on OS X self._run(['fuser', '-k', '%s/tcp' % SKY_SERVER_PORT]) except subprocess.CalledProcessError as e: pass pids.clear() class AndroidDevice(object): # _state used in this manner gives a simple way to treat AndroidDevice # as a singleton while easily allowing subclassing for mocks. All # AndroidDevices created in a given session will share the same state. _state = {} def __init__(self): self.__dict__ = AndroidDevice._state self._update_paths() # Checking for lollipop only needs to be done if we are starting an # app, but it has an important side effect, which is to discard any # progress messages if the adb server is restarted. self._check_for_adb() self._check_for_lollipop_or_later() def _update_paths(self): if 'adb_path' in self.__dict__: return if 'ANDROID_HOME' in os.environ: android_home_dir = os.environ['ANDROID_HOME'] adb_location1 = os.path.join(android_home_dir, 'sdk', 'platform-tools', 'adb') adb_location2 = os.path.join(android_home_dir, 'platform-tools', 'adb') if os.path.exists(adb_location1): self.adb_path = adb_location1 elif os.path.exists(adb_location2): self.adb_path = adb_location2 else: logging.warning('"adb" not found at\n "%s" or\n "%s"\nusing default path "%s"' % (adb_location1, adb_location2, ADB_PATH)) self.adb_path = ADB_PATH else: self.adb_path = ADB_PATH def _is_valid_adb_version(self, adb_version): # Sample output: 'Android Debug Bridge version 1.0.31' version_fields = re.search('(\d+)\.(\d+)\.(\d+)', adb_version) if version_fields: major_version = int(version_fields.group(1)) minor_version = int(version_fields.group(2)) patch_version = int(version_fields.group(3)) if major_version > 1: return True if major_version == 1 and minor_version > 0: return True if major_version == 1 and minor_version == 0 and patch_version >= 32: return True return False else: logging.warn('Unrecognized adb version string. Skipping version check.') return True def _check_for_adb(self): if 'has_valid_adb' in self.__dict__: return try: cmd = [self.adb_path, 'version'] logging.info(' '.join(cmd)) adb_version = subprocess.check_output(cmd) if self._is_valid_adb_version(adb_version): self.has_valid_adb = True return cmd = ['which', ADB_PATH] logging.info(' '.join(cmd)) adb_path = subprocess.check_output(cmd).rstrip() logging.error('"%s" is too old. Need 1.0.32 or later. ' 'Try setting ANDROID_HOME to use Android builds. Android builds are unavailable.' % adb_path) self.has_valid_adb = False except OSError: logging.warning('"adb" (from the Android SDK) not in $PATH, Android builds are unavailable.') self.has_valid_adb = False def _check_for_lollipop_or_later(self): if 'has_valid_android' in self.__dict__: return try: # If the server is automatically restarted, then we get irrelevant # output lines like this, which we want to ignore: # adb server is out of date. killing.. # * daemon started successfully * cmd = [self.adb_path, 'start-server'] logging.info(' '.join(cmd)) subprocess.call(cmd) cmd = [self.adb_path, 'shell', 'getprop', 'ro.build.version.sdk'] logging.info(' '.join(cmd)) sdk_version = subprocess.check_output(cmd).rstrip() # Sample output: '22' if not sdk_version.isdigit(): logging.error('Unexpected response from getprop: "%s".' % sdk_version) self.has_valid_android = False return if int(sdk_version) < 22: logging.error('Version "%s" of the Android SDK is too old. ' 'Need Lollipop (22) or later. ' % sdk_version) self.has_valid_android = False return except subprocess.CalledProcessError as e: # adb printed the error, so we print nothing. self.has_valid_android = False return self.has_valid_android = True def is_package_installed(self, package_name): if not self.is_connected(): return False pm_path_cmd = [self.adb_path, 'shell', 'pm', 'path', package_name] logging.info(' '.join(pm_path_cmd)) return subprocess.check_output(pm_path_cmd).strip() != '' def get_device_apk_sha1(self, apk_path): # We might need to install a new APK, so check SHA1 cmd = [self.adb_path, 'shell', 'cat', SHA1_PATH] logging.info(' '.join(cmd)) return subprocess.check_output(cmd) def get_source_sha1(self, apk_path): return hashlib.sha1(open(apk_path, 'rb').read()).hexdigest() # TODO(iansf): get rid of need for args def get_apk_path(self, args): if args.android_build_available and args.use_release: return os.path.join(os.path.normpath(args.sky_src_path), args.android_release_build_path, 'apks', APK_NAME) elif args.android_build_available and args.local_build: return os.path.join(os.path.normpath(args.sky_src_path), args.android_debug_build_path, 'apks', APK_NAME) else: return os.path.join(APK_DIR, APK_NAME) def is_connected(self): return self.has_valid_android def needs_install(self, args): apk_path = self.get_apk_path(args) if not self.is_package_installed(ANDROID_PACKAGE): logging.info('%s is not on the device. Installing now...' % APK_NAME) return True elif self.get_device_apk_sha1(apk_path) != self.get_source_sha1(apk_path): logging.info('%s on the device is out of date. Installing now...' % APK_NAME) return True return False def install_apk(self, apk_path): if not os.path.exists(apk_path): logging.error('"%s" does not exist.' % apk_path) return False cmd = [self.adb_path, 'install', '-r', apk_path] logging.info(' '.join(cmd)) subprocess.check_call(cmd) # record the SHA1 of the APK we just pushed with tempfile.NamedTemporaryFile() as fp: fp.write(self.get_source_sha1(apk_path)) fp.seek(0) cmd = [self.adb_path, 'push', fp.name, SHA1_PATH] logging.info(' '.join(cmd)) subprocess.check_call(cmd) return True # TODO(iansf): refactor setup_servers def setup_servers(self, sky_server_root, main_dart, pids, args): if not self.is_connected(): return False # Set up port forwarding for observatory observatory_port_string = 'tcp:%s' % OBSERVATORY_PORT cmd = [ self.adb_path, 'forward', observatory_port_string, observatory_port_string ] logging.info(' '.join(cmd)) subprocess.check_call(cmd) sky_server_port = SKY_SERVER_PORT pids['sky_server_port'] = sky_server_port if _port_in_use(sky_server_port): logging.info(('Port %s already in use. ' ' Not starting server for %s') % (sky_server_port, sky_server_root)) else: sky_server_pid = _start_http_server(sky_server_port, sky_server_root) pids['sky_server_pid'] = sky_server_pid pids['sky_server_root'] = sky_server_root port_string = 'tcp:%s' % sky_server_port cmd = [ self.adb_path, 'reverse', port_string, port_string ] logging.info(' '.join(cmd)) subprocess.check_call(cmd) pids['remote_sky_server_port'] = sky_server_port # The load happens on the remote device, use the remote port. url = _url_for_path(pids['remote_sky_server_port'], sky_server_root, main_dart) if args.poke: url += '?rand=%s' % random.random() cmd = [ self.adb_path, 'shell', 'am', 'start', '-a', 'android.intent.action.VIEW', '-d', url, ] if args.checked: cmd += ['--ez', 'enable-checked-mode', 'true'] cmd += [ANDROID_COMPONENT] logging.info(' '.join(cmd)) subprocess.check_output(cmd) return True def logs(self, clear=False): def do_logs(): if clear: cmd = [ self.adb_path, 'logcat', '-c' ] logging.info(' '.join(cmd)) subprocess.check_call(cmd) cmd = [ self.adb_path, 'logcat', '-v', 'tag', # Only log the tag and the message '-s', 'sky', 'chromium', ] logging.info(' '.join(cmd)) log_process = subprocess.Popen(cmd, bufsize=1, stdout=subprocess.PIPE) while True: try: log_line = log_process.stdout.readline() if log_line == '': if log_process.poll() != None: logging.error('The Android logging process has quit unexpectedly. Please call the "logs" command again.') break sys.stdout.write('ANDROID: ' + log_line) sys.stdout.flush() except KeyboardInterrupt: break log_reader = multiprocessing.Process(target=do_logs) log_reader.daemon = True log_reader.start() return log_reader class IOSDevice(object): _has_ios_deploy = None @classmethod def has_ios_deploy(cls): if cls._has_ios_deploy is not None: return cls._has_ios_deploy try: cmd = [ 'which', 'ios-deploy' ] logging.info(' '.join(cmd)) out = subprocess.check_output(cmd) match = re.search(r'ios-deploy', out) cls._has_ios_deploy = match is not None except subprocess.CalledProcessError: cls._has_ios_deploy = False return cls._has_ios_deploy _is_connected = False @classmethod def is_connected(cls): if not cls.has_ios_deploy(): return False if cls._is_connected: return True try: cmd = [ 'ios-deploy', '--detect', '--timeout', '1' ] logging.info(' '.join(cmd)) subprocess.check_output(cmd) cls._is_connected = True except subprocess.CalledProcessError: cls._is_connected = False return cls._is_connected @classmethod def get_app_path(cls, args): if args.use_release: return os.path.join(args.sky_src_path, args.ios_release_build_path, IOS_APP_NAME) else: return os.path.join(args.sky_src_path, args.ios_debug_build_path, IOS_APP_NAME) @classmethod def needs_install(cls, args): return cls.is_connected() @classmethod def install_app(cls, ios_app_path): if not cls.has_ios_deploy(): return False try: cmd = [ 'ios-deploy', '--justlaunch', '--timeout', '10', # Smaller timeouts cause it to exit before having launched the app '--bundle', ios_app_path ] logging.info(' '.join(cmd)) subprocess.check_call(cmd) except subprocess.CalledProcessError: return False return True @classmethod def copy_file(cls, bundle_id, local_path, device_path): if not cls.has_ios_deploy(): return try: cmd = [ 'ios-deploy', '-t', '1', '--bundle_id', bundle_id, '--upload', local_path, '--to', device_path ] logging.info(' '.join(cmd)) subprocess.check_call(cmd) except subprocess.CalledProcessError: pass @classmethod def logs(cls, clear=False): try: cmd = [ 'which', 'idevicesyslog' ] logging.info(' '.join(cmd)) subprocess.check_call(cmd) except subprocess.CalledProcessError: logging.error('"log" command only works with iOS devices if you have installed idevicesyslog. Run "brew install libimobiledevice" to install it with homebrew.') return None def do_logs(): cmd = [ 'idevicesyslog', ] logging.info(' '.join(cmd)) log_process = subprocess.Popen(cmd, stdout=subprocess.PIPE) while True: try: log_line = log_process.stdout.readline() if log_line == '': if log_process.poll() != None: logging.error('The iOS logging process has quit unexpectedly. Please call the "logs" command again.') break if re.match(r'.*SkyShell.*', log_line) is not None: sys.stdout.write('IOS DEV: ' + log_line) sys.stdout.flush() except KeyboardInterrupt: break log_reader = multiprocessing.Process(target=do_logs) log_reader.daemon = True log_reader.start() return log_reader class IOSSimulator(object): @classmethod def is_booted(cls): if platform.system() != 'Darwin': return False return cls.get_simulator_device_id() is not None @classmethod def is_connected(cls): return cls.is_booted() _device_id = None @classmethod def get_simulator_device_id(cls): if cls._device_id is not None: return cls._device_id cmd = SIMCTL_PATH + [ 'list', 'devices', ] logging.info(' '.join(cmd)) out = subprocess.check_output(cmd) match = re.search(r'[^\(]+\(([^\)]+)\) \(Booted\)', out) if match is not None and match.group(1) is not None: cls._device_id = match.group(1) return cls._device_id else: logging.info('No running simulators found') # TODO: Maybe start the simulator? return None if err is not None: print(err) exit(-1) _simulator_path = None @classmethod def get_simulator_path(cls): if cls._simulator_path is not None: return cls._simulator_path home_dir = os.path.expanduser('~') device_id = cls.get_simulator_device_id() if device_id is None: # TODO: Maybe start the simulator? return None cls._simulator_path = os.path.join(home_dir, 'Library', 'Developer', 'CoreSimulator', 'Devices', device_id) return cls._simulator_path _simulator_app_id = None @classmethod def get_simulator_app_id(cls): if cls._simulator_app_id is not None: return cls._simulator_app_id simulator_path = cls.get_simulator_path() cmd = [ 'find', os.path.join(simulator_path, 'data', 'Containers', 'Data', 'Application'), '-name', SKY_SHELL_APP_ID ] logging.info(' '.join(cmd)) out = subprocess.check_output(cmd) match = re.search(r'Data\/Application\/([^\/]+)\/Documents\/' + SKY_SHELL_APP_ID, out) if match is not None and match.group(1) is not None: cls._simulator_app_id = match.group(1) return cls._simulator_app_id else: logging.warning(SKY_SHELL_APP_ID + ' is not installed on the simulator') # TODO: Maybe install the app? return None if err is not None: print(err) exit(-1) _simulator_app_documents_dir = None @classmethod def get_simulator_app_documents_dir(cls): if cls._simulator_app_documents_dir is not None: return cls._simulator_app_documents_dir if not cls.is_booted(): return None simulator_path = cls.get_simulator_path() simulator_app_id = cls.get_simulator_app_id() if simulator_app_id is None: return None cls._simulator_app_documents_dir = os.path.join(simulator_path, 'data', 'Containers', 'Data', 'Application', simulator_app_id, 'Documents') return cls._simulator_app_documents_dir @classmethod def get_app_path(cls, args): if args.use_release: return os.path.join(args.sky_src_path, args.ios_sim_release_build_path, IOS_APP_NAME) else: return os.path.join(args.sky_src_path, args.ios_sim_debug_build_path, IOS_APP_NAME) @classmethod def needs_install(cls, args): return cls.is_booted() @classmethod def logs(cls, clear=False): def do_logs(): cmd = [ 'tail', '-f', os.path.expanduser('~/Library/Logs/CoreSimulator/' + cls.get_simulator_device_id() + '/system.log'), ] logging.info(' '.join(cmd)) log_process = subprocess.Popen(cmd, stdout=subprocess.PIPE) while True: try: log_line = log_process.stdout.readline() if log_line == '': if log_process.poll() != None: logging.error('The iOS Simulator logging process has quit unexpectedly. Please call the "logs" command again.') break if re.match(r'.*SkyShell.*', log_line) is not None: sys.stdout.write('IOS SIM: ' + log_line) sys.stdout.flush() except KeyboardInterrupt: break log_reader = multiprocessing.Process(target=do_logs) log_reader.daemon = True log_reader.start() return log_reader @classmethod def fork_install_app(cls, ios_app_path): try: cmd = [ os.path.abspath(__file__), 'ios_sim', '-p', ios_app_path, 'launch' ] logging.info(' '.join(cmd)) subprocess.check_call(cmd) return True except subprocess.CalledProcessError: return False def _process_args(self, args): if args.ios_sim_build_path is None: if args.ios_sim_build_available: if args.use_release: args.ios_sim_build_path = os.path.join(args.sky_src_path, args.ios_sim_release_build_path, IOS_APP_NAME) elif args.local_build: args.ios_sim_build_path = os.path.join(args.sky_src_path, args.ios_sim_debug_build_path, IOS_APP_NAME) if args.ios_sim_build_path is None: logging.error('ios_sim commands require a valid -p argument if not using the --release or --local-build global arguments') sys.exit(2) def get_application_identifier(self, path): cmd = PLIST_BUDDY_PATH + [ '-c', 'Print CFBundleIdentifier', os.path.join(path, 'Info.plist') ] logging.info(' '.join(cmd)) identifier = subprocess.check_output(cmd) return identifier.strip() def is_simulator_booted(self): cmd = SIMCTL_PATH + [ 'list', 'devices' ] logging.info(' '.join(cmd)) devices = subprocess.check_output(cmd).strip().split('\n') for device in devices: if re.search(r'\(Booted\)', device): return True return False # Launch whatever simulator the user last used, rather than try to guess which of their simulators they might want to use def boot_simulator(self, args, pids): # Guarantee that the args get processed, since all of the commands funnel through here. self._process_args(args) if self.is_simulator_booted(): return # Use Popen here because launching the simulator from the command line in this manner doesn't return, so we can't check the result. if args.ios_sim_path: logging.info(args.ios_sim_path) subprocess.Popen(args.ios_sim_path) else: logging.info(IOS_SIM_PATH) subprocess.Popen(IOS_SIM_PATH) while not self.is_simulator_booted(): print('Waiting for iOS Simulator to boot...') time.sleep(0.5) def install_app(self, args, pids): self.boot_simulator(args, pids) cmd = SIMCTL_PATH + [ 'install', 'booted', args.ios_sim_build_path, ] logging.info(' '.join(cmd)) return subprocess.check_call(cmd) def install_launch_and_wait(self, args, pids, wait): res = self.install_app(args, pids) if res != 0: return res identifier = self.get_application_identifier(args.ios_sim_build_path) launch_args = SIMCTL_PATH + ['launch'] if wait: launch_args += [ '-w' ] launch_args += [ 'booted', identifier, '-target', args.target, '-server', args.server ] logging.info(' '.join(launch_args)) return subprocess.check_output(launch_args).strip() def launch_app(self, args, pids): self.install_launch_and_wait(args, pids, False) def debug_app(self, args, pids): launch_res = self.install_launch_and_wait(args, pids, True) launch_pid = re.search('.*: (\d+)', launch_res).group(1) cmd = XCRUN_PATH + [ 'lldb', # TODO(iansf): get this working again # '-s', # os.path.join(os.path.dirname(__file__), 'lldb_start_commands.txt'), '-p', launch_pid, ] logging.info(' '.join(cmd)) return subprocess.call(cmd) def add_subparser(self, subparsers): simulator_parser = subparsers.add_parser('ios_sim', help='A script that launches an' ' application in the simulator and attaches' ' the debugger to it.') simulator_parser.add_argument('-p', dest='ios_sim_build_path', required=False, help='Path to the simulator app build. Defaults to values specified by ' 'the sky_src_path and ios_sim_[debug|release]_build_path parameters, ' 'which are normally specified by using the local-build or release ' 'parameters. Not normally required.') simulator_parser.add_argument('-t', dest='target', required=False, default='examples/demo_launcher/lib/main.dart', help='Sky server-relative path to the Sky app to run. Not normally required.') simulator_parser.add_argument('-s', dest='server', required=False, default='localhost:8080', help='Sky server address. Not normally required.') simulator_parser.add_argument('--ios_sim_path', dest='ios_sim_path', help='Path to your iOS Simulator executable. ' 'Not normally required.') subparsers = simulator_parser.add_subparsers() install_parser = subparsers.add_parser('install', help='Install app') install_parser.set_defaults(func=self.install_app) launch_parser = subparsers.add_parser('launch', help='Launch app. Automatically installs.') launch_parser.set_defaults(func=self.launch_app) debug_parser = subparsers.add_parser('debug', help='Debug app. Automatically installs and launches.') debug_parser.set_defaults(func=self.debug_app) class StartListening(object): def __init__(self): self.watch_cmd = None def add_subparser(self, subparsers): listen_parser = subparsers.add_parser('listen', help=('Listen for changes to files and reload the running app on all connected devices')) listen_parser.set_defaults(func=self.run) def watch_dir(self, directory): if self.watch_cmd is None: name = platform.system() if name == 'Linux': try: cmd = [ 'which', 'inotifywait' ] logging.info(' '.join(cmd)) out = subprocess.check_output(cmd) except subprocess.CalledProcessError: logging.error('"listen" command is only useful if you have installed inotifywait on Linux. Run "apt-get install inotify-tools" or equivalent to install it.') return False self.watch_cmd = [ 'inotifywait', '-r', '-e', 'modify,close_write,move,create,delete', # Only listen for events that matter, to avoid triggering constantly from the editor watching files directory ] elif name == 'Darwin': try: cmd = [ 'which', 'fswatch' ] logging.info(' '.join(cmd)) out = subprocess.check_output(cmd) except subprocess.CalledProcessError: logging.error('"listen" command is only useful if you have installed fswatch on Mac. Run "brew install fswatch" to install it with homebrew.') return False self.watch_cmd = [ 'fswatch', '-r', '-v', '-1', directory ] else: logging.error('"listen" command is only available on Mac and Linux.') return False logging.info(' '.join(self.watch_cmd)) subprocess.check_call(self.watch_cmd) return True def run(self, args, pids): if args.use_release: logging.info('Note that the listen command is not compatible with the ' 'release flag for iOS and iOS simulator builds. If you have ' 'installed iOS release builds, your Sky app will fail to ' 'reload while using listen.') tempdir = tempfile.mkdtemp() currdir = os.getcwd() while True: logging.info('Updating running Sky apps...') device = AndroidDevice() if device.is_connected(): # Restart the app on Android. Android does not currently restart using skyx files. cmd = [ sys.executable, os.path.abspath(__file__), 'start', '--poke' ] logging.info(' '.join(cmd)) subprocess.check_call(cmd) if args.local_build: # Currently sending to iOS only works if you are building Sky locally # since we aren't shipping the sky_snapshot binary yet. # Check if we can make a snapshot sky_snapshot_path = None if args.ios_sim_build_available: sky_snapshot_path = os.path.join(args.sky_src_path, args.ios_sim_debug_build_path, 'clang_x64', 'sky_snapshot') elif args.ios_build_available: sky_snapshot_path = os.path.join(args.sky_src_path, args.ios_debug_build_path, 'clang_x64', 'sky_snapshot') if sky_snapshot_path is not None: # If we can make a snapshot, do so and then send it to running iOS instances cmd = [ sky_snapshot_path, '--package-root=packages', '--snapshot=' + os.path.join(tempdir, 'snapshot_blob.bin'), os.path.join('lib', 'main.dart') ] logging.info(' '.join(cmd)) subprocess.check_call(cmd) os.chdir(tempdir) # Turn the snapshot into an app.skyx file cmd = [ 'zip', '-r', 'app.skyx', 'snapshot_blob.bin', 'action', 'content', 'navigation' ] logging.info(' '.join(cmd)) subprocess.check_call(cmd) os.chdir(currdir) # Copy the app.skyx to the running simulator simulator_app_documents_dir = IOSSimulator.get_simulator_app_documents_dir() if simulator_app_documents_dir is not None: cmd = [ 'cp', os.path.join(tempdir, 'app.skyx'), simulator_app_documents_dir ] logging.info(' '.join(cmd)) subprocess.check_call(cmd) # Copy the app.skyx to the attached iOS device if IOSDevice.is_connected(): IOSDevice.copy_file(SKY_SHELL_APP_ID, os.path.join(tempdir, 'app.skyx'), 'Documents/app.skyx') # Watch filesystem for changes if not self.watch_dir(currdir): return 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): cmd = [ ADB_PATH, 'shell', 'am', 'broadcast', '-a', 'org.domokit.sky.shell.TRACING_START' ] logging.info(' '.join(cmd)) subprocess.check_output(cmd) 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): cmd = [ADB_PATH, 'logcat', '-c'] logging.info(' '.join(cmd)) subprocess.check_output(cmd) cmd = [ ADB_PATH, 'shell', 'am', 'broadcast', '-a', 'org.domokit.sky.shell.TRACING_STOP' ] logging.info(' '.join(cmd)) subprocess.check_output(cmd) device_path = None is_complete = False while not is_complete: time.sleep(0.2) cmd = [ADB_PATH, 'logcat', '-d'] logging.info(' '.join(cmd)) log = subprocess.check_output(cmd) 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 logging.info('Downloading trace %s ...' % os.path.basename(device_path)) if device_path: cmd = [ADB_PATH, 'root'] logging.info(' '.join(cmd)) output = subprocess.check_output(cmd) match = re.match(r'.*cannot run as root.*', output) if match is not None: logging.error('Unable to download trace file %s\n' 'You need to be able to run adb as root ' 'on your android device' % device_path) return 2 cmd = [ADB_PATH, 'pull', device_path] logging.info(' '.join(cmd)) subprocess.check_output(cmd) cmd = [ADB_PATH, 'shell', 'rm', device_path] logging.info(' '.join(cmd)) subprocess.check_output(cmd) class SkyShellRunner(object): def _check_for_dart(self): try: cmd = [DART_PATH, '--version'] logging.info(' '.join(cmd)) subprocess.check_output(cmd, stderr=subprocess.STDOUT) except OSError: logging.error('"dart" (from the Dart SDK) not in $PATH, cannot continue.') return False return True def main(self): logging.basicConfig(format='%(levelname)s: %(message)s', level=logging.WARNING) real_path = os.path.realpath(__file__) if re.match(r'.*src\/sky\/packages\/sky\/', real_path) is not None: logging.warning('Using overridden sky package located at ' + os.path.dirname(os.path.dirname(real_path))) parser = argparse.ArgumentParser(description='Sky App Runner') parser.add_argument('--verbose', dest='verbose', action='store_true', help='Noisy logging, including all shell commands executed') parser.add_argument('--release', dest='use_release', action='store_true', help='Set this if you are building Sky locally and want to use the release build products. ' 'When set, attempts to automaticaly determine sky-src-path if sky-src-path is ' 'not set. Using this flag automatically turns on local-build as well, so you do not ' 'need to specify both. Note that --release is not compatible with the listen command ' 'on iOS devices and simulators. Not normally required.') parser.add_argument('--local-build', dest='local_build', action='store_true', help='Set this if you are building Sky locally and want to use those build products. ' 'When set, attempts to automaticaly determine sky-src-path if sky-src-path is ' 'not set. Not normally required.') parser.add_argument('--sky-src-path', dest='sky_src_path', help='Path to your Sky src directory, if you are building Sky locally. ' 'Ignored if local-build is not set. Not normally required.') parser.add_argument('--android-debug-build-path', dest='android_debug_build_path', help='Path to your Android Debug out directory, if you are building Sky locally. ' 'This path is relative to sky-src-path. Not normally required.', default='out/android_Debug/') parser.add_argument('--android-release-build-path', dest='android_release_build_path', help='Path to your Android Release out directory, if you are building Sky locally. ' 'This path is relative to sky-src-path. Not normally required.', default='out/android_Release/') parser.add_argument('--ios-debug-build-path', dest='ios_debug_build_path', help='Path to your iOS Debug out directory, if you are building Sky locally. ' 'This path is relative to sky-src-path. Not normally required.', default='out/ios_Debug/') parser.add_argument('--ios-release-build-path', dest='ios_release_build_path', help='Path to your iOS Release out directory, if you are building Sky locally. ' 'This path is relative to sky-src-path. Not normally required.', default='out/ios_Release/') parser.add_argument('--ios-sim-debug-build-path', dest='ios_sim_debug_build_path', help='Path to your iOS Simulator Debug out directory, if you are building Sky locally. ' 'This path is relative to sky-src-path. Not normally required.', default='out/ios_sim_Debug/') parser.add_argument('--ios-sim-release-build-path', dest='ios_sim_release_build_path', help='Path to your iOS Simulator Release out directory, if you are building Sky locally. ' 'This path is relative to sky-src-path. Not normally required.', default='out/ios_sim_Release/') subparsers = parser.add_subparsers(help='sub-command help') for command in [SkyLogs(), InstallSky(), StartSky(), StopSky(), StartListening(), StartTracing(), StopTracing(), IOSSimulator()]: command.add_subparser(subparsers) args = parser.parse_args() if args.verbose: logging.getLogger().setLevel(logging.INFO) if args.use_release: args.local_build = True # TODO(iansf): args is unfortunately just a global context variable. For now, add some additional context to it. args.android_build_available = False args.ios_build_available = False args.ios_sim_build_available = False # Also make sure that args is consistent with machine state for local builds if args.local_build and args.sky_src_path is None: real_sky_path = os.path.realpath(os.path.join(PACKAGES_DIR, 'sky')) match = re.match(r'pub.dartlang.org/sky', real_sky_path) if match is not None: args.local_build = False else: sky_src_path = os.path.dirname( os.path.dirname( os.path.dirname( os.path.dirname(real_sky_path)))) if sky_src_path == '/' or sky_src_path == '': args.local_build = False else: args.sky_src_path = sky_src_path if not args.local_build: logging.warning('Unable to detect a valid sky install. Disabling local-build flag.\n' 'The recommended way to use a local build of Sky is to add the following\n' 'to your pubspec.yaml file and then run pub get again:\n' 'dependency_overrides:\n' ' material_design_icons:\n' ' path: /path/to/sky_engine/src/sky/packages/material_design_icons\n' ' sky:\n' ' path: /path/to/sky_engine/src/sky/packages/sky\n') if args.local_build: if not os.path.isdir(args.sky_src_path): logging.warning('The selected sky-src-path (' + args.sky_src_path + ') does not exist.' 'Disabling local-build flag.') args.local_build = False if args.local_build and args.use_release: if os.path.isdir(os.path.join(args.sky_src_path, args.android_release_build_path)): args.android_build_available = True if os.path.isdir(os.path.join(args.sky_src_path, args.ios_release_build_path)): args.ios_build_available = True if os.path.isdir(os.path.join(args.sky_src_path, args.ios_sim_release_build_path)): args.ios_sim_build_available = True elif args.local_build: if os.path.isdir(os.path.join(args.sky_src_path, args.android_debug_build_path)): args.android_build_available = True if os.path.isdir(os.path.join(args.sky_src_path, args.ios_debug_build_path)): args.ios_build_available = True if os.path.isdir(os.path.join(args.sky_src_path, args.ios_sim_debug_build_path)): args.ios_sim_build_available = True else: if os.path.isdir(APK_DIR): args.android_build_available = True if not self._check_for_dart(): sys.exit(2) pids = Pids.read_from(PID_FILE_PATH, PID_FILE_KEYS) atexit.register(pids.write_to, PID_FILE_PATH) exit_code = 0 try: exit_code = args.func(args, pids) except subprocess.CalledProcessError as e: # Don't print a stack trace if the adb command fails. logging.error(e) exit_code = 2 sys.exit(exit_code) if __name__ == '__main__': sys.exit(SkyShellRunner().main())