import os import sys import re import time import tempfile import shutil import threading from wlauto.core.device_manager import DeviceManager from wlauto import Parameter, Alias from wlauto.utils.types import boolean, regex from wlauto.utils.android import adb_command from wlauto.exceptions import WorkerThreadError from devlib.target import AndroidTarget SCREEN_STATE_REGEX = re.compile('(?:mPowerState|mScreenOn|Display Power: state)=([0-9]+|true|false|ON|OFF)', re.I) SCREEN_SIZE_REGEX = re.compile(r'mUnrestrictedScreen=\(\d+,\d+\)\s+(?P\d+)x(?P\d+)') class AndroidDevice(DeviceManager): name = "android" target_type = AndroidTarget aliases = [ Alias('generic_android'), ] parameters = [ Parameter('adb_name', default=None, kind=str, description='The unique ID of the device as output by "adb devices".'), Parameter('android_prompt', kind=regex, default=re.compile('^.*(shell|root)@.*:/\S* [#$] ', re.MULTILINE), # ## description='The format of matching the shell prompt in Android.'), Parameter('working_directory', default='/sdcard/wa-working', override=True), Parameter('binaries_directory', default='/data/local/tmp', override=True), Parameter('package_data_directory', default='/data/data', description='Location of of data for an installed package (APK).'), Parameter('external_storage_directory', default='/sdcard', description='Mount point for external storage.'), Parameter('logcat_poll_period', kind=int, description=""" If specified and is not ``0``, logcat will be polled every ``logcat_poll_period`` seconds, and buffered on the host. This can be used if a lot of output is expected in logcat and the fixed logcat buffer on the device is not big enough. The trade off is that this introduces some minor runtime overhead. Not set by default. """), # ## Parameter('enable_screen_check', kind=boolean, default=False, description=""" Specified whether the device should make sure that the screen is on during initialization. """), Parameter('swipe_to_unlock', kind=str, default=None, allowed_values=[None, "horizontal", "vertical"], description=""" If set a swipe of the specified direction will be performed. This should unlock the screen. """), # ## ] def __init__(self, **kwargs): super(AndroidDevice, self).__init__(**kwargs) self.connection_settings = self._make_connection_settings() self.platform = self.platform_type(core_names=self.core_names, # pylint: disable=E1102 core_clusters=self.core_clusters) self.target = self.target_type(connection_settings=self.connection_settings, connect=False, platform=self.platform, working_directory=self.working_directory, executables_directory=self.binaries_directory,) self._logcat_poller = None def connect(self): self.target.connect() def initialize(self, context): super(AndroidDevice, self).initialize(context) if self.enable_screen_check: self.target.ensure_screen_is_on() if self.swipe_to_unlock: self.target.swipe_to_unlock(direction=self.swipe_to_unlock) def start(self): if self.logcat_poll_period: if self._logcat_poller: self._logcat_poller.close() self._logcat_poller = _LogcatPoller(self, self.logcat_poll_period, timeout=self.default_timeout) self._logcat_poller.start() else: self.target.clear_logcat() def _make_connection_settings(self): connection_settings = {} connection_settings['device'] = self.adb_name return connection_settings def dump_logcat(self, outfile, filter_spec=None): """ Dump the contents of logcat, for the specified filter spec to the specified output file. See http://developer.android.com/tools/help/logcat.html :param outfile: Output file on the host into which the contents of the log will be written. :param filter_spec: Logcat filter specification. see http://developer.android.com/tools/debugging/debugging-log.html#filteringOutput """ if self._logcat_poller: return self._logcat_poller.write_log(outfile) else: if filter_spec: command = 'logcat -d -s {} > {}'.format(filter_spec, outfile) else: command = 'logcat -d > {}'.format(outfile) return adb_command(self.adb_name, command) class _LogcatPoller(threading.Thread): join_timeout = 5 def __init__(self, target, period, timeout=None): super(_LogcatPoller, self).__init__() self.target = target self.logger = target.logger self.period = period self.timeout = timeout self.stop_signal = threading.Event() self.lock = threading.RLock() self.buffer_file = tempfile.mktemp() self.last_poll = 0 self.daemon = True self.exc = None def run(self): self.logger.debug('Starting logcat polling.') try: while True: if self.stop_signal.is_set(): break with self.lock: current_time = time.time() if (current_time - self.last_poll) >= self.period: self._poll() time.sleep(0.5) except Exception: # pylint: disable=W0703 self.exc = WorkerThreadError(self.name, sys.exc_info()) self.logger.debug('Logcat polling stopped.') def stop(self): self.logger.debug('Stopping logcat polling.') self.stop_signal.set() self.join(self.join_timeout) if self.is_alive(): self.logger.error('Could not join logcat poller thread.') if self.exc: raise self.exc # pylint: disable=E0702 def clear_buffer(self): self.logger.debug('Clearing logcat buffer.') with self.lock: self.target.clear_logcat() with open(self.buffer_file, 'w') as _: # NOQA pass def write_log(self, outfile): self.logger.debug('Writing logbuffer to {}.'.format(outfile)) with self.lock: self._poll() if os.path.isfile(self.buffer_file): shutil.copy(self.buffer_file, outfile) else: # there was no logcat trace at this time with open(outfile, 'w') as _: # NOQA pass def close(self): self.logger.debug('Closing logcat poller.') if os.path.isfile(self.buffer_file): os.remove(self.buffer_file) def _poll(self): with self.lock: self.last_poll = time.time() self.target.dump_logcat(self.buffer_file, append=True)