<file_sep># coding=utf-8
# Copyright 2021 DeepMind Technologies Limited.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

"""A utility class to run a function in a separate daemon thread."""

import abc
import enum
import queue
import threading
from typing import Optional

from absl import logging


class ThreadFunction(metaclass=abc.ABCMeta):
  """Base class that encapsulates long-lived functions in a separate thread."""

  class Signal(enum.IntEnum):
    """Defines commands we can use to communicate with the internal thread."""
    # The thread should be stopped to allow for graceful termination.
    KILL = 1

  def __init__(self, block_input: bool, block_output: bool, name: str):
    """Initializes this ThreadFunction.

    Args:
      block_input: Whether to block this thread when reading its input queue.
      block_output: Whether to block this thread when writing to its
          output queue.
      name: Name of the thread. Used to keep track of threads in logging.
    """
    self._block_input = block_input
    self._block_output = block_output
    self._name = name

    self._input_queue = queue.Queue()
    self._output_queue = queue.Queue()
    self._should_run = True
    self._internal_thread = threading.Thread(target=self._run)
    self._internal_thread.daemon = True
    self._internal_thread.start()

  def read(self, block: bool = True, timeout: Optional[float] = None):
    """'Public' method for clients to read values _from_ this thread.

    Args:
      block: Whether the client should block.
      timeout: Timeout for getting output from the queue, in seconds.
    Returns:
      The value produced by the underlying thread.
    """
    try:
      return self._output_queue.get(block=block, timeout=timeout)
    except queue.Empty:
      return None

  def write(self, value, block: bool = True, timeout: Optional[float] = None):
    """'Public' method for clients to write values _to_ this thread.

    Args:
      value: The value to send to the underlying thread.
      block: Whether the client should block.
      timeout: Timeout for setting input in the queue, in seconds.
    Returns:
      The value produced by the underlying thread.
    """
    self._input_queue.put(value, block=block, timeout=timeout)

  @abc.abstractmethod
  def main(self):
    """main() function that subclasses need to override."""
    pass

  def kill(self):
    """Shorthand for clients to terminate this thread."""
    logging.info('Killing %s thread', self._name)
    # Sending a kill signal to clean up blocked read_values.
    self.write(ThreadFunction.Signal.KILL, block=True)
    # Stopping the _run loop.
    self._should_run = False

  def _run(self):
    """'Private' method that reruns main() until explicit termination."""
    logging.info('Starting %s thread.', self._name)
    while self._should_run:
      self.main()
    logging.info('Finished %s thread.', self._name)

  def _read_value(self):
    """'Protected' method for subclasses to read values from their input."""
    try:
      return self._input_queue.get(block=self._block_input)
    except queue.Empty:
      pass  # Ignore empty queues. Keep going.

  def _write_value(self, value):
    """'Protected' method for subclasses to write values to their output."""
    self._output_queue.put(value, block=self._block_output)
<file_sep># coding=utf-8
# Copyright 2021 DeepMind Technologies Limited.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

"""Tests for android_env.components.logcat_thread."""

import re
import subprocess
import threading
from typing import Match, Pattern

from absl.testing import absltest
from android_env.components import logcat_thread
from android_env.proto import task_pb2
import mock


class FakeStream():
  """This class simulates the logs coming from ADB."""

  def __init__(self):
    self._values = []
    self._kill = False
    self._lock = threading.Lock()

  def send_value(self, value):
    with self._lock:
      self._values.append(value)

  def has_next_value(self):
    return bool(self._values)

  def kill(self):
    self._kill = True

  def __iter__(self):
    while True:
      if self._kill:
        return
      if not self._values:
        continue
      else:
        with self._lock:
          next_value = self._values.pop(0)
        yield next_value


def _log_parsing_config():
  """Returns log_parsing_config object for testing."""

  return task_pb2.LogParsingConfig(
      filters=['AndroidRLTask:V'])


class FakeProc():
  """Fake process that exposes a fake stdout stream."""

  def __init__(self):
    self.stdout = FakeStream()
    self.is_alive = True

  def kill(self):
    self.stdout.kill()
    self.is_alive = False


def make_stdout(data):
  """Returns a valid log output with given data as message."""
  return '         1553110400.424  5583  5658 D Tag: %s' % data


class LogcatThreadTest(absltest.TestCase):

  def setUp(self):
    super().setUp()
    self.mock_popen = self.enter_context(
        mock.patch.object(subprocess, 'Popen', autospec=True))
    self.fake_proc = FakeProc()
    self.mock_popen.return_value = self.fake_proc

  def tearDown(self):
    self.fake_proc.stdout.kill()
    super().tearDown()

  def test_cmd(self):
    _ = logcat_thread.LogcatThread(
        adb_command_prefix=['adb_bin', '-P', '12345', '-s', 'my_device'],
        log_parsing_config=_log_parsing_config())
    expected_cmd = [
        'adb_bin', '-P', '12345', '-s', 'my_device', 'logcat', '-v', 'epoch',
        'AndroidRLTask:V', '*:S',
    ]
    self.mock_popen.assert_called_once_with(
        expected_cmd,
        stdout=subprocess.PIPE,
        stderr=subprocess.STDOUT,
        bufsize=1,
        universal_newlines=True)

  def test_cmd_with_filters(self):
    _ = logcat_thread.LogcatThread(
        adb_command_prefix=['adb', '-P', '5037'],
        log_parsing_config=_log_parsing_config())
    expected_cmd = [
        'adb', '-P', '5037', 'logcat', '-v', 'epoch', 'AndroidRLTask:V', '*:S'
    ]
    self.mock_popen.assert_called_once_with(
        expected_cmd,
        stdout=subprocess.PIPE,
        stderr=subprocess.STDOUT,
        bufsize=1,
        universal_newlines=True)

  def test_kill(self):
    logcat = logcat_thread.LogcatThread(
        adb_command_prefix=['adb', '-P', '5037'],
        log_parsing_config=_log_parsing_config())
    self.assertTrue(self.fake_proc.is_alive)
    logcat.kill()
    self.assertFalse(self.fake_proc.is_alive)

  def test_listeners(self):
    """Ensures that we can wait for a specific message without polling."""
    logcat = logcat_thread.LogcatThread(
        adb_command_prefix=['adb', '-P', '5037'],
        log_parsing_config=_log_parsing_config(),
        print_all_lines=True)
    self.mock_popen.assert_called_once()

    # Set up a listener that modifies an arbitrary state.
    some_state = False

    def my_handler(event: Pattern[str], match: Match[str]):
      del event, match
      nonlocal some_state
      some_state = True

    # Create a desired event and hook up the listener.
    my_event = re.compile('Hello world')
    listener = logcat_thread.EventListener(my_event, my_handler)
    logcat.add_event_listener(listener)
    self.fake_proc.stdout.send_value('Hi there!')  # This should not match.
    self.assertFalse(some_state)
    self.fake_proc.stdout.send_value(make_stdout('Hello world'))
    logcat.wait(event=my_event, timeout_sec=1.0)
    self.assertTrue(some_state)

    # Waiting for any events should also trigger the listener.
    some_state = False
    self.fake_proc.stdout.send_value(make_stdout('Hello world'))
    logcat.wait(event=None, timeout_sec=1.0)
    self.assertTrue(some_state)

    # After removing the listener, it should not be called anymore.
    some_state = False
    logcat.remove_event_listener(listener)
    self.fake_proc.stdout.send_value(make_stdout('Hello world'))
    logcat.wait(event=my_event, timeout_sec=1.0)
    self.assertFalse(some_state)


if __name__ == '__main__':
  absltest.main()
<file_sep># coding=utf-8
# Copyright 2021 DeepMind Technologies Limited.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

"""Tests for android_env.components.task_manager.py."""

import json

from absl.testing import absltest
from android_env.components import adb_controller
from android_env.components import dumpsys_thread
from android_env.components import logcat_thread
from android_env.components import setup_step_interpreter
from android_env.components import task_manager
from android_env.proto import task_pb2
import mock
import numpy as np


class TaskManagerTest(absltest.TestCase):

  def setUp(self):
    super().setUp()
    self.addCleanup(mock.patch.stopall)  # Disable previous patches.

    self._adb_controller = mock.create_autospec(adb_controller.AdbController)
    self._setup_step_interpreter = mock.create_autospec(
        setup_step_interpreter.SetupStepInterpreter)
    self._dumpsys_thread = mock.create_autospec(dumpsys_thread.DumpsysThread)
    self._logcat_thread = mock.create_autospec(logcat_thread.LogcatThread)

    mock.patch.object(
        adb_controller, 'AdbController',
        return_value=self._adb_controller).start()
    mock.patch.object(
        setup_step_interpreter,
        'SetupStepInterpreter',
        return_value=self._setup_step_interpreter).start()
    mock.patch.object(
        dumpsys_thread, 'DumpsysThread',
        return_value=self._dumpsys_thread).start()
    mock.patch.object(
        logcat_thread, 'LogcatThread',
        return_value=self._logcat_thread).start()

  def test_setup_task(self):
    task_mgr = task_manager.TaskManager(task=task_pb2.Task())
    task_mgr.setup_task(adb_controller=self._adb_controller)
    assert hasattr(task_mgr, '_logcat_thread')
    assert hasattr(task_mgr, '_setup_step_interpreter')

  def test_get_current_reward(self):
    # Replace `LogcatThread.add_event_listener` with one that simply calls `fn`
    # right away.
    def my_add_ev_listener(event_listener: logcat_thread.EventListener):
      # Check that the event matches what's expected.
      match = event_listener.regexp.match('Reward: 123.0')
      if match is None:  # Ignore events that are not rewards.
        return

      event_listener.handler_fn(event_listener.regexp, match)

    task = task_pb2.Task()
    task.log_parsing_config.log_regexps.reward.extend([
        '^[Rr]eward: ([-+]?[0-9]*\\.?[0-9]*)$'
    ])
    task_mgr = task_manager.TaskManager(task=task)
    self._logcat_thread.add_event_listener.side_effect = my_add_ev_listener
    task_mgr.setup_task(adb_controller=self._adb_controller)
    self.assertEqual(task_mgr.get_current_reward(), 123.0)

  def test_reward_event(self):
    # Replace `LogcatThread.add_event_listener` with one that simply calls `fn`
    # right away.
    def my_add_ev_listener(event_listener: logcat_thread.EventListener):
      # Check that the event matches what's expected.
      match_1 = event_listener.regexp.match('foo_1')
      match_2 = event_listener.regexp.match('foo_2')
      match_3 = event_listener.regexp.match('Reward: 2.0')
      if match_1:
        event_listener.handler_fn(event_listener.regexp, match_1)
      if match_2:
        event_listener.handler_fn(event_listener.regexp, match_2)
      if match_3:
        event_listener.handler_fn(event_listener.regexp, match_3)

    task = task_pb2.Task()
    reward_event_1 = task_pb2.LogParsingConfig.LogRegexps.RewardEvent(
        event='foo_1', reward=5.0)
    reward_event_2 = task_pb2.LogParsingConfig.LogRegexps.RewardEvent(
        event='foo_2', reward=-1.0)
    task.log_parsing_config.log_regexps.reward_event.extend(
        [reward_event_1, reward_event_2])
    task.log_parsing_config.log_regexps.reward.extend(
        ['^[Rr]eward: ([-+]?[0-9]*\\.?[0-9]*)$'])
    task_mgr = task_manager.TaskManager(task=task)
    self._logcat_thread.add_event_listener.side_effect = my_add_ev_listener
    task_mgr.setup_task(adb_controller=self._adb_controller)
    self.assertEqual(task_mgr.get_current_reward(), 6.0)

  def test_get_current_reward_via_score(self):
    # Replace `LogcatThread.add_event_listener` with one that simply calls `fn`
    # right away.
    def my_add_ev_listener(event_listener: logcat_thread.EventListener):
      # Check that the event matches what's expected.
      event = event_listener.regexp
      match = event.match('score: 200.0')
      if match is None:  # Ignore events that are not scores.
        return

      event_listener.handler_fn(event, match)

      # Scores are accumulated by their differences, so a subsequent lower score
      # means that the final reward decreases.
      match = event.match('score: 185')
      event_listener.handler_fn(event, match)

    task = task_pb2.Task()
    task.log_parsing_config.log_regexps.score = (
        '^score: ([-+]?[0-9]*\\.?[0-9]*)$')
    task_mgr = task_manager.TaskManager(task=task)
    self._logcat_thread.add_event_listener.side_effect = my_add_ev_listener
    task_mgr.setup_task(adb_controller=self._adb_controller)
    self.assertEqual(task_mgr.get_current_reward(), 185.0)

  def test_get_current_extras(self):
    # Replace `LogcatThread.add_event_listener` with one that simply calls `fn`
    # right away.
    def my_add_ev_listener(event_listener: logcat_thread.EventListener):
      # Check that the event matches what's expected.
      event = event_listener.regexp
      match = event.match('extra: some_extra [1, 2]')
      if match is None:  # Ignore events that are not extras.
        return

      # Emit events.
      fn = event_listener.handler_fn
      fn(event, event.match('extra: an_extra [1, 2, 3]'))
      fn(event, event.match('extra: an_extra [4, 5, 6]'))
      fn(event, event.match('extra: another_extra 0.5'))
      fn(event, event.match('extra: multi_dimension_extra [[9,8,7],[6,5,4]]'))
      fn(event, event.match('extra: boolean_extra'))

    # Setup the task and trigger the listener.
    task = task_pb2.Task()
    task.log_parsing_config.log_regexps.extra.extend([
        '^extra: (?P<name>[^ ]*)[ ]?(?P<extra>.*)$'
    ])
    task_mgr = task_manager.TaskManager(task=task)
    self._logcat_thread.add_event_listener.side_effect = my_add_ev_listener
    task_mgr.setup_task(adb_controller=self._adb_controller)

    # Check expectations.
    extras = task_mgr.get_current_extras()
    np.testing.assert_almost_equal([[1, 2, 3], [4, 5, 6]],
                                   extras.get('an_extra'))
    np.testing.assert_almost_equal([0.5], extras.get('another_extra'))
    np.testing.assert_almost_equal([[[9, 8, 7], [6, 5, 4]]],
                                   extras.get('multi_dimension_extra'))
    np.testing.assert_equal([1], extras.get('boolean_extra'))

  def test_get_current_extras_json_format(self):
    # Replace `LogcatThread.add_event_listener` with one that simply calls `fn`
    # right away.
    def my_add_ev_listener(event_listener: logcat_thread.EventListener):
      # Check that the event matches what's expected.
      event = event_listener.regexp
      match = event.match('json_extra: {}')
      if match is None:  # Ignore events that are not extras.
        return

      # Emit events.
      extra = {
          'extra_scalar': 0,
          'extra_list': [1, 2, 3, 4],
          'extra_dict': {
              'foo': 'bar'
          },
          'extra_string': 'a_string'
      }
      extra_update = {'extra_string': 'a_new_string', 'extra_float': 0.6}
      fn = event_listener.handler_fn
      fn(event, event.match(f'json_extra: {json.dumps(extra)}'))
      fn(event, event.match(f'json_extra: {json.dumps(extra_update)}'))

    # Setup the task and trigger the listener.
    task = task_pb2.Task()
    task.log_parsing_config.log_regexps.json_extra.extend([
        '^json_extra: (?P<json_extra>.*)$'
    ])
    task_mgr = task_manager.TaskManager(task=task)
    self._logcat_thread.add_event_listener.side_effect = my_add_ev_listener
    task_mgr.setup_task(adb_controller=self._adb_controller)

    # Check expectations.
    expected_extra = {
        'extra_scalar': [0],
        'extra_list': [[1, 2, 3, 4]],
        'extra_dict': [{
            'foo': 'bar'
        }],
        'extra_string': ['a_string', 'a_new_string'],
        'extra_float': [0.6]
    }
    extras = task_mgr.get_current_extras()
    np.testing.assert_almost_equal(
        expected_extra.get('extra_scalar'), extras.get('extra_scalar'))
    np.testing.assert_almost_equal(
        expected_extra.get('extra_list'), extras.get('extra_list'))
    np.testing.assert_equal(
        expected_extra.get('extra_string'), extras.get('extra_string'))
    np.testing.assert_almost_equal(
        expected_extra.get('extra_float'), extras.get('extra_float'))
    np.testing.assert_equal(
        expected_extra.get('extra_dict'), extras.get('extra_dict'))

  def test_multi_log_regexp(self):
    # Replace `LogcatThread.add_event_listener` with one that simply calls `fn`
    # right away.
    def my_add_ev_listener(event_listener: logcat_thread.EventListener):
      # Check that the event matches what's expected.
      match = event_listener.regexp.match('Reward_2: 123.0')
      if match is None:  # Ignore events that are not rewards.
        return

      event_listener.handler_fn(event_listener.regexp, match)

    task = task_pb2.Task()
    task.log_parsing_config.log_regexps.reward.extend([
        '^[Rr]eward_1: ([-+]?[0-9]*\\.?[0-9]*)$',
        '^[Rr]eward_2: ([-+]?[0-9]*\\.?[0-9]*)$'
    ])
    task_mgr = task_manager.TaskManager(task=task)
    self._logcat_thread.add_event_listener.side_effect = my_add_ev_listener
    task_mgr.setup_task(adb_controller=self._adb_controller)
    self.assertEqual(task_mgr.get_current_reward(), 123.0)

  def test_multi_reward_regexp(self):
    # Replace `LogcatThread.add_event_listener` with one that simply calls `fn`
    # right away.'

    def my_add_ev_listener(event_listener: logcat_thread.EventListener):
      # Check that the event matches what's expected.
      match_1 = event_listener.regexp.match('Reward_1: 5.0')
      match_2 = event_listener.regexp.match('Reward_2: 10.0')

      if match_1:
        event_listener.handler_fn(event_listener.regexp, match_1)

      if match_2:
        event_listener.handler_fn(event_listener.regexp, match_2)

    task = task_pb2.Task()
    task.log_parsing_config.log_regexps.reward.extend([
        '^[Rr]eward_1: ([-+]?[0-9]*\\.?[0-9]*)$',
        '^[Rr]eward_2: ([-+]?[0-9]*\\.?[0-9]*)$',
    ])
    task_mgr = task_manager.TaskManager(task=task)
    self._logcat_thread.add_event_listener.side_effect = my_add_ev_listener
    task_mgr.setup_task(adb_controller=self._adb_controller)
    self.assertEqual(task_mgr.get_current_reward(), 15.0)

  def test_check_episode_end(self):
    # Replace `LogcatThread.add_event_listener` with one that simply calls `fn`
    # right away.
    def my_add_ev_listener(event_listener: logcat_thread.EventListener):
      # Check that the event matches what's expected.
      event = event_listener.regexp
      match = event.match('I am done!')
      if match is None:  # Ignore events that are not episode end.
        return

      event_listener.handler_fn(event, match)

    task = task_pb2.Task()
    task.log_parsing_config.log_regexps.episode_end.extend(['I am done!'])
    task_mgr = task_manager.TaskManager(task=task)
    self._logcat_thread.add_event_listener.side_effect = my_add_ev_listener
    task_mgr.setup_task(adb_controller=self._adb_controller)
    episode_end = task_mgr.check_if_episode_ended()
    self.assertTrue(episode_end)

if __name__ == '__main__':
  absltest.main()
<file_sep># coding=utf-8
# Copyright 2021 DeepMind Technologies Limited.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

"""Tests for android_env.components.emulator_launcher."""

import os
from absl.testing import absltest
from android_env.components import emulator_launcher
import mock
from pexpect import popen_spawn
import builtins


class EmulatorLauncherTest(absltest.TestCase):

  def setUp(self):
    super().setUp()

    self._emulator_path = 'fake/path/emulator'
    self._adb_port = 5554
    self._adb_server_port = 1234
    self._emulator_console_port = 5555
    self._avd_name = 'my_avd_name'

    self._emulator = mock.create_autospec(popen_spawn.PopenSpawn)
    self._emulator.after = 'after'
    self._emulator_output = mock.create_autospec(open)
    self._emulator_output.close = lambda: None

    self._expected_command = [
        self._emulator_path,
        '-no-snapshot',
        '-gpu', 'swiftshader_indirect',
        '-no-audio',
        '-verbose',
        '-avd', self._avd_name,
    ]
    self._ports = ['-ports', f'{self._emulator_console_port},{self._adb_port}']

    base_lib_dir = self._emulator_path[:-8] + 'lib64/'
    ld_library_path = ':'.join([
        base_lib_dir + 'x11/',
        base_lib_dir + 'qt/lib/',
        base_lib_dir + 'gles_swiftshader/',
        base_lib_dir
    ])

    self._expected_env_vars = {
        'ANDROID_HOME': '',
        'ANDROID_SDK_ROOT': '',
        'ANDROID_AVD_HOME': '',
        'ANDROID_EMULATOR_KVM_DEVICE': '/dev/kvm',
        'ANDROID_ADB_SERVER_PORT': '1234',
        'LD_LIBRARY_PATH': ld_library_path,
        'QT_DEBUG_PLUGINS': '1',
        'QT_XKB_CONFIG_ROOT': str(self._emulator_path[:-8] + 'qt_config/'),
    }

  @mock.patch.object(os, 'environ', autospec=True, return_value=dict())
  def test_launch(self, os_environ):
    del os_environ

    launcher = emulator_launcher.EmulatorLauncher(
        adb_port=self._adb_port,
        adb_server_port=self._adb_server_port,
        emulator_console_port=self._emulator_console_port,
        emulator_path=self._emulator_path,
        avd_name=self._avd_name,
        grpc_port=-1)

    with mock.patch.object(
       popen_spawn, 'PopenSpawn', autospec=True,
       return_value=self._emulator) as emulator_init, \
       mock.patch.object(
           builtins, 'open', autospec=True,
           return_value=self._emulator_output):

      launcher.launch()
      emulator_init.assert_called_once_with(
          cmd=self._expected_command + self._ports,
          logfile=self._emulator_output,
          env=self._expected_env_vars)

  @mock.patch.object(os, 'environ', autospec=True, return_value=dict())
  def test_grpc_port(self, os_environ):
    del os_environ

    launcher = emulator_launcher.EmulatorLauncher(
        adb_port=self._adb_port,
        adb_server_port=self._adb_server_port,
        emulator_console_port=self._emulator_console_port,
        emulator_path=self._emulator_path,
        avd_name=self._avd_name,
        grpc_port=8554)

    with mock.patch.object(
       popen_spawn, 'PopenSpawn', autospec=True,
       return_value=self._emulator) as emulator_init, \
       mock.patch.object(
           builtins, 'open', autospec=True,
           return_value=self._emulator_output):
      launcher.launch()
      emulator_init.assert_called_once_with(
          cmd=self._expected_command + ['-grpc', '8554'] + self._ports,
          logfile=self._emulator_output,
          env=self._expected_env_vars)

  @mock.patch.object(os, 'environ', autospec=True, return_value=dict())
  def test_restart(self, os_environ):
    del os_environ

    launcher = emulator_launcher.EmulatorLauncher(
        adb_port=self._adb_port,
        adb_server_port=self._adb_server_port,
        emulator_console_port=self._emulator_console_port,
        emulator_path=self._emulator_path,
        avd_name=self._avd_name,
        grpc_port=-1)

    with mock.patch.object(
       popen_spawn, 'PopenSpawn', autospec=True,
       return_value=self._emulator) as emulator_init, \
       mock.patch.object(
           builtins, 'open', autospec=True,
           return_value=self._emulator_output):
      launcher.launch()
      launcher.restart()
      self._emulator.kill.assert_called_once()
      emulator_init.assert_has_calls([mock.call(
          cmd=self._expected_command + self._ports,
          logfile=self._emulator_output,
          env=self._expected_env_vars)]*2)


if __name__ == '__main__':
  absltest.main()
<file_sep># coding=utf-8
# Copyright 2021 DeepMind Technologies Limited.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

"""Extends Android observation with the latest action taken."""

from typing import Dict

from android_env.components import action_type
from android_env.components import utils
from android_env.wrappers import base_wrapper
import dm_env
from dm_env import specs
import numpy as np


class LastActionWrapper(base_wrapper.BaseWrapper):
  """Extends Android observations with information about the last action taken.

  The position of the last action is denoted by a single white pixel (with a
  value of 255) in a channel of all black pixels (with a value of 0).
  As this wrapper makes use of temporarily stored information about the
  last action taken, it is important to apply on the environment side rather
  than the agent side. Recommended not to apply before an ImageRescaleWrapper,
  to avoid distortion of the single pixel denoting the action position.
  """

  def __init__(self,
               env: dm_env.Environment,
               concat_to_pixels: bool = True):
    """Initializes the internal state of this wrapper.

    Args:
      env: the environment to wrap.
      concat_to_pixels: If True, will add a channel to the pixel observation.
        If False, will pass the action as an extra observation.
    """
    super().__init__(env)
    self._concat_to_pixels = concat_to_pixels
    self._screen_dimensions = self._env.observation_spec()['pixels'].shape[:2]

  def _process_timestep(self, timestep: dm_env.TimeStep) -> dm_env.TimeStep:
    observation = timestep.observation.copy()
    processed_observation = self._process_observation(observation)
    return timestep._replace(observation=processed_observation)

  def _process_observation(
      self, observation: Dict[str, np.ndarray]
  ) -> Dict[str, np.ndarray]:
    """Extends observation with last_action data."""
    processed_observation = observation.copy()
    last_action_layer = self._get_last_action_layer(observation['pixels'])
    if self._concat_to_pixels:
      pixels = observation['pixels'].copy()
      processed_pixels = np.dstack((pixels, last_action_layer))
      processed_observation['pixels'] = processed_pixels
    else:
      processed_observation['last_action'] = last_action_layer
    return processed_observation

  def _get_last_action_layer(self, pixels: np.ndarray) -> np.ndarray:
    """Makes sure the rescaling doesn't distort the last_action layer."""

    last_action = self._env.raw_action
    last_action_layer = np.zeros(self._screen_dimensions, dtype=pixels.dtype)

    if ('action_type' in last_action and
        last_action['action_type'] == action_type.ActionType.TOUCH):
      touch_position = last_action['touch_position']
      x, y = utils.touch_position_to_pixel_position(
          touch_position, width_height=self._screen_dimensions[::-1])
      last_action_layer[y, x] = 255

    return last_action_layer

  def reset(self) -> dm_env.TimeStep:
    timestep = self._env.reset()
    return self._process_timestep(timestep)

  def step(self, action) -> dm_env.TimeStep:
    timestep = self._env.step(action)
    return self._process_timestep(timestep)

  def observation_spec(self) -> Dict[str, specs.Array]:
    parent_spec = self._env.observation_spec().copy()
    shape = parent_spec['pixels'].shape
    if self._concat_to_pixels:
      parent_spec['pixels'] = specs.Array(
          shape=(shape[0], shape[1], shape[2] + 1),
          dtype=parent_spec['pixels'].dtype,
          name=parent_spec['pixels'].name)
    else:
      parent_spec.update({
          'last_action':
              specs.Array(
                  shape=(shape[0], shape[1]),
                  dtype=parent_spec['pixels'].dtype,
                  name='last_action')
      })
    return parent_spec
<file_sep># AndroidEnv - Available tasks

This page gives a detailed overview of the example tasks provided with
AndroidEnv. The purpose is to give researchers an idea of the different kinds of
challenges that AndroidEnv poses.

To use any of these tasks in your own experiments, click on **Download** to
download a ZIP file containing textprotos and the corresponding APKs. After
downloading, move the `.apk` and `.textproto` files to a directory of your
choice and take note of their path. This information is needed for
[running](instructions.md#create-the-env) an AndroidEnv instance with the given
task.

<!-- mdformat off(multi-line tables are not supported well) -->

| App / Game                                          | Interface            | Time reactive  | Multi-level                   |  Rewards   | Extras  | Download |
| --------------------------------------------------- | -------------------- | -------------- | ----------------------------- | ---------- | ------------ | -------- |
| [Vokram (MDP)](#vokram)                             | Tapping (buttons)    | No             | Yes (4 variants)              | Dense      | Yes          | [Download](https://storage.googleapis.com/android_env-tasks/mdp.tar.gz) |
| [Accessibility Forwarder](#accessibility-forwarder)  | Tapping (buttons)    | No             | No                            | Sparse     | Yes          | [Download](https://storage.googleapis.com/android_env-tasks/accessibility_forwarder.tar.gz) |
| [Apple Flinger](#apple-flinger)                     | Drag & drop          | No             | Yes (6 variants)              | Dense      | No           | [Download](https://storage.googleapis.com/android_env-tasks/apple_flinger.tar.gz) |
| [Blockinger](#blockinger)                           | Tapping (buttons)    | Yes            | No                            | Sparse     | Yes          | [Download](https://storage.googleapis.com/android_env-tasks/blockinger.tar.gz) |
| [Catch](#catch)                                     | Touch                | Yes            | No                            | Dense      | Yes          | [Download](https://storage.googleapis.com/android_env-tasks/catch_the_ball.tar.gz) |
| [Classic 2048](#classic-2048)                       | Swiping              | No             | No                            | Dense      | Yes          | [Download](https://storage.googleapis.com/android_env-tasks/classic_2048.tar.gz) |
| [Dodge](#dodge)                                     | Tapping              | Yes            | No                            | Sparse     | Yes          | [Download](https://storage.googleapis.com/android_env-tasks/dodge.tar.gz)  |
| [DroidFish (Chess)](#droidfish)                     | Tapping              | No             | Yes (3 levels)                | Sparse     | Yes          | [Download](https://storage.googleapis.com/android_env-tasks/droidfish.tar.gz) |
| [FlappyDroid](#flappydroid)                         | Tapping              | Yes            | Yes (2 levels)                | Dense      | No           | [Download](https://storage.googleapis.com/android_env-tasks/systemui_egg_land.tar.gz)  |
| [Frozen Bubble](#frozen-bubble)                     | Dragging, tapping    | No             | No                            | Sparse     | No           | [Download](https://storage.googleapis.com/android_env-tasks/frozen_bubble.tar.gz) |
| [Memory Game](#memory-game)                         | Tapping              | No             | Yes (6 levels)                | Sparse     | Yes          | [Download](https://storage.googleapis.com/android_env-tasks/memory_game.tar.gz) |
| [Minesweeper](#minesweeper)                         | Tapping              | No             | Yes (3 levels)                | Sparse     | Yes          | [Download](https://storage.googleapis.com/android_env-tasks/minesweeper.tar.gz) |
| [Open Sudoku](#open-sudoku)                         | Tapping (buttons)    | No             | Yes (3 levels)                | Sparse     | Yes          | [Download](https://storage.googleapis.com/android_env-tasks/open_sudoku.tar.gz) |
| [Perfection](#perfection)                           | Drag & drop          | No             | Yes (3 game types)            | Dense      | Yes          | [Download](https://storage.googleapis.com/android_env-tasks/perfection.tar.gz)  |
| [Rocket Sleigh](#rocket-sleigh)                     | Tapping              | Yes            | No                            | Dense      | No           | [Download](https://storage.googleapis.com/android_env-tasks/rocket_sleigh.tar.gz) |
| [Pong](#pong)                                       | Drag                 | Yes            | Yes (3 levels)                | Sparse     | Yes          | [Download](https://storage.googleapis.com/android_env-tasks/pong.tar.gz)   |
| [SGT Puzzles - Blackbox](#sgt-puzzles-blackbox)     | Tapping              | No             | Yes (4 levels)                | Sparse     | Yes          | [Download](https://storage.googleapis.com/android_env-tasks/sgtpuzzles.tar.gz) |
| [SGT Puzzles - Bridge](#sgt-puzzles-bridge)         | Drag & drop          | No             | Yes (5 levels)                | Sparse     | Yes          | [Download](https://storage.googleapis.com/android_env-tasks/sgtpuzzles.tar.gz) |
| [SGT Puzzles - Cube](#sgt-puzzles-cube)             | Tapping              | No             | Yes (3 levels)                | Sparse     | Yes          | [Download](https://storage.googleapis.com/android_env-tasks/sgtpuzzles.tar.gz)  |
| [SGT Puzzles - Dominosa](#sgt-puzzles-dominosa)     | Tapping              | No             | Yes (5 levels)                | Sparse     | Yes          | [Download](https://storage.googleapis.com/android_env-tasks/sgtpuzzles.tar.gz)  |
| [SGT Puzzles - Fifteen](#sgt-puzzles-fifteen)       | Tapping              | No             | Yes (4 levels)                | Sparse     | Yes          | [Download](https://storage.googleapis.com/android_env-tasks/sgtpuzzles.tar.gz) |
| [SGT Puzzles - Flip](#sgt-puzzles-flip)             | Tapping              | No             | Yes (3 levels)                | Sparse     | Yes          | [Download](https://storage.googleapis.com/android_env-tasks/apple_flinger.tar.gz) |
| [SGT Puzzles - Flood](#sgt-puzzles-flood)           | Tapping              | No             | Yes (3 levels)                | Sparse     | Yes          | [Download](https://storage.googleapis.com/android_env-tasks/sgtpuzzles.tar.gz) |
| [SGT Puzzles - Galaxies](#sgt-puzzles-galaxies)     | Tapping              | No             | Yes (6 sizes)                 | Sparse     | Yes          | [Download](https://storage.googleapis.com/android_env-tasks/sgtpuzzles.tar.gz) |
| [SGT Puzzles - Guess](#sgt-puzzles-guess)           | Tapping              | No             | Yes (4 levels)                | Sparse     | Yes          | [Download](https://storage.googleapis.com/android_env-tasks/sgtpuzzles.tar.gz) |
| [SGT Puzzles - Inertia](#sgt-puzzles-inertia)       | Tapping              | No             | Yes (2 sizes)                 | Sparse     | Yes          | [Download](https://storage.googleapis.com/android_env-tasks/sgtpuzzles.tar.gz) |
| [SGT Puzzles - Light Up](#sgt-puzzles-light-up)     | Tapping              | No             | Yes (5 sizes)                 | Sparse     | Yes          | [Download](https://storage.googleapis.com/android_env-tasks/sgtpuzzles.tar.gz) |
| [SGT Puzzles - Loopy](#sgt-puzzles-loopy)           | Tapping              | No             | Yes (3 sizes)                 | Sparse     | Yes          | [Download](https://storage.googleapis.com/android_env-tasks/sgtpuzzles.tar.gz) |
| [SGT Puzzles - Net](#sgt-puzzles-net)               | Tapping              | No             | Yes (5 sizes)                 | Sparse     | Yes          | [Download](https://storage.googleapis.com/android_env-tasks/sgtpuzzles.tar.gz) |
| [Shattered Pixel Dungeon](#shattered-pixel-dungeon) | Tapping              | Yes            | Yes (4 variants)              | Sparse     | No           | [Download](https://storage.googleapis.com/android_env-tasks/shattered_pixel_dungeon.tar.gz) |
| [Simple Solitaire](#simple-solitaire)               | Drag & drop          | No             | Yes (19 tasks)                | Sparse     | Yes          | [Download](https://storage.googleapis.com/android_env-tasks/simple_solitaire.tar.gz) |
| [Snake](#snake)                                     | Tapping (buttons)    | Yes            | No                            | Sparse     | Yes          | [Download](https://storage.googleapis.com/android_env-tasks/aosp_samples_snake.tar.gz) |
| [Vector Pinball](#vector-pinball)                   | Tapping              | Yes            | Yes (5 variants)              | Sparse     | No           | [Download](https://storage.googleapis.com/android_env-tasks/vector_pinball.tar.gz) |

<!-- mdformat on -->

## Vokram

Vokram is our in-house implementation of an Android app that displays a
Markov-Decision-Process (MDP) graph as buttons on the screen which the agent
must use to select its actions. The observation is simply the color of the
background, and the actions are the buttons themselves which are presented in
different colors.

*   **mdp_0000**: This is a task that presents the agent with two colored, but
    unlabeled buttons on the screen. Pressing one of the buttons gives the agent
    a reward of `-1` and redraws the buttons on the screen. The other button
    gives a reward of `+1` and terminates the episode. The color of the buttons
    is the same throughout the episode. The sizes of the buttons are randomized
    at each screen draw. Pressing anywhere else on the screen gives a reward of
    zero. The task lasts up 60 seconds, at which point the episode is restarted.
    The underlying dynamics governing the buttons is a simple 2-state 2-action
    Markov Decision Process (MDP). The MDP is intentionally simple environment
    that can be used to debug agents.

*   **mdp_0001**: This is similar to `mdp_0000` but it's even simpler. It
    presents the agent with a single button which gives a reward of `+1` and
    terminates the episode when pressed. This task can be used for example to
    train agents to click buttons.

*   **mdp_0002**: In this task there are two buttons, pressing either of which
    will terminate the episode with a return of `+1`.

*   **mdp_0003**: An equivalent of `mdp_0000` with rewards reversed: the episode
    ends when the wrong button is clicked, and carries on with a new set of
    buttons when the correct one is clicked.

<details>
  <summary>Extras returned</summary>

*   `actions`:
    -   Set of all buttons present, e.g. `['A', 'B']`.
    -   Returned when any button is pressed.
    -   Has `shape=[2], dtype=STRING_U1`.
*   `clicks`:
    -   Character representing the button pressed.
    -   Returned when any button is pressed.
    -   Has `shape=[1], dtype=STRING_U1`.
*   `buttons`:
    -   Coordinates of the top left and bottom right corners of each button, e.g
        `[[x_a_0, y_a_0, x_a_1, y_a_1], [x_b_0, y_a_0, x_b_1, y_b_1]]`.
    -   Returned when any button is pressed.
    -   Has `shape=[2, 4], dtype=INT32`.

</details>

**mdp_0000**                                     | **mdp_0001**                                     | **mdp_0002**                                     | **mdp_0003**
------------------------------------------------ | ------------------------------------------------ | ------------------------------------------------ | ------------
![Screenshot of 'mdp_0000'](images/mdp_0000.gif) | ![Screenshot of 'mdp_0001'](images/mdp_0001.gif) | ![Screenshot of 'mdp_0002'](images/mdp_0002.gif) | ![Screenshot of 'mdp_0003'](images/mdp_0003.gif)

## Accessibility Forwarder

This application returns all
[accessibility service events](https://developer.android.com/reference/android/accessibilityservice/AccessibilityService)
in the form of task extras, and, if specified, rewards the agent for specific
events. Most often this means receiving the description of an accessibility
event in string format, such as `A button was pressed.` or `A menu was opened.`
etc. The two specific tasks provided here associate rewards with opening the
"History" menu in the Calculator app, and setting and resetting a timer in the
Clock app, but any goal can be provided in a similar manner if it can be
associated with an accessibility event.

<details>
  <summary>Extras returned</summary>

*   `clicks`:
    -   Content description of UI element (e.g. "Settings").
    -   Returned the given UI element is clicked.
    -   Has `shape=[1], dtype=STRING_U250`.
*   `event`:
    -   Text of the accessibility event (e.g. button text "Accept & Continue").
    -   Returned upon accessibility event.
    -   Has `shape=[1], dtype=STRING_U250`.

</details>

**calculator_history**                                               | **clock_set_timer**
-------------------------------------------------------------------- | -------------------
![Screenshot of 'calculator_history'](images/calculator_history.gif) | ![Screenshot of 'clock_set_timer'](images/clock.gif)

## Apple Flinger

A clone of Angry Birds. Even though the game offers many levels, we currently
expose six levels. See the original github repo for more info:
https://gitlab.com/ar-/apple-flinger.

<details>
  <summary>Extras returned</summary>
  Returns no extras.
</details>

**apple_flinger_M_1_1**                                                | **apple_flinger_M_1_2**                                                | **apple_flinger_M_1_18**
---------------------------------------------------------------------- | ---------------------------------------------------------------------- | ------------------------
![Screenshot of 'apple_flinger_M_1_1'](images/apple_flinger_M_1_1.gif) | ![Screenshot of 'apple_flinger_M_1_2'](images/apple_flinger_M_1_2.gif) | ![Screenshot of 'apple_flinger_M_1_18'](images/apple_flinger_M_1_18.gif)

**apple_flinger_M_2_1**                                                 | **apple_flinger_M_2_2**                                                | **apple_flinger_M_2_18**
----------------------------------------------------------------------- | ---------------------------------------------------------------------- | ------------------------
![Screenshot of 'apple_flinger_M_2_1'](images/apple_flinger_M_2_1.gif)) | ![Screenshot of 'apple_flinger_M_2_2'](images/apple_flinger_M_2_2.gif) | ![Screenshot of 'apple_flinger_M_2_18'](images/apple_flinger_M_2_18.gif)

## Blockinger

This is a Tetris clone implemented with on-screen controls. See the original
github repo for more info: https://github.com/tasioleiva/Blockinger.git.

<details>
  <summary>Extras returned</summary>

*   `down_pressed`, `left_pressed`, `right_pressed`, `rotate_right_pressed`,
    `drop_pressed`:
    -   Indicates that said button has been pressed.
    -   Returned when said button has been pressed.
    -   Has `shape=[1], dtype=INT32`.
*   `current_board`:
    -   One-hot encoded state of the board.
    -   Has `shape=[18, 10], dtype=INT32`.
*   `current_line`, `cleared_lines`:
    -   Index of the relevant line.
    -   Has `shape=[1], dtype=INT32`.
*   `current_piece`, `next_piece`:
    -   Index representing the type of piece.
    -   Has `shape=[1], dtype=INT32`.

</details>

![Screenshot of 'blockinger'](images/blockinger.gif)

## Catch

Classic Catch game.

<details>
  <summary>Extras returned</summary>

*   `ball`:
    -   `x, y` coordinates of the ball.
    -   Returned every timestep.
    -   Has `shape=[2], dtype=INT32`.
*   `paddle`:
    -   `x, y` coordinates of the paddle.
    -   Returned every timestep.
    -   Has `shape=[2], dtype=INT32`.
*   `paddle_width`:
    -   Width of the paddle.
    -   Returned every timestep.
    -   Has `shape=[1], dtype=INT32`.
*   `lives`:
    -   Number of lives left.
    -   Returned every timestep.
    -   Has `shape=[1], dtype=INT32`.

</details>

![Screenshot of 'catch_the_ball_default'](images/catch_the_ball_default.gif)

## Classic 2048

This is an Android implementation of a popular game in the 2010s. See the
original github repo for more info: https://github.com/tpcstld/2048.git.

<details>
  <summary>Extras returned</summary>

*   `grid`:
    -   State of the board.
    -   Returned when the board changes.
    -   Has `shape=[4, 4], dtype=INT32`.
*   `direction`:
    -   Index representing the direction of the last swipe (between 0-3).
    -   Returned when the swipe prompted a board change.
    -   Has `shape=[1], dtype=INT32`.

</details>

![Screenshot of 'classic_2048'](images/classic_2048.gif)

## Dodge

Guide the ball from the red line to the green line without getting hit by the
floating dots.

<details>
  <summary>Extras returned</summary>

*   `lives`:
    -   Number of lives left.
    -   Returned when its value changes.
    -   Has `shape=[1], dtype=INT32`.
*   `level`:
    -   Current level.
    -   Returned when its value changes.
    -   Has `shape=[1], dtype=INT32`.

</details>

![Screenshot of 'dodge_default'](images/dodge_default.gif)

## DroidFish

Standard chess game. You can choose whether to play as a specific player
(black/white), or have the player colour randomly assigned at the beginning of
each episode. The numbers 1, 10 and 100 indicate the level of difficulty. Take a
look at a few sample moves below to get an idea of roughly how well the bot
plays for each level of difficulty. You can see that the 1% and 10% bots often
make very obvious mistakes. See the original github repo for more info:
https://github.com/peterosterlund2/droidfish.

<details>
  <summary>Extras returned</summary>

*   `board`:
    -   State of the board, representing pieces by indices. No piece - 0
    -   White pieces - 1: king, 2: queen, 3: rook, 4: bishop, 5: knight, 6: pawn
    -   Black pieces - 7: king, 8: queen, 9: rook, 10: bishop, 11: knight, 12:
        pawn
    -   Returned when the board changes.
    -   Has `shape=[8, 8], dtype=INT32`.
*   `selection`:
    -   Coordinate of selected piece (between 0-64, -1 if selection is removed)
    -   Returned when a piece is selected (or unselected).
    -   Has `shape=[1], dtype=INT32`.
*   `moved`:
    -   Coordinates "from" and "to" cells when a piece is moved (between 0-64)
    -   Returned when a piece is moved.
    -   Has `shape=[2], dtype=INT32`.
*   `invalid`:
    -   Coordinates "from" and "to" cells of an invalid move attempt (between
        0-64)
    -   Returned upon invalid move request
    -   Has `shape=[2], dtype=INT32`.

</details>

**droidfish_black_1**                                              | **droidfish_black_10**                                               | **droidfish_black_100**
------------------------------------------------------------------ | -------------------------------------------------------------------- | -----------------------
![Screenshot of 'droidfish_black_1'](images/droidfish_black_1.gif) | ![Screenshot of 'droidfish_black_10'](images/droidfish_black_10.gif) | ![Screenshot of 'droidfish_black_100'](images/droidfish_black_100.gif)

**droidfish_white_1**                                              | **droidfish_white_10**                                               | **droidfish_white_100**
------------------------------------------------------------------ | -------------------------------------------------------------------- | -----------------------
![Screenshot of 'droidfish_white_1'](images/droidfish_white_1.gif) | ![Screenshot of 'droidfish_white_10'](images/droidfish_white_10.gif) | ![Screenshot of 'droidfish_white_100'](images/droidfish_white_100.gif)

**droidfish_random_1**                                               | **droidfish_random_10**                                                | **droidfish_random_100**
-------------------------------------------------------------------- | ---------------------------------------------------------------------- | ------------------------
![Screenshot of 'droidfish_random_1'](images/droidfish_random_1.gif) | ![Screenshot of 'droidfish_random_10'](images/droidfish_random_10.gif) | ![Screenshot of 'droidfish_random_100'](images/droidfish_random_100.gif)

## FlappyDroid

A clone of the well-known game Flappy Birds.

<details>
  <summary>Extras returned</summary>
  Returns no extras.
</details>

**systemui_egg_land_default**                                                      | **systemui_egg_land_half_speed**
---------------------------------------------------------------------------------- | --------------------------------
![Screenshot of 'systemui_egg_land_default'](images/systemui_egg_land_default.gif) | ![Screenshot of 'systemui_egg_land_half_speed'](images/systemui_egg_land_half_speed.gif)

## Frozen Bubble

Shoot the coloured bubbles in a direction of your choice. Groups of bubbles with
the same colour will drop. Remove all bubbles from the board before the time
runs out. See the original github repo for more info:
https://github.com/robinst/frozen-bubble-android.git.

<details>
  <summary>Extras returned</summary>
  Returns no extras.
</details>

![Screenshot of 'frozen_bubble'](images/frozen_bubble.gif)

## Memory Game

Classic memory game. Find the pairs of images. See the original github repo for
more info: https://github.com/sromku/memory-game/.

<details>
  <summary>Extras returned</summary>

*   `flip`:
    -   Index of the card flipped
    -   Returned when a card is clicked.
    -   Has `shape=[1], dtype=INT32`.
*   `cards`:
    -   Number of cards still on the board.
    -   Returned upon finding a pair.
    -   Has `shape=[1], dtype=INT32`.
*   `remained`:
    -   Number of cards remaining at the end of the episode.
    -   Returned when an episode is over.
    -   Has `shape=[1], dtype=INT32`.
*   `stars`:
    -   Number of stars achieved at the end of the game.
    -   Returned upon finishing the game.
    -   Has `shape=[1], dtype=INT32`.
*   `achieved`:
    -   Score obtained by the end of the episode.
    -   Returned when an episode is over.
    -   Has `shape=[1], dtype=INT32`.

</details>

**memory_game_animals_beginner**                                                         | **memory_game_animals_easy**                                                     | **memory_game_monsters_medium**
---------------------------------------------------------------------------------------- | -------------------------------------------------------------------------------- | -------------------------------
![Screenshot of 'memory_game_animals_beginner'](images/memory_game_animals_beginner.gif) | ![Screenshot of 'memory_game_animals_easy'](images/memory_game_animals_easy.gif) | ![Screenshot of 'memory_game_monsters_medium'](images/memory_game_monsters_medium.gif)

**memory_game_monsters_hard**                                                      | **memory_game_emojis_hardest**                                                       | **memory_game_emojis_master**
---------------------------------------------------------------------------------- | ------------------------------------------------------------------------------------ | -----------------------------
![Screenshot of 'memory_game_monsters_hard'](images/memory_game_monsters_hard.gif) | ![Screenshot of 'memory_game_emojis_hardest'](images/memory_game_emojis_hardest.gif) | ![Screenshot of 'memory_game_emojis_master'](images/memory_game_emojis_master.gif)

## Minesweeper

This is an Android implementation of a popular game on Desktop in the 1990s. See
the original github repo for more info: https://gitlab.com/ar-/apple-flinger.

<details>
  <summary>Extras returned</summary>

*   `hidden`:
    -   Number of hidden cells.
    -   Returned whenever the board changes.
    -   Has `shape=[1], dtype=INT32`.
*   `revealed`:
    -   Number of revealed cells.
    -   Returned whenever the board changes.
    -   Has `shape=[1], dtype=INT32`.
*   `bombs`:
    -   Number of bombs in the game.
    -   Returned whenever the board changes.
    -   Has `shape=[1], dtype=INT32`.
*   `click`:
    -   Coordinates of the cell clicked (row, column).
    -   Returned whenever the board changes.
    -   Has `shape=[2], dtype=INT32`.
*   `grid`:
    -   State of the board.
    -   -1 = hidden, -2 = marked, 9 = bomb, 0-8 = number of nearby bombs
    -   Returned whenever the board changes.
    -   Has `shape=[grid_height, grid_width], dtype=INT32`.

</details>

**minesweeper_easy**                                             | **minesweeper_medium**                                               | **minesweeper_hard**
---------------------------------------------------------------- | -------------------------------------------------------------------- | --------------------
![Screenshot of 'minesweeper_easy'](images/minesweeper_easy.gif) | ![Screenshot of 'minesweeper_medium'](images/minesweeper_medium.gif) | ![Screenshot of 'minesweeper_hard'](images/minesweeper_hard.gif)

## Open Sudoku

Classic Sudoku game with different levels of difficulty. The board is randomised
over a set of 30 boards for each level. See the original github repo for more
info: https://github.com/ogarcia/opensudoku.

<details>
  <summary>Extras returned</summary>

*   `value`:
    -   Number pressed (between 1-9, 0 if the "delete" button is pressed).
    -   Returned upon clicking said button.
    -   Has `shape=[1], dtype=INT32`.

</details>

**open_sudoku_easy**                                             | **open_sudoku_medium**                                               | **open_sudoku_hard**
---------------------------------------------------------------- | -------------------------------------------------------------------- | --------------------
![Screenshot of 'open_sudoku_easy'](images/open_sudoku_easy.gif) | ![Screenshot of 'open_sudoku_medium'](images/open_sudoku_medium.gif) | ![Screenshot of 'open_sudoku_hard'](images/open_sudoku_hard.gif)

## Perfection

Drag the items corresponding to the targets with the same shape.

<details>
  <summary>Extras returned</summary>

*   `moving`:
    -   The ID of the piece being dragged on the screen or 0.
    -   Returned when its value changes.
    -   Has `shape=[1], dtype=INT32`.
*   `todo`:
    -   Number of pieces yet to be moved to a hole.
    -   Returned when its value changes.
    -   Has `shape=[1], dtype=INT32`.
*   `done`:
    -   Number of pieces correctly moved to a hole.
    -   Returned when its value changes.
    -   Has `shape=[1], dtype=INT32`.

</details>

**perfection_1_circle_static**                                                       | **perfection_1_cube_static**                                                         | **perfection_1_plus_static**                                                     | **perfection_1_triangle_static**
------------------------------------------------------------------------------------ | ------------------------------------------------------------------------------------ | -------------------------------------------------------------------------------- | --------------------------------
![Screenshot of 'perfection_1_circle_static'](images/perfection_1_circle_static.gif) | ![Screenshot of 'perfection_1_square_static'](images/perfection_1_square_static.gif) | ![Screenshot of 'perfection_1_plus_static'](images/perfection_1_plus_static.gif) | ![Screenshot of 'perfection_1_triangle_static'](images/perfection_1_triangle_static.gif)

**perfection_default**                                               | **perfection_4_colors_square_static**                                                              | **perfection_4_pieces_static**
-------------------------------------------------------------------- | -------------------------------------------------------------------------------------------------- | ------------------------------
![Screenshot of 'perfection_default'](images/perfection_default.gif) | ![Screenshot of 'perfection_4_colors_square_static'](images/perfection_4_colors_square_static.gif) | ![Screenshot of 'perfection_4_pieces_static'](images/perfection_4_pieces_static.gif)

## Rocket Sleigh

A Flappy Bird-like game where you have to collect christmas presents while
avoiding trees. The sleigh is powered by a rocket that needs to recharge over
time after you use up its fuel.

<details>
  <summary>Extras returned</summary>
  Returns no extras.
</details>

![Screenshot of 'rocket_sleigh_default'](images/rocket_sleigh.gif)

## Pong

Classic Pong game.

<details>
  <summary>Extras returned</summary>

*   `ball`:
    -   The ball coordinates: [left, top, right, bottom].
    -   Returned when its value changes.
    -   Has `shape=[4], dtype=INT32`.
*   `computer`:
    -   The computer paddle coordinates: [left, top, right, bottom].
    -   Returned when its value changes.
    -   Has `shape=[4], dtype=INT32`.
*   `human`:
    -   The human paddle coordinates: [left, top, right, bottom].
    -   Returned when its value changes.
    -   Has `shape=[4], dtype=INT32`.
*   `collision`:
    -   Indicates collision of paddle and ball: (0=no collision, 1=collision).
    -   Returned when its value changes.
    -   Has `shape=[1], dtype=INT32`.
*   `state`:
    -   The current state of the game: (0=pause, 1=ready, 2=running, 3=lose,
        4=win).
    -   Returned when its value changes.
    -   Has `shape=[1], dtype=INT32`.

</details>

**pong_easy**                                      | **pong_default**                                         | **pong_hard**
-------------------------------------------------- | -------------------------------------------------------- | -------------
![Screenshot of 'pong_easy'](images/pong_easy.gif) | ![Screenshot of 'pong_default'](images/pong_default.gif) | ![Screenshot of 'pong_hard'](images/pong_hard.gif)

## SGT Puzzles - Blackbox

There's an invisible laser beam originating from each of the cells at the edge
of the grid. There are also a given number of balls inside the grid, hidden from
the player whose aim is to guess where those balls are. The player can figure
out where those balls might be by looking at how they *deflect* the laser beams.
Clicking on an edge cell the player can reveal information about how that
particular laser beam travels. Click on a cell; if the cell reveals an `H`, it
means the straight laser beam leaving this cell hits a ball frontally. If the
cell reveals a *number* along with another cell with the same number, that means
the laser beam originating in the first cell ends up getting absorbed in the
corresponding pair cell. If the cell reveals an `R`, that means the laser beam
was *reflected*: either its origin and the cell it gets absorbed in is the
*same*, or the beam gets bent before entering the grid. See the description
below.

The balls affect the travel of the laser beam in the following way:

*   If a laser beam hits it straight, it gets absorbed. This is denoted by the
    letter `H`.
*   If a laser beam hits *its corner*, the beam gets deflected by 90 degrees.
*   If a laser beam hits a balls corner right at the edge of the grid, i.e.
    before it enters the grid, it is considered *reflected*.
*   If a laser beam enters the same cell that it originally left, it is
    considered *reflected* too.

Once the player has placed the given number of balls on the screen, a green dot
appears that allows the player to check if their solution was correct. See the
original github repo for more info: https://github.com/chrisboyle/sgtpuzzles.

<details>
  <summary>Extras returned</summary>

*   `balls`:
    -   The number of balls in the game arena.
    -   Returned whenever the grid changes.
    -   Has `shape=[1], dtype=INT32`.
*   `guesses`:
    -   The number of guessed balls made by the agent.
    -   Returned whenever the grid changes.
    -   Has `shape=[1], dtype=INT32`.
*   `wrong`:
    -   1 if the guesses are wrong, 0 otherwise.
    -   Returned whenever the grid changes.
    -   Has `shape=[1], dtype=INT32`.
*   `lasers`:
    -   The number of lasers in the grid
    -   Returned whenever the grid changes.
    -   Has `shape=[1], dtype=INT32`.
*   `grid`:
    -   Representation of the grid cells:
    -   In the arena: `G`=guessed ball `' '`=empty
    -   In the range: `[0-9]`=the number of lasers, `H`=beam hit, `R`=beam
        reflected,
    -   `?` unknown, `' '` for the corners
    -   Returned whenever the grid changes.
    -   Has `shape=[grid_size, grid_size], dtype=STRING_U1`.

</details>

**blackbox_3x3_1_ball**                                                           | **blackbox_5x5_3_balls**                                                            | **blackbox_8x8_5_balls**                                                            | **blackbox_10x10_5_balls**
--------------------------------------------------------------------------------- | ----------------------------------------------------------------------------------- | ----------------------------------------------------------------------------------- | --------------------------
![Screenshot of 'blackbox_3x3_1_ball'](images/sgtpuzzles_blackbox_3x3_1_ball.gif) | ![Screenshot of 'blackbox_5x5_3_balls'](images/sgtpuzzles_blackbox_5x5_3_balls.gif) | ![Screenshot of 'blackbox_8x8_5_balls'](images/sgtpuzzles_blackbox_8x8_5_balls.gif) | ![Screenshot of 'blackbox_10x10_5_balls'](images/sgtpuzzles_blackbox_10x10_5_balls.gif)

## SGT Puzzles - Bridge

Connect nodes on the board so that each number denotes the degree of the given
vertex. Edges are not allowed to cross each other and the graph has to be
connected. Edges have to be horizontal or vertical, and there can be at most two
parallel bridges between any pair of nodes. See the original github repo for
more info: https://github.com/chrisboyle/sgtpuzzles.

<details>
  <summary>Extras returned</summary>

*   `islands`:
    -   Number of nodes on the board.
    -   Returned whenever the grid changes.
    -   Has `shape=[1], dtype=INT32`.
*   `grid`:
    -   Representation of the current state of the board.
    -   `[0-9]=island, ' '=empty`
    -   `'|'=vertical line, '"'=double vertical line, '!'=wrong vertical line`
    -   `'-'=horizontal line, '='=double horizontal line, '~'=wrong horizontal
        line`
    -   Returned whenever the grid changes.
    -   Has `shape=[grid_size, grid_size], dtype=STRING_U1`.

</details>

**bridge_7x7_easy**                                                                  | **bridge_7x7_medium**                                                                    | **bridge_7x7_hard**                                                                  | **bridge_10x10_medium**                                                                      | **bridge_15x15_medium**
------------------------------------------------------------------------------------ | ---------------------------------------------------------------------------------------- | ------------------------------------------------------------------------------------ | -------------------------------------------------------------------------------------------- | -----------------------
![Screenshot of 'sgtpuzzles_bridge_7x7_easy'](images/sgtpuzzles_bridge_7x7_easy.gif) | ![Screenshot of 'sgtpuzzles_bridge_7x7_medium'](images/sgtpuzzles_bridge_7x7_medium.gif) | ![Screenshot of 'sgtpuzzles_bridge_7x7_hard'](images/sgtpuzzles_bridge_7x7_hard.gif) | ![Screenshot of 'sgtpuzzles_bridge_10x10_medium'](images/sgtpuzzles_bridge_10x10_medium.gif) | ![Screenshot of 'sgtpuzzles_bridge_15x15_medium'](images/sgtpuzzles_bridge_15x15_medium.gif)

## SGT Puzzles - Cube

There are six coloured squares that you have to collect with a moving cube. If
the cube rolls on top of a coloured cell, the colour gets attached to that side
of the cube; and if a coloured side of the cube rolls on an empty cell, the
colour is removed from the cube. The goal is to have all six sides of the cube
coloured. See the original github repo for more info:
https://github.com/chrisboyle/sgtpuzzles.

<details>
  <summary>Extras returned</summary>

*   `current`:
    -   Index of the current grid cell the cube is on.
    -   Returned whenever the cube moves.
    -   Has `shape=[1], dtype=INT32`.
*   `previous`:
    -   Index of the previous grid cell the cube was on.
    -   Returned whenever the cube moves.
    -   Has `shape=[1], dtype=INT32`.
*   `grid`:
    -   The grid state (0 = dark, 1 = blue)
    -   Returned whenever the cube moves.
    -   Has `shape=[grid_size, grid_size], dtype=INT32`.
*   `face_count`:
    -   The number of dark faces on the cube.
    -   Returned whenever the cube moves.
    -   Has `shape=[1], dtype=INT32`.
*   `face_colour_count`:
    -   The number of blue faces on the cube.
    -   Returned whenever the cube moves.
    -   Has `shape=[1], dtype=INT32`.
*   `faces`:
    -   The cube faces (0 = dark, 1 = blue)
    -   Returned whenever the cube moves.
    -   Has `shape=[6], dtype=INT32`.

</details>

**cube_c3x3**                                                            | **cube_c4x4**                                                            | **cube_c8x8**
------------------------------------------------------------------------ | ------------------------------------------------------------------------ | -------------
![Screenshot of 'sgtpuzzles_cube_c3x3'](images/sgtpuzzles_cube_c3x3.gif) | ![Screenshot of 'sgtpuzzles_cube_c4x4'](images/sgtpuzzles_cube_c4x4.gif) | ![Screenshot of 'sgtpuzzles_cube_c8x8'](images/sgtpuzzles_cube_c8x8.gif)

## SGT Puzzles - Dominosa

Place 2x1 size dominoes on the board such that the full board is covered, making
sure that no two dominoes have the same pair of numbers on them. There needs to
be exactly one of (0, 0), (0, 1) (0, 2), ... (1, 1), (1, 2) etc. See the
original github repo for more info: https://github.com/chrisboyle/sgtpuzzles.

<details>
  <summary>Extras returned</summary>

*   `numbers`:
    -   Numbers as they appear in the grid.
    -   Returned whenever the grid changes.
    -   Has `shape=[height, width], dtype=INT32`.
*   `grid`:
    -   Letters representing the dominoes currently placed on the board.
    -   'R=right, L=left, T=top, B=bottom'
    -   Returned whenever the grid changes.
    -   Has `shape=[height, with], dtype=INT32`.
*   `clash`:
    -   Represents clashes on the board (i.e. if two dominoes have the same
        pair)
    -   '1=clash, 0=no clash'
    -   Returned whenever the grid changes.
    -   Has `shape=[height, width], dtype=INT32`.

</details>

**dominosa_1**                                                             | **dominosa_3**                                                             | **dominosa_3a**                                                              | **dominosa_6**                                                             | **dominosa_9**
-------------------------------------------------------------------------- | -------------------------------------------------------------------------- | ---------------------------------------------------------------------------- | -------------------------------------------------------------------------- | --------------
![Screenshot of 'sgtpuzzles_dominosa_1'](images/sgtpuzzles_dominosa_1.gif) | ![Screenshot of 'sgtpuzzles_dominosa_3'](images/sgtpuzzles_dominosa_3.gif) | ![Screenshot of 'sgtpuzzles_dominosa_3a'](images/sgtpuzzles_dominosa_3a.gif) | ![Screenshot of 'sgtpuzzles_dominosa_6'](images/sgtpuzzles_dominosa_6.gif) | ![Screenshot of 'sgtpuzzles_dominosa_9'](images/sgtpuzzles_dominosa_9.gif)

## SGT Puzzles - Fifteen

Order the tiles in increasing order, starting from the top left corner. See the
original github repo for more info: https://github.com/chrisboyle/sgtpuzzles.

<details>
  <summary>Extras returned</summary>

*   `grid`:
    -   Current state of the grid.
    -   Returned whenever the grid changes.
    -   Has `shape=[grid_size, grid_size], dtype=INT32`.
*   `empty`:
    -   Index of the single empty cell in the grid.
    -   Returned whenever the grid changes.
    -   Has `shape=[1], dtype=INT32`.
*   `movecount`:
    -   Number of moves made so far.
    -   Returned whenever the grid changes.
    -   Has `shape=[1], dtype=INT32`.

</details>

**fifteen_2x2**                                                              | **fifteen_3x3**                                                              | **fifteen_4x4**                                                              | **fifteen_6x6**
---------------------------------------------------------------------------- | ---------------------------------------------------------------------------- | ---------------------------------------------------------------------------- | ---------------
![Screenshot of 'sgtpuzzles_fifteen_2x2'](images/sgtpuzzles_fifteen_2x2.gif) | ![Screenshot of 'sgtpuzzles_fifteen_3x3'](images/sgtpuzzles_fifteen_3x3.gif) | ![Screenshot of 'sgtpuzzles_fifteen_4x4'](images/sgtpuzzles_fifteen_4x4.gif) | ![Screenshot of 'sgtpuzzles_fifteen_6x6'](images/sgtpuzzles_fifteen_6x6.gif)

## SGT Puzzles - Flip

Clicking on a cell will flip the colour of some of its neighbours, which are
determined by the symbol in the cell. The goal is to make all the cells have the
same colour. See the original github repo for more info:
https://github.com/chrisboyle/sgtpuzzles.

<details>
  <summary>Extras returned</summary>

*   `light`:
    -   The number of light cells.
    -   Returned whenever the grid changes.
    -   Has `shape=[1], dtype=INT32`.
*   `dark`:
    -   The number of dark cells
    -   Returned whenever the grid changes.
    -   Has `shape=[1], dtype=INT32`.
*   `moves`:
    -   The number of moves made by the player.
    -   Returned whenever the grid changes.
    -   Has `shape=[1], dtype=INT32`.
*   `grid`:
    -   State of the board (0 = dark, 1 = light).
    -   Returned whenever the grid changes.
    -   Has `shape=[grid_size, grid_size], dtype=INT32`.
*   `gridMatrix`:
    -   The grid matrix of square neighbours (-1 = outside, 1 = neighbour, 0 =
        not neighbour)
    -   Returned whenever the grid changes.
    -   Has `shape=[grid_size, grid_size, 3, 3s], dtype=INT32`.

</details>

**flip_3x3c**                                                            | **flip_4x4c**                                                            | **flip_5x5r**
------------------------------------------------------------------------ | ------------------------------------------------------------------------ | -------------
![Screenshot of 'sgtpuzzles_flip_3x3c'](images/sgtpuzzles_flip_3x3c.gif) | ![Screenshot of 'sgtpuzzles_flip_4x4c'](images/sgtpuzzles_flip_4x4c.gif) | ![Screenshot of 'sgtpuzzles_flip_5x5r'](images/sgtpuzzles_flip_5x5r.gif)

## SGT Puzzles - Flood

FloodIt is a game where the player needs to fill the board with a single color.
The dynamics of the game are driven by colored areas of the board, which when
pressed cause the currently active region to change its color to the color of
the pressed button. When this active region changes color it absorbs neighboring
squares that have the same color, thus expanding the active region. The active
region starts as a single square at the top-left corner of the board. The game
gives a single reward at the end of the game if the player manages to fill the
entire board with the same color within the maximum number steps, otherwise the
reward is just zero. See the original github repo for more info:
https://github.com/chrisboyle/sgtpuzzles.

<details>
  <summary>Extras returned</summary>

*   `board`:
    -   State of the board, representing colours by their indices.
    -   0: red, 1: yellow, 2: green, 3: blue, 4: orange, 5: purple,
    -   6: brown, 7: light blue, 8: light green, 9: pink
    -   Returned whenever the grid changes.
    -   Has `shape=[grid_size, grid_size], dtype=INT32`.

</details>

**sgtpuzzles_flood_3x3_easy**                                                      | **sgtpuzzles_flood_12x12_medium**                                                          | **sgtpuzzles_flood_16x16_hard**
---------------------------------------------------------------------------------- | ------------------------------------------------------------------------------------------ | -------------------------------
![Screenshot of 'sgtpuzzles_flood_3x3_easy'](images/sgtpuzzles_flood_3x3_easy.gif) | ![Screenshot of 'sgtpuzzles_flood_12x12_medium'](images/sgtpuzzles_flood_12x12_medium.gif) | ![Screenshot of 'sgtpuzzles_flood_16x16_hard'](images/sgtpuzzles_flood_16x16_hard.gif)

## SGT Puzzles - Galaxies

Split the grid up into centrally symmetric areas. The centre of symmetry for
each area is denoted by a dot on the grid. See the original github repo for more
info: https://github.com/chrisboyle/sgtpuzzles.

<details>
  <summary>Extras returned</summary>

*   `dot`:
    -   Number of dots on the board.
    -   Returned whenever the grid changes.
    -   Has `shape=[1], dtype=INT32`.
*   `grid`:
    -   String representation of the board:
    -   `o`=dot, `' '`=empty, `+`, `-`, `|` = cell corners.
    -   Returned whenever the grid changes.
    -   Has `shape=[grid_size, grid_size], dtype=STRING_U1`.

</details>

**galaxies_3x3_normal**                                                           | **galaxies_5x5_normal**                                                           | **galaxies_7x7_normal**
--------------------------------------------------------------------------------- | --------------------------------------------------------------------------------- | -----------------------
![Screenshot of 'galaxies_3x3_normal'](images/sgtpuzzles_galaxies_3x3_normal.gif) | ![Screenshot of 'galaxies_5x5_normal'](images/sgtpuzzles_galaxies_5x5_normal.gif) | ![Screenshot of 'galaxies_7x7_normal'](images/sgtpuzzles_galaxies_7x7_normal.gif)

**galaxies_7x7_unreasonable**                                                     | **galaxies_10x10_normal**                                                             | **galaxies_15x15_normal**
--------------------------------------------------------------------------------- | ------------------------------------------------------------------------------------- | -------------------------
![Screenshot of 'galaxies_7x7_normal'](images/sgtpuzzles_galaxies_7x7_normal.gif) | ![Screenshot of 'galaxies_10x10_normal'](images/sgtpuzzles_galaxies_10x10_normal.gif) | ![Screenshot of 'galaxies_15x15_normal'](images/sgtpuzzles_galaxies_15x15_normal.gif)

## SGT Puzzles - Guess

The computer has thought of a sequence of colours that you have to guess. Fill
the top row with colours of your choice, and wait for the computer to give you
feedback about your sequence. It will show a black dot for each colour that is
placed in the correct position, and a white dot for each that is present in the
hidden sequence, but not at the position your guess. Try to figure out the
hidden sequence before you run out of guesses! See the original github repo for
more info: https://github.com/chrisboyle/sgtpuzzles.

<details>
  <summary>Extras returned</summary>

*   `peg`:
    -   Indices representing the colours selected in the latest row.
    -   Returned after the row is completed and evaluated.
    -   Has `shape=[row_length], dtype=INT32`.
*   `feedback`:
    -   Evaluation of the latest guess (0: incorrect, 1: correct place, 2:
        correct colour)
    -   Returned after the row is completed and evaluated.
    -   Has `shape=[row_length], dtype=INT32`.

</details>

**guess_basic**                                                              | **guess_quick**                                                   | **guess_standard**                                                      | **guess_super**
---------------------------------------------------------------------------- | ----------------------------------------------------------------- | ----------------------------------------------------------------------- | ---------------
![Screenshot of 'sgtpuzzles_guess_basic'](images/sgtpuzzles_guess_basic.gif) | ![Screenshot of 'guess_quick'](images/sgtpuzzles_guess_quick.gif) | ![Screenshot of 'guess_standard'](images/sgtpuzzles_guess_standard.gif) | ![Screenshot of 'guess_super'](images/sgtpuzzles_guess_super.gif)

## SGT Puzzles - Inertia

Collect all the blue diamonds on the board without colliding into a bomb. You
can move the ball in the 8 main directions (including the diagonals). The ball
will keep on moving in that direction until it hits a wall, a bomb, a diamond or
a circle. Circles and diamonds have grip, i.e. it will stop the ball from
continuing to move in the direction it was going towards. See the original
github repo for more info: https://github.com/chrisboyle/sgtpuzzles.

<details>
  <summary>Extras returned</summary>

*   `gems`:
    -   Current number of gems still on the board.
    -   Returned whenever the grid changes.
    -   Has `shape=[1], dtype=INT32`.
*   `distancemoved`:
    -   Number of cells just moved.
    -   Returned whenever the grid changes.
    -   Has `shape=[1], dtype=INT32`.
*   `grid`:
    -   Symbols of grid cells (b=blank, g=gem, m=mine, s=stop, w=wall)
    -   Returned whenever the grid changes.
    -   Has `shape=[grid_size, grid_size], dtype=INT32`.

</details>

**inertia_5x5**                                                   | **inertia_10x10**
----------------------------------------------------------------- | -----------------
![Screenshot of 'inertia_5x5'](images/sgtpuzzles_inertia_5x5.gif) | ![Screenshot of 'inertia_10x10'](images/sgtpuzzles_inertia_10x10.gif)

## SGT Puzzles - Light Up

You have a grid of squares. Some are empty (black) and some are *walls* (grey);
some of the walls are numbered. Your aim is to *light up* all the empty squares
by placing light bulbs in some of them. The numbers denote how many bulbs' light
hits these directly in a straight sight. Meanwhile, no two bulbs should light up
each other (i.e. only one bulb allowed in straight sight). See the original
github repo for more info: https://github.com/chrisboyle/sgtpuzzles.

<details>
  <summary>Extras returned</summary>

*   `grid`:
    -   String representation of the board:
    -   '#' = blocked, ' ' = empty/black, 'L' = light, 'l' = illuminated,
    -   'X' = impossible, 'number' = number of bulbs hitting this cell.
    -   Returned whenever the grid changes.
    -   Has `shape=[grid_size, grid_size], dtype=STRING_U1`.

</details>

**light_up_3x3_easy**                                                                    | **light_up_5x5_easy**                                                                    | **light_up_7x7_easy**                                                                    | **light_up_10x10_tricky**                                                             | **light_up_14x14_easy**
---------------------------------------------------------------------------------------- | ---------------------------------------------------------------------------------------- | ---------------------------------------------------------------------------------------- | ------------------------------------------------------------------------------------- | -----------------------
![Screenshot of 'sgtpuzzles_light_up_3x3_easy'](images/sgtpuzzles_light_up_3x3_easy.gif) | ![Screenshot of 'sgtpuzzles_light_up_5x5_easy'](images/sgtpuzzles_light_up_5x5_easy.gif) | ![Screenshot of 'sgtpuzzles_light_up_7x7_easy'](images/sgtpuzzles_light_up_7x7_easy.gif) | ![Screenshot of 'light_up_10x10_tricky'](images/sgtpuzzles_light_up_10x10_tricky.gif) | ![Screenshot of 'sgtpuzzles_light_up_14x14_easy'](images/sgtpuzzles_light_up_14x14_easy.gif)

## SGT Puzzles - Loopy

Draw a closed loop along the edges of the grid. A number in a cell denotes the
number of edges adjacent to that cell. The loop cannot intersect itself. See the
original github repo for more info: https://github.com/chrisboyle/sgtpuzzles.

<details>
  <summary>Extras returned</summary>

*   `grid`:
    -   String representation of the board:
    -   The grid lines and cells:
    -   `.` = dots (cell corners)
    -   `0-9` = number on cell face or ` ` for empty face
    -   `?` = unknown (default),`x` = no line,`-` = `|` = line,`~` = `/` = error
    -   Returned whenever the grid changes.
    -   Has `shape=[grid_size, grid_size], dtype=STRING_U1`.

</details>

**loopy_3x3_easy**                                                                 | **loopy_5x5_easy**                                                                 | **loopy_7x7_easy**                                                                 | **loopy_7x7_normal**                                                                   | **loopy_7x7_hard**
---------------------------------------------------------------------------------- | ---------------------------------------------------------------------------------- | ---------------------------------------------------------------------------------- | -------------------------------------------------------------------------------------- | ------------------
![Screenshot of 'sgtpuzzles_loopy_3x3_easy'](images/sgtpuzzles_loopy_3x3_easy.gif) | ![Screenshot of 'sgtpuzzles_loopy_5x5_easy'](images/sgtpuzzles_loopy_5x5_easy.gif) | ![Screenshot of 'sgtpuzzles_loopy_7x7_easy'](images/sgtpuzzles_loopy_7x7_easy.gif) | ![Screenshot of 'sgtpuzzles_loopy_7x7_normal'](images/sgtpuzzles_loopy_7x7_normal.gif) | ![Screenshot of 'sgtpuzzles_loopy_7x7_hard'](images/sgtpuzzles_loopy_7x7_hard.gif)

## SGT Puzzles - Net

There are a number of light bulbs, wires and a single light source in the
middle. Connect the wires such that all bulbs are lit up, without loose ends or
loops in the wiring. You can rotate the tiles by clicking on them. If the task
name has a *w* suffix in it then it is allowed to connect wires on opposing
edges of the grid. See the original github repo for more info:
https://github.com/chrisboyle/sgtpuzzles.

<details>
  <summary>Extras returned</summary>

*   `active`:
    -   The number of active/completed cells.
    -   Returned whenever the grid changes.
    -   Has `shape=[1], dtype=INT32`.
*   `total`:
    -   The total number of cells
    -   Returned whenever the grid changes.
    -   Has `shape=[1], dtype=INT32`.
*   `grid`:
    -   The grid cells represented by numbers.
    -   Returned whenever the grid changes.
    -   Has `shape=[grid_size, grid_size], dtype=INT32`.
*   `gridCompleted`:
    -   The grid cells' active/completed status (0 = false, 1 = true).
    -   Returned whenever the grid changes.
    -   Has `shape=[grid_size, grid_size], dtype=INT32`.

</details>

**net_3x3**                                                          | **net_5x5**                                                          | **net_7x7w**                                                           | **net_9x9**                                                          | **net_11x11w**
-------------------------------------------------------------------- | -------------------------------------------------------------------- | ---------------------------------------------------------------------- | -------------------------------------------------------------------- | --------------
![Screenshot of 'sgtpuzzles_net_3x3'](images/sgtpuzzles_net_3x3.gif) | ![Screenshot of 'sgtpuzzles_net_5x5'](images/sgtpuzzles_net_5x5.gif) | ![Screenshot of 'sgtpuzzles_net_7x7w'](images/sgtpuzzles_net_7x7w.gif) | ![Screenshot of 'sgtpuzzles_net_9x9'](images/sgtpuzzles_net_9x9.gif) | ![Screenshot of 'sgtpuzzles_net_11x11w'](images/sgtpuzzles_net_11x11w.gif)

## Shattered Pixel Dungeon

Shattered Pixel Dungeon is a Roguelike RPG, with pixel art graphics and lots of
variety and replayability. Every game is unique, with four different playable
characters, randomized levels and enemies, and over 150 items to collect and
use. The game is simple to get into, but has lots of depth. Strategy is required
if you want to win! See the original github repo for more info:
https://github.com/00-Evan/shattered-pixel-dungeon.git.

<details>
  <summary>Extras returned</summary>
  Returns no extras.
</details>

huntress                                                                                         | mage                                                                                     | rogue                                                                                      | warrior
------------------------------------------------------------------------------------------------ | ---------------------------------------------------------------------------------------- | ------------------------------------------------------------------------------------------ | -------
![Screenshot of 'shattered_pixel_dungeon_huntress'](images/shattered_pixel_dungeon_huntress.gif) | ![Screenshot of 'shattered_pixel_dungeon_mage'](images/shattered_pixel_dungeon_mage.gif) | ![Screenshot of 'shattered_pixel_dungeon_rogue'](images/shattered_pixel_dungeon_rogue.gif) | ![Screenshot of 'shattered_pixel_dungeon_warrior'](images/shattered_pixel_dungeon_warrior.gif)

## Simple Solitaire

This is an Android implementation of
[Solitaire](https://en.wikipedia.org/wiki/Solitaire) card games. We currently
support 19 variants listed here in alphabetical order. Note that the full
task_IDs take the form `simple_solitaire_aces_up`,
`simple_solitaire_calculation` etc. See the original github repo for more info:
https://github.com/TobiasBielefeld/Simple-Solitaire.git.

<details>
  <summary>Extras returned</summary>

*   `card`:
    -   The new visible card `[kind, suit]`:
    -   kind: `a=ace, k=king, q=queen, j=jack, x=10, 2-9=digit`.
    -   suit: `c=clubs, d=diamonds, h=hearts, s=spades`.
    -   Returned when a card is moved.
    -   Has `shape=[2], dtype=STRING_U1`.
*   `stack_i`:
    -   A non-empty stack of visible cards `[kind, suit]`.
    -   `i` different extras (`stack_0`, `stack_1`, `...`), one corresponding to
        each stack.
    -   Returned when a card is moved.
    -   Has `shape=[52, 2], dtype=STRING_U1`.

</details>

**aces_up**                                                                      | **calculation**                                                                          | **canfield**                                                                       | **forty_eight**                                                                          | **freecell**
-------------------------------------------------------------------------------- | ---------------------------------------------------------------------------------------- | ---------------------------------------------------------------------------------- | ---------------------------------------------------------------------------------------- | ------------
![Screenshot of 'simple_solitaire_aces_up'](images/simple_solitaire_aces_up.gif) | ![Screenshot of 'simple_solitaire_calculation'](images/simple_solitaire_calculation.gif) | ![Screenshot of 'simple_solitaire_canfield'](images/simple_solitaire_canfield.gif) | ![Screenshot of 'simple_solitaire_forty_eight'](images/simple_solitaire_forty_eight.gif) | ![Screenshot of 'simple_solitaire_freecell'](images/simple_solitaire_freecell.gif)

**golf**                                                                   | **grandfathers_clock**                                                                                 | **gypsy**                                                                    | **klondike**                                                                       | **maze**
-------------------------------------------------------------------------- | ------------------------------------------------------------------------------------------------------ | ---------------------------------------------------------------------------- | ---------------------------------------------------------------------------------- | --------
![Screenshot of 'simple_solitaire_golf'](images/simple_solitaire_golf.gif) | ![Screenshot of 'simple_solitaire_grandfathers_clock'](images/simple_solitaire_grandfathers_clock.gif) | ![Screenshot of 'simple_solitaire_gypsy'](images/simple_solitaire_gypsy.gif) | ![Screenshot of 'simple_solitaire_klondike'](images/simple_solitaire_klondike.gif) | ![Screenshot of 'simple_solitaire_maze'](images/simple_solitaire_maze.gif)

**mod3**                                                                   | **napoleons_tomb**                                                                             | **pyramid**                                                                      | **simple_simon**                                                                           | **spider**
-------------------------------------------------------------------------- | ---------------------------------------------------------------------------------------------- | -------------------------------------------------------------------------------- | ------------------------------------------------------------------------------------------ | ----------
![Screenshot of 'simple_solitaire_mod3'](images/simple_solitaire_mod3.gif) | ![Screenshot of 'simple_solitaire_napoleons_tomb'](images/simple_solitaire_napoleons_tomb.gif) | ![Screenshot of 'simple_solitaire_pyramid'](images/simple_solitaire_pyramid.gif) | ![Screenshot of 'simple_solitaire_simple_simon'](images/simple_solitaire_simple_simon.gif) | ![Screenshot of 'simple_solitaire_spider'](images/simple_solitaire_spider.gif)

**spiderette**                                                                         | **tri_peaks**                                                                        | **vegas**                                                                    | **yukon**
-------------------------------------------------------------------------------------- | ------------------------------------------------------------------------------------ | ---------------------------------------------------------------------------- | ---------
![Screenshot of 'simple_solitaire_spiderette'](images/simple_solitaire_spiderette.gif) | ![Screenshot of 'simple_solitaire_tri_peaks'](images/simple_solitaire_tri_peaks.gif) | ![Screenshot of 'simple_solitaire_vegas'](images/simple_solitaire_vegas.gif) | ![Screenshot of 'simple_solitaire_yukon'](images/simple_solitaire_yukon.gif)

## Snake

Classic Snake game.

<details>
  <summary>Extras returned</summary>

*   `move`:
    -   The desired direction of movement: `0`=left, `1`=up, `2`=down,
        `3`=right.
    -   Returned whenever the grid changes.
    -   Has `shape=[1], dtype=INT32`.
*   `direction`:
    -   The direction of the snake: `1`=north, `2`=south, `3`=east, `4`=west.
    -   Returned whenever the grid changes.
    -   Has `shape=[1], dtype=INT32`.
*   `grid`:
    -   The grid cells: `x`=border, `' '`=empty, `s`=snake, `a`=apple.
    -   Returned whenever the grid changes.
    -   Has `shape=[13, 19], dtype=STRING_U1`.

</details>

![Screenshot of 'aosp_samples_snake_default'](images/aosp_samples_snake_default.gif)

## Vector Pinball

A simple vector-based Pinball game with realistic physics. See the original
github repo for more info: https://github.com/dozingcat/Vector-Pinball.

<details>
  <summary>Extras returned</summary>
  Returns no extras.
</details>

**vector_pinball_table_1**                                                   | **vector_pinball_table_2**                                                   | **vector_pinball_table_3**
---------------------------------------------------------------------------- | ---------------------------------------------------------------------------- | --------------------------
![Screenshot of 'vector_pinball_table_1'](images/vector_pinball_table_1.gif) | ![Screenshot of 'vector_pinball_table_2'](images/vector_pinball_table_2.gif) | ![Screenshot of 'vector_pinball_table_3'](images/vector_pinball_table_3.gif)

**vector_pinball_table_4**                                                   | **vector_pinball_table_5**
---------------------------------------------------------------------------- | --------------------------
![Screenshot of 'vector_pinball_table_4'](images/vector_pinball_table_4.gif) | ![Screenshot of 'vector_pinball_table_5'](images/vector_pinball_table_5.gif)
<file_sep># coding=utf-8
# Copyright 2021 DeepMind Technologies Limited.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

"""Tests for android_env.components.setup_step_interpreter."""

from absl.testing import absltest
from android_env.components import adb_controller
from android_env.components import errors
from android_env.components import logcat_thread
from android_env.components import setup_step_interpreter
from android_env.proto import task_pb2
import mock

from google.protobuf import text_format


def _to_proto(proto_class, text):
  proto = proto_class()
  text_format.Parse(text, proto)
  return proto


class SetupStepInterpreterTest(absltest.TestCase):

  def setUp(self):
    super().setUp()
    self.logcat = mock.create_autospec(logcat_thread.LogcatThread)
    self.adb_controller = mock.create_autospec(adb_controller.AdbController)

  def test_empty_setup_steps(self):
    """Simple test where nothing should break, and nothing should be done.

    The test simply expects this test to not crash.
    """
    interpreter = setup_step_interpreter.SetupStepInterpreter(
        adb_controller=self.adb_controller, logcat=self.logcat)
    interpreter.interpret([])

  def test_none_setup_steps(self):
    """Simple test where nothing should break, and nothing should be done.

    The test simply expects this test to not crash.
    """
    interpreter = setup_step_interpreter.SetupStepInterpreter(
        adb_controller=self.adb_controller, logcat=self.logcat)
    # Empty setup steps should be ignored.
    interpreter.interpret([None])

  def test_invalid_setup_step(self):
    interpreter = setup_step_interpreter.SetupStepInterpreter(
        adb_controller=self.adb_controller, logcat=self.logcat)
    # Empty setup steps should be ignored.
    with self.assertRaises(AssertionError):
      interpreter.interpret([_to_proto(task_pb2.SetupStep, '')])

  def test_adb_install_apk_filesystem(self):

    interpreter = setup_step_interpreter.SetupStepInterpreter(
        adb_controller=self.adb_controller, logcat=self.logcat)
    interpreter.interpret([
        _to_proto(
            task_pb2.SetupStep, """
adb_call: {
  install_apk: {
    filesystem: {
      path: "/my/favorite/dir/my_apk.apk"
    }
  }
}""")
    ])
    self.adb_controller.install_apk.assert_called_once_with(
        '/my/favorite/dir/my_apk.apk')

  def test_adb_force_stop(self):
    interpreter = setup_step_interpreter.SetupStepInterpreter(
        adb_controller=self.adb_controller, logcat=self.logcat)
    interpreter.interpret([
        _to_proto(
            task_pb2.SetupStep, """
adb_call: { force_stop: { package_name: "my.app.Activity" } }""")
    ])
    # AdbController should be called exactly once with the following arguments.
    self.adb_controller.force_stop.assert_called_once_with('my.app.Activity')

  def test_adb_clear_cache(self):
    interpreter = setup_step_interpreter.SetupStepInterpreter(
        adb_controller=self.adb_controller, logcat=self.logcat)
    interpreter.interpret([
        _to_proto(
            task_pb2.SetupStep, """
adb_call: { clear_cache: { package_name: "my.app.Activity" } }""")
    ])
    # AdbController should be called exactly once with the following arguments.
    self.adb_controller.clear_cache.assert_called_once_with('my.app.Activity')

  def test_adb_grant_permissions(self):

    interpreter = setup_step_interpreter.SetupStepInterpreter(
        adb_controller=self.adb_controller, logcat=self.logcat)
    interpreter.interpret([
        _to_proto(
            task_pb2.SetupStep, """
adb_call: {
  grant_permissions: {
    package_name: "my.app.Activity"
    permissions: [ "my.namespace.READ_DATA", "another.namespace.WRITE" ]
  }
}""")
    ])
    # AdbController should be called exactly once with the following arguments.
    self.adb_controller.grant_permissions.assert_called_once_with(
        'my.app.Activity',
        ['my.namespace.READ_DATA', 'another.namespace.WRITE'])

  def test_adb_start_activity(self):
    interpreter = setup_step_interpreter.SetupStepInterpreter(
        adb_controller=self.adb_controller, logcat=self.logcat)
    interpreter.interpret([
        _to_proto(
            task_pb2.SetupStep, """
adb_call: {
  start_activity: {
    full_activity: "my.app.Activity"
    extra_args: "arg1"
    extra_args: "arg2"
  }
}""")
    ])
    # AdbController should be called exactly once with the following arguments.
    self.adb_controller.start_activity.assert_called_once_with(
        'my.app.Activity', ['arg1', 'arg2'])

  def test_adb_single_tap(self):
    interpreter = setup_step_interpreter.SetupStepInterpreter(
        adb_controller=self.adb_controller, logcat=self.logcat)
    interpreter.interpret([
        _to_proto(task_pb2.SetupStep, """
adb_call: {
  tap: {
    x: 321
    y: 654
  }
}""")
    ])
    # AdbController should be called exactly once with the following arguments.
    self.adb_controller.input_tap.assert_called_once_with(321, 654)

  def test_adb_rotate(self):
    interpreter = setup_step_interpreter.SetupStepInterpreter(
        adb_controller=self.adb_controller, logcat=self.logcat)
    # Check landscape.
    interpreter.interpret([
        _to_proto(task_pb2.SetupStep,
                  """ adb_call: { rotate: { orientation: LANDSCAPE_90 } }""")
    ])
    # AdbController should be called exactly once with the following arguments.
    self.adb_controller.rotate_device.assert_called_once_with(
        task_pb2.AdbCall.Rotate.Orientation.LANDSCAPE_90)

    self.adb_controller.reset_mock()
    # Check portrait.
    interpreter.interpret([
        _to_proto(task_pb2.SetupStep,
                  """ adb_call: { rotate: { orientation: PORTRAIT_0 } }""")
    ])
    # AdbController should be called exactly once with the following arguments.
    self.adb_controller.rotate_device.assert_called_once_with(
        task_pb2.AdbCall.Rotate.Orientation.PORTRAIT_0)

    self.adb_controller.reset_mock()
    # Check landscape inverted.
    interpreter.interpret([
        _to_proto(task_pb2.SetupStep,
                  """ adb_call: { rotate: { orientation: LANDSCAPE_270} }""")
    ])
    # AdbController should be called exactly once with the following arguments.
    self.adb_controller.rotate_device.assert_called_once_with(
        task_pb2.AdbCall.Rotate.Orientation.LANDSCAPE_270)

    self.adb_controller.reset_mock()
    # Check portrait up-side-down.
    interpreter.interpret([
        _to_proto(task_pb2.SetupStep,
                  """ adb_call: { rotate: { orientation: PORTRAIT_180 } }""")
    ])
    # AdbController should be called exactly once with the following arguments.
    self.adb_controller.rotate_device.assert_called_once_with(
        task_pb2.AdbCall.Rotate.Orientation.PORTRAIT_180)

  def test_adb_press_button(self):
    interpreter = setup_step_interpreter.SetupStepInterpreter(
        adb_controller=self.adb_controller, logcat=self.logcat)
    interpreter.interpret([
        _to_proto(task_pb2.SetupStep,
                  """ adb_call: { press_button: { button: HOME } }""")
    ])
    # AdbController should be called exactly once with the following arguments.
    self.adb_controller.input_key.assert_called_once_with('KEYCODE_HOME')
    self.adb_controller.reset_mock()
    interpreter.interpret([
        _to_proto(task_pb2.SetupStep,
                  """ adb_call: { press_button: { button: BACK } }""")
    ])
    # AdbController should be called exactly once with the following arguments.
    self.adb_controller.input_key.assert_called_once_with('KEYCODE_BACK')

  def test_adb_start_accessibility_service(self):
    interpreter = setup_step_interpreter.SetupStepInterpreter(
        adb_controller=self.adb_controller, logcat=self.logcat)
    interpreter.interpret([
        _to_proto(
            task_pb2.SetupStep, """
adb_call: {
  start_accessibility_service: {
    full_service: "my.app.AccessibilityService"
  }
}""")
    ])
    # AdbController should be called exactly once with the following arguments.
    self.adb_controller.start_accessibility_service.assert_called_once_with(
        'my.app.AccessibilityService')

  def test_adb_start_screen_pinning(self):
    interpreter = setup_step_interpreter.SetupStepInterpreter(
        adb_controller=self.adb_controller, logcat=self.logcat)
    interpreter.interpret([
        _to_proto(
            task_pb2.SetupStep, """
adb_call: {
  start_screen_pinning: {
    full_activity: "my.app.HighlanderApp"  # "There can be only one".
  }
}""")
    ])
    # AdbController should be called once with the following arguments.
    self.adb_controller.start_screen_pinning.assert_called_with(
        u'my.app.HighlanderApp')

  @mock.patch('time.sleep')
  def test_time_sleep(self, mock_sleep):
    interpreter = setup_step_interpreter.SetupStepInterpreter(
        adb_controller=self.adb_controller, logcat=self.logcat)
    interpreter.interpret(
        [_to_proto(task_pb2.SetupStep, """sleep: { time_sec: 0.875 }""")])
    assert mock_sleep.call_count == 2
    mock_sleep.assert_has_calls([mock.call(0.875), mock.call(0.5)])

  @mock.patch('time.sleep')
  def test_wait_for_app_screen_empty_activity(self, unused_mock_sleep):
    interpreter = setup_step_interpreter.SetupStepInterpreter(
        adb_controller=self.adb_controller, logcat=self.logcat)
    with self.assertRaises(errors.StepCommandError):
      interpreter.interpret([
          _to_proto(task_pb2.SetupStep,
                    """success_condition: {wait_for_app_screen: { }}""")
      ])

  def test_wait_for_message_fail(self):
    interpreter = setup_step_interpreter.SetupStepInterpreter(
        adb_controller=self.adb_controller, logcat=self.logcat)
    self.assertRaises(errors.StepCommandError, interpreter.interpret, [
        _to_proto(
            task_pb2.SetupStep, """
success_condition: {
  wait_for_message: {
    message:'foo'
    timeout_sec: 0.0001
  }
}
""")
    ])

  def test_wait_for_message_success(self):
    interpreter = setup_step_interpreter.SetupStepInterpreter(
        adb_controller=self.adb_controller, logcat=self.logcat)

    # Replace `LogcatThread.add_event_listener` with one that simply calls `fn`
    # right away, ignoring `event`.
    def mock_add_ev_listener(event_listener):
      event_listener.handler_fn('some_event', 'some_match')

    self.logcat.add_event_listener.side_effect = mock_add_ev_listener
    # The test checks that this command raises no AssertionError.
    interpreter.interpret([
        _to_proto(
            task_pb2.SetupStep, """
success_condition: {
  wait_for_message: {
    message:'foo'
    timeout_sec: 1.0
  }
}
""")
    ])

  @mock.patch('time.sleep')
  def test_check_install_not_installed(self, unused_mock_sleep):
    self.adb_controller.is_package_installed.return_value = False
    interpreter = setup_step_interpreter.SetupStepInterpreter(
        adb_controller=self.adb_controller, logcat=self.logcat)
    with self.assertRaises(errors.StepCommandError):
      interpreter.interpret([
          _to_proto(
              task_pb2.SetupStep, """
success_condition: {
  check_install: {
    package_name: "faz"
    timeout_sec: 0.0001
  }
}
""")
      ])

  def test_check_install_installed(self):
    self.adb_controller.is_package_installed.return_value = True
    interpreter = setup_step_interpreter.SetupStepInterpreter(
        adb_controller=self.adb_controller, logcat=self.logcat)
    # The test checks that this command raises no AssertionError.
    interpreter.interpret([
        _to_proto(
            task_pb2.SetupStep, """
success_condition: {
  check_install: {
    package_name: "baz"
    timeout_sec: 0.0001
  }
}""")
    ])
    self.adb_controller.is_package_installed.assert_called_once_with('baz')

  def test_num_retries_failure(self):
    self.adb_controller.is_package_installed.side_effect = [False] * 3
    interpreter = setup_step_interpreter.SetupStepInterpreter(
        adb_controller=self.adb_controller, logcat=self.logcat)
    with self.assertRaises(errors.StepCommandError):
      interpreter.interpret([
          _to_proto(
              task_pb2.SetupStep, """
success_condition: {
  check_install: {
    package_name: "faz"
    timeout_sec: 0.0001
  }
  num_retries: 3
}""")
      ])
    # We retried 3 times after the first call, so we expect 3+1 calls.
    self.assertEqual(3, self.adb_controller.is_package_installed.call_count)

  @mock.patch('time.sleep')
  def test_num_retries_success(self, unused_mock_sleep):
    self.adb_controller.is_package_installed.side_effect = [
        False, False, True, False
    ]
    interpreter = setup_step_interpreter.SetupStepInterpreter(
        adb_controller=self.adb_controller, logcat=self.logcat)
    interpreter.interpret([
        _to_proto(
            task_pb2.SetupStep, """
success_condition: {
  check_install: {
    package_name: "bar"
    timeout_sec: 0.0001
  }
  num_retries: 5
}""")
    ])
    # The check should succeed on the third try.
    self.assertEqual(3, self.adb_controller.is_package_installed.call_count)

  def test_retry_step(self):
    self.adb_controller.is_package_installed.side_effect = [False, True]
    interpreter = setup_step_interpreter.SetupStepInterpreter(
        adb_controller=self.adb_controller, logcat=self.logcat)
    interpreter.interpret([
        _to_proto(
            task_pb2.SetupStep, """
adb_call: { press_button: { button: HOME } }
success_condition: {
  check_install: {
    package_name: "bar"
    timeout_sec: 0.0001
  }
  num_retries: 2
}""")
    ])
    # We expect the check to fail twice and succeed on the third pass.
    self.adb_controller.input_key.assert_has_calls(
        [mock.call('KEYCODE_HOME')] * 2)
    self.assertEqual(2, self.adb_controller.is_package_installed.call_count)


if __name__ == '__main__':
  absltest.main()
<file_sep># coding=utf-8
# Copyright 2021 DeepMind Technologies Limited.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

"""Converts pixel observation to from int to float32 between 0.0 and 1.0."""

from typing import Dict

from android_env.components import utils
from android_env.wrappers import base_wrapper
import dm_env
from dm_env import specs
import numpy as np


class FloatPixelsWrapper(base_wrapper.BaseWrapper):
  """Wraps AndroidEnv for Panultimate agent."""

  def __init__(self, env: dm_env.Environment):
    super().__init__(env)
    self._should_convert_int_to_float = np.issubdtype(
        self._env.observation_spec()['pixels'].dtype, np.integer)

  def _process_observation(
      self, observation: Dict[str, np.ndarray]
  ) -> Dict[str, np.ndarray]:
    if self._should_convert_int_to_float:
      float_pixels = utils.convert_int_to_float(
          observation['pixels'],
          self._env.observation_spec()['pixels'],
          np.float32)
      observation['pixels'] = float_pixels
    return observation

  def _process_timestep(self, timestep: dm_env.TimeStep) -> dm_env.TimeStep:
    step_type, reward, discount, observation = timestep
    return dm_env.TimeStep(
        step_type=step_type,
        reward=reward,
        discount=discount,
        observation=self._process_observation(observation))

  def reset(self) -> dm_env.TimeStep:
    timestep = self._env.reset()
    return self._process_timestep(timestep)

  def step(self, action: Dict[str, np.ndarray]) -> dm_env.TimeStep:
    timestep = self._env.step(action)
    return self._process_timestep(timestep)

  def observation_spec(self) -> Dict[str, specs.Array]:
    if self._should_convert_int_to_float:
      observation_spec = self._env.observation_spec()
      observation_spec['pixels'] = specs.BoundedArray(
          shape=self._env.observation_spec()['pixels'].shape,
          dtype=np.float32,
          minimum=0.0,
          maximum=1.0,
          name=self._env.observation_spec()['pixels'].name)
      return observation_spec
    return self._env.observation_spec()
<file_sep># coding=utf-8
# Copyright 2021 DeepMind Technologies Limited.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

"""Wraps the AndroidEnv environment to make its interface flat."""

from typing import Union, Dict, Any

from android_env.wrappers import base_wrapper
import dm_env
from dm_env import specs
import numpy as np

RGB_CHANNELS = (0, 1, 2)


def _extract_screen_pixels(obs: np.ndarray):
  """Get only screen pixels by removing previous action layer."""
  is_grayscale_image = obs.shape[-1] == 2
  if is_grayscale_image:
    return np.expand_dims(obs[..., 0], -1)
  return obs[..., RGB_CHANNELS]


def _get_no_action_observation_spec(obs_spec: specs.BoundedArray):
  """Create an observation spec without the action layer."""
  shape = np.array(obs_spec.shape)
  shape[2] -= 1
  minimum = obs_spec.minimum
  maximum = obs_spec.maximum
  is_scalar = lambda x: np.isscalar(x) or np.ndim(x) == 0
  if not is_scalar(minimum):
    minimum = _extract_screen_pixels(minimum)
  if not is_scalar(maximum):
    maximum = _extract_screen_pixels(maximum)
  return obs_spec.replace(shape=shape, minimum=minimum, maximum=maximum)


class FlatInterfaceWrapper(base_wrapper.BaseWrapper):
  """Simple interface for AndroidEnv.

  Removes the structure from observations and actions, keeping only the pixel
  observations. Also exposes action as an int32 scalar, making it easier to use
  with conventional discrete agents. This wrapper expects a discretized action
  space.
  """

  def __init__(self,
               env: dm_env.Environment,
               flat_actions: bool = True,
               flat_observations: bool = True,
               keep_action_layer: bool = True):
    super().__init__(env)
    self._flat_actions = flat_actions
    self._flat_observations = flat_observations
    self._keep_action_layer = keep_action_layer
    self._action_name = list(self._env.action_spec())[0]
    self._assert_base_env()

  def _assert_base_env(self):
    base_action_spec = self._env.action_spec()
    assert len(base_action_spec) == 1, self._env.action_spec()
    assert isinstance(base_action_spec, dict)
    assert isinstance(base_action_spec[self._action_name], specs.BoundedArray)

  def _process_action(self, action: Union[int, np.ndarray, Dict[str, Any]]):
    if self._flat_actions:
      return {self._action_name: action}
    else:
      return action

  def _process_timestep(self, timestep: dm_env.TimeStep) -> dm_env.TimeStep:
    if self._flat_observations:
      step_type, reward, discount, observation = timestep
      # Keep only the pixels.
      pixels = observation['pixels']
      pixels = pixels if self._keep_action_layer else _extract_screen_pixels(
          pixels)
      return dm_env.TimeStep(
          step_type=step_type,
          reward=reward,
          discount=discount,
          observation=pixels)
    else:
      return timestep

  def reset(self) -> dm_env.TimeStep:
    timestep = self._env.reset()
    return self._process_timestep(timestep)

  def step(self, action: int) -> dm_env.TimeStep:
    timestep = self._env.step(self._process_action(action))
    return self._process_timestep(timestep)

  def observation_spec(self) -> specs.Array:
    if self._flat_observations:
      pixels_spec = self._env.observation_spec()['pixels']
      if not self._keep_action_layer:
        return _get_no_action_observation_spec(pixels_spec)
      return pixels_spec
    else:
      return self._env.observation_spec()

  def action_spec(self) -> specs.Array:
    if self._flat_actions:
      return self._env.action_spec()[self._action_name]
    else:
      return self._env.action_spec()
<file_sep># coding=utf-8
# Copyright 2021 DeepMind Technologies Limited.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

"""A class the launches a thread to read Android logcat outputs."""

import re
import subprocess
import threading
# `typing.Pattern` has been deprecated in Python 3.9 in favor of `re.Pattern`,
# but it is not available even in slightly older Python versions.
# Please see https://www.python.org/dev/peps/pep-0585/
from typing import Callable, List, Match, NamedTuple, Optional, Pattern

from absl import logging
from android_env.components import thread_function
from android_env.proto import task_pb2


class EventListener(NamedTuple):
  regexp: Pattern[str]
  handler_fn: Callable[[Pattern[str], Match[str]], None]


class LogcatThread(thread_function.ThreadFunction):
  """Reads ADB logcat entries in a separate thread."""

  def __init__(
      self,
      adb_command_prefix: List[str],
      log_parsing_config: task_pb2.LogParsingConfig,
      print_all_lines: bool = False,
      name: str = 'logcat',
  ):
    """Initializes this LogcatThread with optional filters.

    Please see https://developer.android.com/studio/command-line/logcat for more
    info on `logcat`.

    Args:
      adb_command_prefix: Command for connecting to a particular ADB.
      log_parsing_config: Determines the types of messages we want logcat to
        match. Contains `filters` and `log_regexps`.
      print_all_lines: Whether to print all lines we observe in the logcat
        stream. This is useful to debug problems in Android itself.
      name: Name of the thread.
    """

    self._print_all_lines = print_all_lines
    self._regexps = log_parsing_config.log_regexps
    self._listeners = {}
    self._desired_event = None
    self._thread_event = threading.Event()
    self._max_buffer_size = 100

    filters = list(log_parsing_config.filters) + ['*:S']
    cmd = adb_command_prefix + ['logcat', '-v', 'epoch'] + filters
    logging.info('Logcat command: %s', ' '.join(cmd))

    self._proc = subprocess.Popen(
        cmd,
        stdout=subprocess.PIPE,
        stderr=subprocess.STDOUT,
        bufsize=1,
        universal_newlines=True)
    self._stdout = self._proc.stdout

    super().__init__(block_input=True, block_output=False, name=name)

  def add_event_listener(self, event_listener: EventListener) -> None:
    """Adds `fn` to the list of handlers to call when `event` occurs."""
    event_regexp = event_listener.regexp
    if event_regexp not in self._listeners:
      self._listeners[event_regexp] = []
    self._listeners[event_regexp].append(event_listener.handler_fn)

  def remove_event_listener(self, event_listener: EventListener) -> None:
    """Removes `fn` from the list of handlers to call when `event` occurs."""
    event_regexp = event_listener.regexp
    if event_regexp not in self._listeners:
      logging.error('Event: %r is not registered.', event_regexp)
      return
    self._listeners[event_regexp].remove(event_listener.handler_fn)

  def wait(self,
           event: Optional[Pattern[str]] = None,
           timeout_sec: Optional[float] = None) -> None:
    """Blocks (caller) execution for up to `timeout_sec` until `event` is fired.

    Args:
      event: Event to wait for. If None, any new event will cause this function
        to return.
      timeout_sec: Maximum time to block waiting for an event.
    """
    self._desired_event = event
    self._thread_event.wait(timeout=timeout_sec)
    self._thread_event.clear()

  def kill(self):
    self._proc.kill()
    super().kill()

  def main(self) -> None:
    # pylint: disable=g-line-too-long
    # Format is: "TIME_SEC PID TID PRIORITY TAG: MESSAGE"
    #
    # Example:
    #  '         1553110400.424  5583  5658 D NostalgicRacer: com.google.example.games.nostalgicracer.views.renderers.OpenGLRenderDriver@912fb8.onSurfaceChanged 480x320'    #
    #
    # If a log_prefix is given, then the format becomes:
    # "TIME_SEC PID TID PRIORITY TAG: LOG_PREFIX MESSAGE"
    # pylint: enable=g-line-too-long

    regexp = r"""
      ^                                   # Beginning of the line.
      [ ]+(?P<timestamp>[0-9]+\.[0-9]+)   # Spaces and a float.
      [ ]+(?P<pid>[0-9]+)                 # Spaces and an int.
      [ ]+(?P<tid>[0-9]+)                 # Spaces and an int.
      [ ]+(?P<priority>.)                 # Spaces and any single character.
      [ ]+(?P<tag>[^:]*):                 # Spaces and any char that's not ':'.
    """

    regexp += r"""[ ](?P<message>.*)$"""
    logline_re = re.compile(regexp, re.VERBOSE)

    for line in self._stdout:
      # We never hand back control to ThreadFunction._run() so we need to
      # explicitly check for self._should_run here.
      if not self._should_run:
        break

      if self._print_all_lines:
        logging.info('line: %r', line)

      if not line:  # Skip empty lines.
        continue

      # We're currently only consuming `message`, but we may use the other
      # fields in the future.
      matches = logline_re.match(line)
      if not matches or len(matches.groups()) != 6:
        continue

      content = matches.group('message')
      for ev, listeners in self._listeners.items():
        ev_matches = ev.match(content)
        if ev_matches:
          # Unblock consumers that may be waiting for events.
          if not self._thread_event.is_set():
            if self._desired_event:
              if self._desired_event == ev:
                self._thread_event.set()
            else:
              self._thread_event.set()

          # Notify listeners.
          for listener in listeners:
            listener(ev, ev_matches)
<file_sep># coding=utf-8
# Copyright 2021 DeepMind Technologies Limited.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

"""Tests for android_env.components.emulator_simulator."""

from absl.testing import absltest
from android_env.components import action_type
from android_env.components import adb_controller
from android_env.components import emulator_launcher
from android_env.components import emulator_simulator
from android_env.proto import emulator_controller_pb2
from android_env.proto import emulator_controller_pb2_grpc

import grpc
import mock
import numpy as np
from PIL import Image


class EmulatorSimulatorTest(absltest.TestCase):

  def setUp(self):
    super().setUp()
    self.addCleanup(mock.patch.stopall)  # Disable previous patches.

    self._adb_controller = mock.create_autospec(adb_controller.AdbController)
    self._launcher = mock.create_autospec(emulator_launcher.EmulatorLauncher)
    self._emulator_stub = mock.create_autospec(
        emulator_controller_pb2_grpc.EmulatorControllerStub)
    self._grpc_channel = mock.create_autospec(grpc.Channel)

    mock.patch.object(
        adb_controller, 'AdbController',
        return_value=self._adb_controller).start()
    mock.patch.object(
        emulator_launcher, 'EmulatorLauncher',
        return_value=self._launcher).start()

  @mock.patch('grpc.aio.insecure_channel')
  @mock.patch('grpc.insecure_channel')
  def test_adb_device_name_not_empty(self, aio_channel, channel):
    tmp_dir = absltest.get_default_test_tmpdir()
    aio_channel.return_value = self._grpc_channel
    channel.return_value = self._grpc_channel
    simulator = emulator_simulator.EmulatorSimulator(
        emulator_launcher_args={'grpc_port': 1234},
        emulator_console_args={},
        adb_path='/my/adb',
        adb_server_port=5037,
        tmp_dir=tmp_dir,
        prompt_regex='awesome>')
    self.assertNotEmpty(simulator.adb_device_name())

  @mock.patch('grpc.aio.insecure_channel')
  @mock.patch('grpc.insecure_channel')
  def test_close(self, aio_channel, channel):
    tmp_dir = absltest.get_default_test_tmpdir()
    aio_channel.return_value = self._grpc_channel
    channel.return_value = self._grpc_channel
    simulator = emulator_simulator.EmulatorSimulator(
        emulator_launcher_args={'grpc_port': 1234},
        emulator_console_args={},
        adb_path='/my/adb',
        adb_server_port=5037,
        tmp_dir=tmp_dir,
        prompt_regex='awesome>')

    # The simulator should launch and not crash.
    self._adb_controller.get_screen_dimensions.return_value = (1234, 5678)
    simulator.launch()

    # For whatever reason clients may want to close the EmulatorSimulator.
    # We just want to check that the simulator does not crash and/or leak
    # resources.
    simulator.close()

  @mock.patch('grpc.aio.insecure_channel')
  @mock.patch('grpc.insecure_channel')
  def test_restart(self, aio_channel, channel):
    tmp_dir = absltest.get_default_test_tmpdir()
    aio_channel.return_value = self._grpc_channel
    channel.return_value = self._grpc_channel
    simulator = emulator_simulator.EmulatorSimulator(
        emulator_launcher_args={'grpc_port': 1234},
        emulator_console_args={},
        adb_path='/my/adb',
        adb_server_port=5037,
        tmp_dir=tmp_dir,
        prompt_regex='awesome>')

    # The simulator should launch and not crash.
    self._adb_controller.get_screen_dimensions.return_value = (1234, 5678)
    simulator.launch()

    # For whatever reason clients may want to restart the EmulatorSimulator.
    simulator.restart()

  @mock.patch('grpc.aio.insecure_channel')
  @mock.patch('grpc.insecure_channel')
  def test_get_observation(self, aio_channel, channel):
    tmp_dir = absltest.get_default_test_tmpdir()
    aio_channel.return_value = self._grpc_channel
    channel.return_value = self._grpc_channel
    simulator = emulator_simulator.EmulatorSimulator(
        emulator_launcher_args={'grpc_port': 1234},
        emulator_console_args={},
        adb_path='/my/adb',
        adb_server_port=5037,
        tmp_dir=tmp_dir,
        prompt_regex='awesome>')

    # The simulator should launch and not crash.
    self._adb_controller.get_screen_dimensions.return_value = (1234, 5678)
    simulator.launch()

    simulator._emulator_stub.getScreenshot = mock.MagicMock(
        return_value=emulator_controller_pb2.Image(
            format=emulator_controller_pb2.ImageFormat(width=5678, height=1234),
            image=Image.new('RGB', (1234, 5678)).tobytes(),
            timestampUs=123))

    observation = simulator.get_observation()
    # The observation should have three components:
    #   - an image
    #   - the timedelta
    #   - the orientation.
    self.assertLen(observation, 3)
    # The first element (the "image") should have the same screen dimensions as
    # reported by ADB and it should have 3 channels (RGB).
    self.assertEqual(observation['pixels'].shape, (1234, 5678, 3))
    self.assertEqual(observation['timedelta'], 123)

  @mock.patch('grpc.aio.insecure_channel')
  @mock.patch('grpc.insecure_channel')
  def test_send_action(self, aio_channel, channel):
    tmp_dir = absltest.get_default_test_tmpdir()
    aio_channel.return_value = self._grpc_channel
    channel.return_value = self._grpc_channel
    simulator = emulator_simulator.EmulatorSimulator(
        emulator_launcher_args={'grpc_port': 1234},
        emulator_console_args={},
        adb_path='/my/adb',
        adb_server_port=5037,
        tmp_dir=tmp_dir,
        prompt_regex='awesome>')

    # The simulator should launch and not crash.
    self._adb_controller.get_screen_dimensions.return_value = (5000, 1000)
    simulator.launch()

    simulator._emulator_stub.sendTouch = mock.MagicMock(return_value=None)

    simulator.send_action(
        {'action_type': np.array([action_type.ActionType.TOUCH]),
         'touch_position': np.array([0.25, 0.75])})
    simulator.send_action(
        {'action_type': np.array([action_type.ActionType.TOUCH]),
         'touch_position': np.array([0.75, 0.50])})
    simulator.send_action(
        {'action_type': np.array([action_type.ActionType.LIFT]),
         'touch_position': np.array([0.66, 0.33])})
    # We expect EmulatorSimulator to send the following calls:
    # 1st call:
    #     x-coordinate: 10000 * 0.25 = 250
    #     y-coordinate: 50000 * 0.75 = 3750
    #     down: True  # It's a touch command.
    # 2nd call:
    #     x-coordinate: 10000 * 0.75 = 750
    #     y-coordinate: 50000 * 0.50 = 2500
    #     down: True  # It's a touch command.
    # 3rd call:
    #     x-coordinate: 0
    #     y-coordinate: 0
    #     down: False  # It's a lift command.

if __name__ == '__main__':
  absltest.main()
<file_sep># coding=utf-8
# Copyright 2021 DeepMind Technologies Limited.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

"""Wraps the AndroidEnv environment to provide tap actions of a given duration."""

from typing import Dict, Sequence

from android_env.components import action_type
from android_env.wrappers import base_wrapper
import dm_env
import numpy as np


ActionType = action_type.ActionType


class TapActionWrapper(base_wrapper.BaseWrapper):
  """AndroidEnv with tap actions."""

  def __init__(self,
               env: dm_env.Environment,
               num_frames: int = 5,
               touch_only: bool = False):
    super().__init__(env)
    assert 'action_type' in env.action_spec()
    self._touch_only = touch_only
    self._num_frames = num_frames
    self._env_steps = 0

  def android_logs(self):
    """Returns a dictionary of metrics logged by the environment."""
    logs = self._env.android_logs()
    logs.update({'env_steps': self._env_steps})
    return logs

  def _process_action(
      self, action: Dict[str, np.ndarray]
  ) -> Sequence[Dict[str, np.ndarray]]:

    if self._touch_only:
      assert action['action_type'] == 0
      touch_action = action.copy()
      touch_action['action_type'] = np.array(ActionType.TOUCH).astype(
          self.action_spec()['action_type'].dtype)
      actions = [touch_action] * self._num_frames
      lift_action = action.copy()
      lift_action['action_type'] = np.array(ActionType.LIFT).astype(
          self.action_spec()['action_type'].dtype)
      actions.append(lift_action)

    else:
      if action['action_type'] == ActionType.TOUCH:
        actions = [action] * self._num_frames
        lift_action = action.copy()
        lift_action['action_type'] = np.array(ActionType.LIFT).astype(
            self.action_spec()['action_type'].dtype)
        actions.append(lift_action)
      else:
        actions = [action] * (self._num_frames + 1)

    return actions

  def step(self, action: Dict[str, np.ndarray]) -> dm_env.TimeStep:
    """Takes a step in the environment."""
    self._env_steps += self._num_frames + 1
    actions = self._process_action(action)
    total_reward = 0
    for idx in range(len(actions)):
      step_type, reward, discount, observation = self._env.step(actions[idx])
      if reward:
        total_reward += reward
      if step_type == dm_env.StepType.LAST:
        return dm_env.TimeStep(
            step_type=step_type,
            reward=total_reward,
            discount=discount,
            observation=observation)
    return dm_env.TimeStep(
        step_type=step_type,
        reward=total_reward,
        discount=discount,
        observation=observation)

  def action_spec(self) -> Dict[str, dm_env.specs.Array]:
    if self._touch_only:
      return {
          'action_type':
              dm_env.specs.DiscreteArray(num_values=1, name='action_type'),
          'touch_position':
              self._env.action_spec()['touch_position'],
      }
    else:
      return self._env.action_spec()

<file_sep># coding=utf-8
# Copyright 2021 DeepMind Technologies Limited.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

"""Determines if the current app screen matches an expected app screen."""

import enum
import re
from typing import Callable, List, Optional, Sequence, Pattern

from absl import logging

from android_env.components import adb_controller as adb_control
from android_env.proto import task_pb2


class DumpsysNode():
  """A node in a dumpsys tree."""

  def __init__(self, data: Optional[str] = None):
    self._children = []
    self._data = data

  @property
  def data(self) -> str:
    return self._data

  @property
  def children(self) -> List['DumpsysNode']:
    return self._children

  def find_child(self,
                 predicate: Callable[['DumpsysNode'], bool],
                 max_levels: int = 0) -> Optional['DumpsysNode']:
    """Returns the first direct child that matches `predicate`, None otherwise.

    Args:
      predicate: Function-like that accepts a DumpsysNode and returns boolean.
      max_levels: Maximum number of levels down the tree to search for a child.
        If non-positive, only direct children will be searched for.

    Returns:
      A DumpsysNode or None.
    """
    if not self.children:
      return None

    try:
      return next(x for x in self.children if predicate(x))
    except StopIteration:
      logging.info('Failed to find child. max_levels: %i.', max_levels)
      # Search children.
      if max_levels:
        for child in self.children:
          child_result = child.find_child(predicate, max_levels - 1)
          if child_result is not None:
            return child_result

      return None

  def __repr__(self):
    return self._data

  def print_tree(self, indent: int = 2):
    """Prints this tree in logging.info()."""
    logging.info(' ' * indent + self.data)
    for c in self.children:
      c.print_tree(indent + 2)


def build_tree_from_dumpsys_output(dumpsys_output: str) -> DumpsysNode:
  """Constructs a tree from a dumpsys string output.

  Args:
    dumpsys_output: string Verbatim output from adb dumpsys. The expected format
      is a list where each line is a node and the indentation marks the
      relationship with its parent or sibling.

  Returns:
    DumpsysNode The root of the tree.
  """
  lines = dumpsys_output.split('\n')  # Split by lines.
  lines = [x.rstrip(' \r') for x in lines]
  lines = [x for x in lines if len(x)]  # Remove empty lines.

  root = DumpsysNode('___root___')  # The root of all nodes.
  parents_stack = [root]
  for line in lines:
    stripped_line = line.lstrip(' ')
    indent = len(line) - len(stripped_line)  # Number of indent spaces.
    new_node = DumpsysNode(stripped_line)  # Create a node without indentation.

    parent = parents_stack.pop()
    if parent.data == '___root___':  # The root is an exception for indentation.
      parent_indent = -2
    else:
      parent_indent = (len(parents_stack) - 1) * 2

    if indent == parent_indent:  # `new_node` is a sibiling.
      parent = parents_stack.pop()
    elif indent < parent_indent:  # Indentation reduced (i.e. a block finished)
      num_levels = (indent // 2) + 1
      parents_stack = parents_stack[:num_levels]
      parent = parents_stack.pop()
    elif indent > parent_indent:  # `new_node` is a child.
      pass  # No need to change the current parent.

    parent.children.append(new_node)
    parents_stack.append(parent)
    parents_stack.append(new_node)

  return root


def matches_path(dumpsys_activity_output: str,
                 expected_view_hierarchy_path: Sequence[Pattern[str]],
                 max_levels: int = 0) -> bool:
  """Returns True if the current dumpsys output matches the expected path.

  Args:
    dumpsys_activity_output: The output of running `dumpsys activity ...`.
    expected_view_hierarchy_path: [regex] A list of regular expressions to be
      tested at each level of the tree.
    max_levels: How many levels to search from root for View Hierarchy.

  Returns:
    True if the dumpsys tree contains one path that matches all regexes.
  """
  root = build_tree_from_dumpsys_output(dumpsys_activity_output)

  # Find the View Hierarchy.
  view_hierarchy = root.find_child(
      lambda x: x.data.startswith('View Hierarchy'), max_levels)
  if view_hierarchy is None:
    logging.error(
        'view_hierarchy is None. Dumpsys activity output: %s. tree: %r',
        str(dumpsys_activity_output), root.print_tree())
    logging.error('Tree root: %s', str(root))
    return None

  current_node = view_hierarchy
  for i, regex in enumerate(expected_view_hierarchy_path):

    def regex_predicate(node, expr=regex):
      matches = expr.match(node.data)
      return matches is not None

    child = current_node.find_child(regex_predicate)
    if child is None:
      logging.error('Mismatched regex (%i, %s). current_node: %s', i,
                    regex.pattern, current_node)
      logging.error('Dumpsys activity output: %s', str(dumpsys_activity_output))
      logging.error('Tree root: %s', str(root))
      return None
    else:
      current_node = child
  return True


class AppScreenChecker():
  """Checks that the current app screen matches an expected screen."""

  class Outcome(enum.IntEnum):
    """Possible return vales from checking the current app screen."""
    # The current app screen matches the expected app screen.
    SUCCESS = 0
    # There's no activity to check.
    EMPTY_EXPECTED_ACTIVITY = 1
    # We were unable to determine the current activity.
    FAILED_ACTIVITY_EXTRACTION = 2
    # The current activity does not match the expected activity.
    UNEXPECTED_ACTIVITY = 3
    # The current view hierarchy does not match the expected view hierarchy.
    UNEXPECTED_VIEW_HIERARCHY = 4

  def __init__(self,
               adb_controller: adb_control.AdbController,
               expected_app_screen: task_pb2.AppScreen):
    self._adb_controller = adb_controller
    self._expected_activity = expected_app_screen.activity
    self._expected_view_hierarchy_path = [
        re.compile(regex) for regex in expected_app_screen.view_hierarchy_path
    ]

  # Return type is AppScreenChecker.Outcome, but pytype doesn't understand that.
  def matches_current_app_screen(self) -> enum.IntEnum:
    """Determines whether the current app screen matches `expected_app_screen`."""
    if not self._expected_activity:
      return AppScreenChecker.Outcome.EMPTY_EXPECTED_ACTIVITY

    # Check if we are still on the expected Activity.
    current_activity = self._adb_controller.get_current_activity()
    if current_activity is None:
      return AppScreenChecker.Outcome.FAILED_ACTIVITY_EXTRACTION

    if current_activity != self._expected_activity:
      logging.error('current_activity: %s,  expected_activity: %s',
                    current_activity, self._expected_activity)
      return AppScreenChecker.Outcome.UNEXPECTED_ACTIVITY

    # Extract just the package name from the full activity name.
    package_name = self._expected_activity.split('/')[0]

    # Check if we are in the expected view hierarchy path.
    if self._expected_view_hierarchy_path:
      dumpsys_activity_output = self._adb_controller.get_activity_dumpsys(
          package_name)
      if dumpsys_activity_output:
        if not matches_path(
            dumpsys_activity_output,
            self._expected_view_hierarchy_path,
            max_levels=3):
          return AppScreenChecker.Outcome.UNEXPECTED_VIEW_HIERARCHY

    return AppScreenChecker.Outcome.SUCCESS
<file_sep># coding=utf-8
# Copyright 2021 DeepMind Technologies Limited.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

"""A component that parses and processes SetupSteps."""

import copy
import random
import re
import time
from typing import Any, Dict, Optional, Sequence

from absl import logging
from android_env.components import adb_controller as adb_control
from android_env.components import app_screen_checker
from android_env.components import errors
from android_env.components import logcat_thread
from android_env.proto import task_pb2


class SetupStepInterpreter():
  """An interpreter for SetupSteps."""

  def __init__(self,
               adb_controller: adb_control.AdbController,
               logcat: logcat_thread.LogcatThread):
    """Initializes this interpreter.

    Args:
      adb_controller: An object to communicate with Android via ADB.
      logcat: A LogcatThread instance connected to the same Android simulator
        as AdbController.
    """
    self._adb_controller = adb_controller
    self._logcat_thread = logcat
    self._last_activity = ''
    self._log_dict = {
        'error_count_adb_call': 0,
        'error_count_wait_for_app_screen': 0,
        'error_count_check_install': 0,
        'error_count_wait_for_message': 0,
        'total_time_waiting_for_app_screen': 0
    }

  def log_dict(self) -> Dict[str, Any]:
    return copy.deepcopy(self._log_dict)

  def interpret(self, setup_steps: Sequence[task_pb2.SetupStep]) -> None:
    """Returns True if parsing and processing `setup_steps` is successful."""
    if setup_steps:
      logging.info('Executing setup steps: %s', setup_steps)
      for step in setup_steps:
        self._process_step_command(step)
      logging.info('Done executing setup steps.')

  def _process_step_command(self, step_cmd: task_pb2.SetupStep) -> None:
    """Processes a single step command from a reset or extra setup."""

    if not step_cmd:
      logging.info('Empty step_cmd')
      return

    logging.info('Executing step_cmd: %r', step_cmd)
    step_type = step_cmd.WhichOneof('step')
    success_condition = step_cmd.success_condition
    success_check = success_condition.WhichOneof('check')
    assert step_type or success_check, (
        'At least one of step and success_condition must be defined.')

    num_tries = 0
    max_retries = max(success_condition.num_retries, 3)
    while num_tries < max_retries:

      num_tries += 1

      try:
        self._execute_step_cmd(step_cmd, step_type)
        time.sleep(0.5)
        self._check_success(success_check, success_condition)
        return

      except NotImplementedError:
        logging.exception('Not implemented error! Skipping this step command.')
        return

      except errors.AdbControllerError:
        self._log_dict['error_count_adb_call'] += 1
        logging.warning('ADB call [%r] has failed. Try %d of %d.',
                        step_cmd.adb_call, num_tries, max_retries)

      except errors.WaitForAppScreenError:
        self._log_dict['error_count_wait_for_app_screen'] += 1
        logging.warning('Failed to wait for app screen. Try %d of %d.',
                        num_tries, max_retries)

      except errors.WaitForMessageError:
        self._log_dict['error_count_wait_for_message'] += 1
        logging.warning('Failed to wait for message. Try %d of %d.',
                        num_tries, max_retries)

      except errors.CheckInstallError:
        self._log_dict['error_count_check_install'] += 1
        logging.warning('Package [%r] not installed. Try %d of %d.',
                        success_condition.check_install.package_name,
                        num_tries, max_retries)

    raise errors.StepCommandError('Step failed: [%r]' % step_cmd)

  def _execute_step_cmd(self,
                        step_cmd: task_pb2.SetupStep,
                        step_type: Optional[str]) -> None:
    """Executes a step command of given type."""

    if not step_type:
      return

    if step_type == 'sleep':
      time.sleep(step_cmd.sleep.time_sec)
    elif step_type == 'adb_call':
      self._parse_adb_call(step_cmd.adb_call)
    else:
      raise NotImplementedError('No step command of type [%s].' % step_type)

  def _check_success(self,
                     success_check: Optional[str],
                     success_condition: task_pb2.SuccessCondition) -> None:
    """Checks whether the given success condition was met."""

    if not success_check:
      return

    if success_check == 'wait_for_app_screen':
      self._wait_for_app_screen(success_condition.wait_for_app_screen)
    elif success_check == 'check_install':
      self._check_install(success_condition.check_install)
    elif success_check == 'wait_for_message':
      self._wait_for_message(success_condition.wait_for_message)
    else:
      raise NotImplementedError('No success check called [%s].' % success_check)

  def _parse_adb_call(self, adb_cmd: task_pb2.AdbCall) -> None:
    """Parses an adb command into set of allowed calls."""

    call_type = adb_cmd.WhichOneof('command')
    logging.info('Parsing ADB call of type: %s', call_type)

    if call_type == 'tap':
      tap = adb_cmd.tap
      self._adb_controller.input_tap(tap.x, tap.y)

    elif call_type == 'rotate':
      self._adb_controller.rotate_device(adb_cmd.rotate.orientation)

    elif call_type == 'press_button':
      if (adb_cmd.press_button.button ==
          task_pb2.AdbCall.PressButton.Button.HOME):
        self._adb_controller.input_key('KEYCODE_HOME')
      elif (adb_cmd.press_button.button ==
            task_pb2.AdbCall.PressButton.Button.BACK):
        self._adb_controller.input_key('KEYCODE_BACK')

    elif call_type == 'start_random_activity':
      self._last_activity = random.choice(
          list(adb_cmd.start_random_activity.activity_list))
      logging.info('Random activity: %s', self._last_activity)
      self._adb_controller.start_activity(
          self._last_activity, list(adb_cmd.start_random_activity.extra_args))
      wait_proto = task_pb2.WaitForAppScreen()
      wait_proto.app_screen.activity = self._last_activity
      wait_proto.timeout_sec = adb_cmd.start_random_activity.timeout_sec
      self._wait_for_app_screen(wait_proto)

    elif call_type == 'start_activity':
      self._adb_controller.start_activity(
          adb_cmd.start_activity.full_activity,
          list(adb_cmd.start_activity.extra_args))

    elif call_type == 'start_intent':
      self._adb_controller.start_intent(
          action=adb_cmd.start_intent.action,
          data_uri=adb_cmd.start_intent.data_uri,
          package_name=adb_cmd.start_intent.package_name)

    elif call_type == 'force_stop':
      self._adb_controller.force_stop(adb_cmd.force_stop.package_name)

    elif call_type == 'force_stop_random_activity':
      if self._last_activity:
        package_name = self._last_activity.split('/')[0]
        logging.info('Force stop package (%s)', package_name)
        self._adb_controller.force_stop(package_name)

    elif call_type == 'clear_cache':
      self._adb_controller.clear_cache(adb_cmd.clear_cache.package_name)

    elif call_type == 'clear_cache_random_activity':
      if self._last_activity:
        package_name = self._last_activity.split('/')[0]
        logging.info('Clear cache package (%s)', package_name)
        self._adb_controller.force_stop(package_name)

    elif call_type == 'grant_permissions':
      self._adb_controller.grant_permissions(
          adb_cmd.grant_permissions.package_name,
          adb_cmd.grant_permissions.permissions)

    elif call_type == 'install_apk':
      install_apk_cmd = adb_cmd.install_apk
      location_type = install_apk_cmd.WhichOneof('location')
      logging.info('location_type: %s', location_type)
      if location_type == 'filesystem':
        self._adb_controller.install_apk(install_apk_cmd.filesystem.path)
      else:
        logging.error('Unsupported location type: %r', install_apk_cmd)

    elif call_type == 'start_accessibility_service':
      self._adb_controller.start_accessibility_service(
          adb_cmd.start_accessibility_service.full_service)

    elif call_type == 'start_screen_pinning':
      self._adb_controller.start_screen_pinning(
          adb_cmd.start_screen_pinning.full_activity)

    elif call_type == 'disable_animations':
      self._adb_controller.disable_animations()

    else:
      raise NotImplementedError('No ADB call type [%s].' % call_type)

  def _wait_for_app_screen(
      self, wait_for_app_screen: task_pb2.WaitForAppScreen) -> None:
    """Waits for a given `app_screen` to be the current screen."""

    logging.info('Waiting for app screen...')
    app_screen = wait_for_app_screen.app_screen
    screen_checker = app_screen_checker.AppScreenChecker(
        self._adb_controller, app_screen)

    start_time = time.time()
    while time.time() - start_time < wait_for_app_screen.timeout_sec:
      if (screen_checker.matches_current_app_screen() ==
          app_screen_checker.AppScreenChecker.Outcome.SUCCESS):
        wait_time = time.time() - start_time
        self._log_dict['total_time_waiting_for_app_screen'] += wait_time
        logging.info('Successfully waited for app screen in %r seconds: [%r]',
                     wait_time, app_screen)
        return
      time.sleep(0.1)

    wait_time = time.time() - start_time
    self._log_dict['total_time_waiting_for_app_screen'] += wait_time
    logging.error('Failed to wait for app screen in %r seconds: [%r].',
                  wait_time, app_screen)

    raise errors.WaitForAppScreenError()

  def _wait_for_message(
      self, wait_for_message: task_pb2.WaitForMessage) -> None:
    """Waits for a given message in logcat."""

    message = wait_for_message.message
    logging.info('Waiting for message: %s...', message)
    event = re.compile(message)
    got_message = False

    def f(ev, match):
      del ev, match
      nonlocal got_message
      got_message = True

    listener = logcat_thread.EventListener(regexp=event, handler_fn=f)
    self._logcat_thread.add_event_listener(listener)
    self._logcat_thread.wait(
        event=event, timeout_sec=wait_for_message.timeout_sec)
    self._logcat_thread.remove_event_listener(listener)
    if got_message:
      logging.info('Message received: [%r]', message)
    else:
      logging.error('Failed to wait for message: [%r].', message)
      raise errors.WaitForMessageError()

  def _check_install(self, check_install: task_pb2.CheckInstall) -> None:
    """Checks that the given package is installed."""

    package = check_install.package_name
    logging.info('Checking if package is installed: [%r]', package)

    start_time = time.time()
    while time.time() - start_time < check_install.timeout_sec:
      if self._adb_controller.is_package_installed(package):
        logging.info('Done confirming that package is installed.')
        return
      time.sleep(0.1)

    logging.error('Package not found.')
    raise errors.CheckInstallError()
<file_sep># coding=utf-8
# Copyright 2021 DeepMind Technologies Limited.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

"""Wraps the AndroidEnv to expose an OpenAI Gym interface."""

from typing import Any, Dict, Tuple

from android_env.wrappers import base_wrapper
import dm_env
from dm_env import specs
import gym
from gym import spaces
import numpy as np


class GymInterfaceWrapper(base_wrapper.BaseWrapper, gym.Env):
  """AndroidEnv with OpenAI Gym interface."""

  def __init__(self, env: dm_env.Environment):

    base_wrapper.BaseWrapper.__init__(self, env)
    self.spec = None
    self.action_space = self._spec_to_space(self._env.action_spec())
    self.observation_space = self._spec_to_space(self._env.observation_spec())
    self.metadata = {'render.modes': ['rgb_array']}

  def _spec_to_space(self, spec: specs.Array) -> spaces.Space:
    """Converts dm_env specs to OpenAI Gym spaces."""

    if isinstance(spec, list):
      return spaces.Tuple([self._spec_to_space(s) for s in spec])

    if isinstance(spec, dict):
      return spaces.Dict(
          {name: self._spec_to_space(s) for name, s in spec.items()})

    if isinstance(spec, specs.DiscreteArray):
      return spaces.Box(
          shape=(),
          dtype=spec.dtype,
          low=0,
          high=spec.num_values-1)

    if isinstance(spec, specs.BoundedArray):
      return spaces.Box(
          shape=spec.shape,
          dtype=spec.dtype,
          low=spec.minimum,
          high=spec.maximum)

    if isinstance(spec, specs.Array):
      return spaces.Box(
          shape=spec.shape,
          dtype=spec.dtype,
          low=-np.inf,
          high=np.inf)

    raise ValueError('Unknown type for specs: {}'.format(spec))

  def render(self, mode='rgb_array'):
    """Renders the environment."""
    if mode == 'rgb_array':
      if self._latest_observation is None:
        return
      return self._latest_observation['pixels']
    else:
      raise ValueError('Only supported render mode is rgb_array.')

  def reset(self) -> np.ndarray:
    timestep = self._env.reset()
    return timestep.observation

  def step(self, action: Dict[str, int]) -> Tuple[Any, ...]:
    """Take a step in the base environment."""
    timestep = self._env.step(self._process_action(action))
    observation = timestep.observation
    reward = timestep.reward
    done = timestep.step_type == dm_env.StepType.LAST
    info = {'discount': timestep.discount}
    return observation, reward, done, info
<file_sep># AndroidEnv - Environment features

AndroidEnv is a complex environment that, while offering an almost endless range
of possibilites for RL research and investigation, poses multiple kinds of
challenges simultaneously. In this document we outline AndroidEnv's main
features that render it such a unique learning environment.

## Real-time environment

AndroidEnv is built on top of an emulated Android device, allowing the agent to
communicate with the emulator through touch actions. Android emulators are
created independently from our environment implementation and simulate real
Android devices in the most realistic manner. This simulation runs real-time,
independently of the agent, meaning that the simulaton will not wait for agent
input between frames. This aspect of the environment renders the RL setup
similar to a robotics problem, where the challenges of real-time interaction and
consequent noise in observations have to be overcome. Please note there is
currently no straightforward way to slow down the simulation either.

## Action space

<img align="right" src="images/classic_2048.gif" width="160" height="240">

Perhaps one of the most interesting features of AndroidEnv is its large and
complex action interface. The raw action space of the environment consists of a
tuple `(x,y) in [0,1]x[0,1]` determining the location of the action on the
Android screen, and a discrete value `ActionType in {LIFT, TOUCH, REPEAT}`
indicating whether the agent wants to touch the screen at this chosen location
or not. This action space is uniform across all tasks/apps.

**Gestures.** The complexity of the interface arises from the fact that
individual raw actions on their own do not neccessarily trigger a meaningful
change in the environment. Most Android applications are designed such that they
can be controlled/navigated through common touchscreen gestures such as pressing
buttons, swiping, scrolling, pinching, drag and drop etc. Each of these can be
thought of as particular sequences of raw actions: for example, *touching* the
screen at a particular location, then immediately *lifting* the imaginary finger
might be interpreted as a *press of a button*; while a sequence of *touches*
aligned in a vertical line might be interpreted as *scrolling*. We note that
AndroidEnv does not support multitouch actions at the moment, but it is a
possible feature to add.

It is important to point out that it is out of the environment's control to
determine how particular sequences of raw actions get interpreted by the Android
simulator - much like when humans interact with physical devices, a certain
gesture on the screen might be interpreted differently if it is performed at a
slightly different angle or speed.

Tap                                                      | Double Tap                                                             | Touch & Hold                                                             | Flick Left                                                             | Flick Right                                                              | Scroll (H)                                                                           | Scroll (V)                                                                       | Drag & Drop
-------------------------------------------------------- | ---------------------------------------------------------------------- | ------------------------------------------------------------------------ | ---------------------------------------------------------------------- | ------------------------------------------------------------------------ | ------------------------------------------------------------------------------------ | -------------------------------------------------------------------------------- | -----------
![Screenshot of 'tap'](images/gestures/1-Finger-Tap.gif) | ![Screenshot of 'double_tap'](images/gestures/1-Finger-Double-Tap.gif) | ![Screenshot of 'touch_hold'](images/gestures/1-Finger-Touch-&-Hold.gif) | ![Screenshot of 'flick_left'](images/gestures/1-Finger-Flick-Left.gif) | ![Screenshot of 'flick_right'](images/gestures/1-Finger-Flick-Right.gif) | ![Screenshot of 'horizontal_scroll'](images/gestures/1-Finger-Horizontal-Scroll.gif) | ![Screenshot of 'vertical_scroll'](images/gestures/1-Finger-Vertical-Scroll.gif) | ![Screenshot of 'move'](images/gestures/1-Finger-Move.gif)

**Wrappers.** It is possible to alter the raw action space of the environment by
applying [wrappers](#wrappers). For example one might discretize the action
space by splitting the screen up into a grid of a desired size; restrict the
ActionType to *touch* only; or fix certain gesture skills. We note here that
these wrappers, again, will not alter how the particular sequence of performed
raw actions gets interpreted by the Android simulator.

## Observation space

The observation space of AndroidEnv consists of three main components:
(`pixels`, `timedelta`, `orientation`), the most notable of these being
`pixels`. The original screen size will depend on the type of emulator used, but
given that it will correspond to real device screen sizes, this will usually be
quite large (of course, this can be scaled down, e.g. with wrappers). The
`timedelta` component captures the amount of time passed since the last
observation was fetched. The `orientation`, even though it does not affect the
layout of the RGB image in the observation, might carry relevant information for
the agent. For example, if there is text on the screen, it is important to know
how it is oriented. Again, a benefit of this observation space is that it is
uniform across all tasks. As mentioned above, observations often carry spatial
cues and are suggestive of the kind of actions/gestures that are meaningful to
perform in a given state.

## Task extras

On top of the default observations (`pixels`, `timedelta`, `orientation`), some
tasks might expose additional structured observations after each step. An
*extra* in AndroidEnv is any information that an app may send to aid the
understanding of the task. The type of information sent through this channel is
usually something difficult to obtain from raw pixels and may include meaningful
information such as:

*   The current board configuration (e.g. of a chess game or of a tetris game)
    in matrix or string form.
*   The position of the avatar in a map.
*   Events such as whether a button was pressed or whether a checkpoint was
    achieved.

Note that these are entirely optional and may not be available at all.

Analogously to `env.observation_spec()` and `env.action_spec()`, AndroidEnv
exposes `env.task_extras_spec()`, which will return a dictionary of
`dm_env.specs.Arrays`, capturing the specs of different extra information
provided by the given task.

To request extras from the environment, you can call `env.task_extras()` after
each `env.step()`, which will return a dictionary of all the extra observations
observed during the previous step (or an empty dict is there's none available).
For example:

```python
for _ in range(FLAGS.n_steps):

  action = agent.select_action(timestep.observation)
  timestep = env.step(action)
  logging.info('observation: %s', timestep.observation)
  logging.info('extra observations: %s', env.task_extras())
```

Please note however that the env might not return extras at every timestep, only
when something meaningful happened (e.g. only when a button was pressed, or when
the state of the board has changed).

When integrating your own APK as a new task for the environment, you can define
your own extras by following the instructions
[here](tasks_guide.md#log-messages-and-custom-apks).

## Wrappers

AndroidEnv's action- and observation spaces can be altered by applying suitable
wrappers. While custom wrappers can be built easily, we have provided a number
of useful wrappers that demonstrate their usage:

*   `discrete_action_wrapper`: Discretizes the action space into an `nxk` grid.
*   `flat_interface_wrapper`: Removes the dictionary structure from the
    observation and action specs.
*   `float_pixels_wrapper`: Projects the pixel RGB values from the integer range
    `[0, 255]` to the float range `[0, 1]`.
*   `image_rescale_wrapper`: Resizes the pixel observations by the selected
    ratio.
*   `gym_wrapper`: Changes the environment interface from
    [dm_env](https://github.com/deepmind/dm_env) to
    [OpenAI](https://gym.openai.com/) gym interface.
*   `last_action_wrapper`: Extends the observation with a one-hot encoded
    location of the previously taken action, in order to aid agents without
    built-in memory.

## Internal structure of AndroidEnv

The chart below gives an overview of the internal workings of the system,
illustrating how different classes interact with each other and what their
individual roles are. See the source code for more details.

![Components Chart](images/misc/components_chart.svg)
<file_sep># coding=utf-8
# Copyright 2021 DeepMind Technologies Limited.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

"""Acme DQN agent interacting with AndroidEnv."""

from absl import app
from absl import flags
from absl import logging
import acme
from acme import specs
from acme import wrappers as acme_wrappers
from acme.agents.tf import dqn
from acme.tf import networks
import android_env
from android_env import wrappers

# Simulator args
flags.DEFINE_string('avd_name', None, 'Name of AVD to use.')
flags.DEFINE_string('android_avd_home', '~/.android/avd', 'Path to AVD.')
flags.DEFINE_string('android_sdk_root', '~/Android/Sdk', 'Path to SDK.')
flags.DEFINE_string('emulator_path',
                    '~/Android/Sdk/emulator/emulator', 'Path to emulator.')
flags.DEFINE_string('adb_path',
                    '~/Android/Sdk/platform-tools/adb', 'Path to ADB.')

# Environment args
flags.DEFINE_string('task_path', None, 'Path to task textproto file.')

# Experiment args
flags.DEFINE_integer('num_episodes', 100, 'Number of episodes.')

FLAGS = flags.FLAGS


def apply_wrappers(env):
  """Applies a series of wrappers to the environment."""
  env = wrappers.DiscreteActionWrapper(env, action_grid=(10, 10))
  env = wrappers.ImageRescaleWrapper(env, zoom_factors=(0.25, 0.25))
  env = wrappers.FloatPixelsWrapper(env)
  env = acme_wrappers.SinglePrecisionWrapper(env)
  return env


def main(_):

  env = android_env.load(
      emulator_path=FLAGS.emulator_path,
      android_sdk_root=FLAGS.android_sdk_root,
      android_avd_home=FLAGS.android_avd_home,
      avd_name=FLAGS.avd_name,
      adb_path=FLAGS.adb_path,
      task_path=FLAGS.task_path,
      run_headless=False,
  )
  env = apply_wrappers(env)
  env_spec = specs.make_environment_spec(env)

  agent = dqn.DQN(
      environment_spec=env_spec,
      network=networks.DQNAtariNetwork(
          num_actions=env_spec.actions.num_values),
      batch_size=10,
      samples_per_insert=2,
      min_replay_size=10)

  loop = acme.EnvironmentLoopV2(env, agent)
  loop.run(num_episodes=FLAGS.num_episodes)

  env.close()


if __name__ == '__main__':
  logging.set_verbosity('info')
  logging.set_stderrthreshold('info')
  flags.mark_flags_as_required(['task_path', 'avd_name'])
  app.run(main)
<file_sep># AndroidEnv - Running the environment

In order to create an AndroidEnv instance you will need to provide two main
components: a [simulator](#the-simulator) and a [task](#the-task). In the
following sections you will learn how you can create them.

### The simulator

First, you will need to provide your Android virtual device (AVD) that the
environment (and through it, the agent) can communicate with. While this can also
be a physical device, in most cases you will need a virtual emulated device.
There are many ways to emulate an AVD - in our examples, we will use
[Android Studio](https://developer.android.com/studio) to create one.

1. In Android Studio, create a virtual device by following this step-by-step
[guide](emulator_guide.md).
2. Follow the steps below to attach the AVD to your environment.

### The task and examples with games and other apps

A `task` is a particular definition of an RL problem that the agent will be
interacting with. A `task` may include critical RL information, such as what the
rewards are, when the episodes are supposed to terminate, and what the reset
procedures are that the environment should perform upon episode termination
(e.g. start or relaunch an app, clear cache, etc.). This information is packaged
into a `Task()` proto message, which gets passed passed to AndroidEnv.

* For ready-made example tasks provided with AndroidEnv, check out
  the [Available tasks](example_tasks.md), featuring Vokram (with Markov Decision
  Process (MDP)), Pong, DroidFish (a chess clone), Blockinger (a tetris clone),
  and more.

* See the [Tasks guide](tasks_guide.md) for details on features and
  capabilities of tasks, as well as how to create custom ones.

### Create the environment

After setting up the simulator and creating a task, you may find the 
[`android_env.load()`](https://github.com/deepmind/android_env/blob/main/android_env/loader.py)
function handy for creating an environment instance by providing relevant
arguments, such as:

*   `task_path`: the path pointing to the `.textproto` file describing the
    desired task.
*   `avd_name`: the name of the AVD specified when your created it in Android
    Studio.
*   `android_avd_home` (Optional): the path to where the AVD is installed. 
    (default value: `~/.android/avd`).
*   `android_sdk_root` (Optional): the root directory of the Android SDK. 
    (default value: `~/Android/Sdk`).
*   `emulator_path` (Optional): the path to the emulator binary. (default:
    `~/Android/Sdk/emulator/emulator`).
*   `adb_path` (Optional): the path to the ADB
    ([Android Debug Bridge](https://developer.android.com/studio/command-line/adb)).
    (default value: `~/Android/Sdk/platform-tools/adb`).
 
For the AVD name and path, in Android Studio, go to **Tools** >
**AVD Manager**, right click on your virtual device, and select
**View Details**, where you will find the `avd_name` and its path.

For the Android SDK location, in Android Studio, go to **Preferences** >
**Appearance & Behavior** > **System Settings** > **Android SDKs** and note the
_Android SDK Location_. In the SDK folder, you will find `/emulator/emulator`
as well as the ADB path (`/platform-tools/adb`).

Your example configuration may look like this, depending on how you set up your
emulator:

```python
import android_env

env = android_env.load(
    avd_name='my_avd',
    android_avd_home='/Users/username/.android/avd',
    android_sdk_root='/Users/username/Library/Android/sdk',
    emulator_path='/Users/username/Library/Android/sdk/emulator/emulator',
    adb_path='/Users/username/Library/Android/sdk/platform-tools/adb',
    task_path='/Users/username/android_env/my_tasks/my_task.textproto',
)
```

## Example RL agent scripts

The `examples` directory contains a few simple example agent setups, such as:

*   [`run_random_agent.py`](https://github.com/deepmind/android_env/blob/main/examples/run_random_agent.py):
    Runs a simple loop performing randomly selected actions in the environment.
*   [`run_acme_agent.py`](https://github.com/deepmind/android_env/blob/main/examples/run_acme_agent.py):
    (Linux only) Runs a training loop with an
    [Acme](https://deepmind.com/research/publications/Acme)
    DQN agent, implemented in the popular DeepMind RL framework. This will
    require to install the [`acme`](https://github.com/deepmind/acme)
    dependency.
*   [`run_human_agent.py`](https://github.com/deepmind/android_env/blob/main/examples/run_human_agent.py):
    Creates a [`pygame`](https://www.pygame.org) instance that lets a human user
    interact with the environment and observe environment mechanics, such as
    rewards or task extras. You will need to install the [PyGame] dependency.

For instance, here is how you can run
[`run_random_agent.py`](https://github.com/8bitmp3/android_env/blob/main/examples/run_random_agent.py)
in a folder where you have your APK file, such as
[Apple Flinger](https://github.com/deepmind/android_env/blob/main/docs/example_tasks.md#apple-flinger)
from
[Example tasks](https://github.com/deepmind/android_env/blob/main/docs/example_tasks.md).
(The downloaded TAR file contains the APK file and `.textproto` definitions.)

```shell
python3 run_random_agent.py \
--avd_name='my_avd',
--android_avd_home=/Users/username/.android/avd \
--android_sdk_root=/Users/username/Library/Android/sdk \
--emulator_path=/Users/username/Library/Android/sdk/emulator/emulator \
--adb_path=/Users/username/Library/Android/sdk/platform-tools/adb \
--num_steps=1000 \
--task_path=/Users/username/<PATH-TO-APP-TASK-FILES>/apple_flinger_M_1_1.textproto
```
<file_sep># coding=utf-8
# Copyright 2021 DeepMind Technologies Limited.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

"""Utils for AndroidEnv."""

from typing import Sequence, Tuple


from android_env.proto import task_pb2
import numpy as np


def touch_position_to_pixel_position(
    touch_position: np.ndarray,
    width_height: Sequence[int],
) -> Tuple[int, int]:
  """Maps touch position in [0,1] to the corresponding pixel on the screen."""
  touch_pixels = (touch_position * width_height).astype(np.int32)
  cap_idx = lambda v, idx_len: min(v, idx_len - 1)
  return tuple(map(cap_idx, touch_pixels, width_height))


def transpose_pixels(frame: np.ndarray) -> np.ndarray:
  """Converts image from shape (H, W, C) to (W, H, C) and vice-versa."""
  return np.transpose(frame, axes=(1, 0, 2))


def orient_pixels(
    frame: np.ndarray,
    orientation: task_pb2.AdbCall.Rotate.Orientation) -> np.ndarray:
  """Rotates screen pixels according to the given orientation."""
  if orientation == task_pb2.AdbCall.Rotate.Orientation.LANDSCAPE_90:
    frame = np.rot90(frame, k=3, axes=(0, 1))
  elif orientation == task_pb2.AdbCall.Rotate.Orientation.PORTRAIT_180:
    frame = np.rot90(frame, k=2, axes=(0, 1))
  elif orientation == task_pb2.AdbCall.Rotate.Orientation.LANDSCAPE_270:
    frame = np.rot90(frame, k=1, axes=(0, 1))
  return frame
<file_sep># coding=utf-8
# Copyright 2021 DeepMind Technologies Limited.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

"""Tests for android_env.components.app_screen_checker."""

import re
from typing import Sequence

from absl.testing import absltest
from android_env.components import app_screen_checker


def flatten_tree(tree: app_screen_checker.DumpsysNode,
                 flat_tree: Sequence[str],
                 indent: int = 2):
  """Appends a list of strings to `flat_tree` from `tree`."""
  flat_tree.append(' ' * indent + tree.data)
  for c in tree.children:
    flatten_tree(c, flat_tree, indent + 2)


class AppScreenCheckerTest(absltest.TestCase):

  # Ensures that build_tree_from_dumpsys_output produces a node whose flat
  # representation matches our expectation from an arbitrary hierarchy.
  def test_build_tree_from_dumpsys_output(self):
    dumpsys_output = """
Queen Elizabeth II
  Charles
    William
      George
      Charlotte
      Louis
    Harry
      Archie
  Anne
    Peter
      Savannah
      Isla
    Zara
      Mia
      Lena
  Andrew
    Beatrice
    Eugenie
  Edward
    Louise
    James
"""
    tree = app_screen_checker.build_tree_from_dumpsys_output(dumpsys_output)
    flat_tree = []
    flatten_tree(tree, flat_tree, indent=2)
    self.assertEqual(flat_tree, [
        '  ___root___',
        '    Queen Elizabeth II',
        '      Charles',
        '        William',
        '          George',
        '          Charlotte',
        '          Louis',
        '        Harry',
        '          Archie',
        '      Anne',
        '        Peter',
        '          Savannah',
        '          Isla',
        '        Zara',
        '          Mia',
        '          Lena',
        '      Andrew',
        '        Beatrice',
        '        Eugenie',
        '      Edward',
        '        Louise',
        '        James',
    ])

  # Ensures that build_tree_from_dumpsys_output produces a node whose flat
  # representation matches our expectation from an arbitrary hierarchy.
  def test_build_forest_from_dumpsys_output(self):
    dumpsys_output = """
Tree1
  Branch1
    Leaf1
    Leaf2
  Branch2
    Leaf3
    Leaf4
    Leaf5
Tree2
  Branch3
    Leaf6
    Leaf7
  Branch4
    Leaf8
    Leaf9
    Leaf10
  Leaf11
"""
    tree = app_screen_checker.build_tree_from_dumpsys_output(dumpsys_output)
    flat_tree = []
    flatten_tree(tree, flat_tree, indent=2)
    self.assertEqual(flat_tree, [
        '  ___root___',
        '    Tree1',
        '      Branch1',
        '        Leaf1',
        '        Leaf2',
        '      Branch2',
        '        Leaf3',
        '        Leaf4',
        '        Leaf5',
        '    Tree2',
        '      Branch3',
        '        Leaf6',
        '        Leaf7',
        '      Branch4',
        '        Leaf8',
        '        Leaf9',
        '        Leaf10',
        '      Leaf11',
    ])

  def test_no_view_hierarchy_matches_path(self):
    dumpsys_output = """
TASK
  ACTIVITY
    Missing View Hierarchy
      A
        B
        C
      D
        E
          F
"""
    expected_path = ['^A$', 'B$']
    expected_view_hierarchy_path = [
        re.compile(regex) for regex in expected_path
    ]
    self.assertFalse(
        app_screen_checker.matches_path(dumpsys_output,
                                        expected_view_hierarchy_path))

  def test_matches_path(self):
    dumpsys_output = """
TASK
  ACTIVITY
    Some node we don't care
      Blah

    View Hierarchy
      Hirohito
        Akihito
          Naruhito
            Aiko
          Fumihito
            Mako
            Kako
            Hisahito
        Masahito
"""
    expected_path = ['^Hirohito$', 'Akihito$', 'Fumihito$', 'Kako$']
    expected_view_hierarchy_path = [
        re.compile(regex) for regex in expected_path
    ]
    self.assertTrue(
        app_screen_checker.matches_path(
            dumpsys_output, expected_view_hierarchy_path, max_levels=2))

    # Also check that the following path does not match anything in the tree.
    expected_path = ['^Hirohito$', 'Akihito$', 'Fumihito$', 'Kenji$']
    expected_view_hierarchy_path = [
        re.compile(regex) for regex in expected_path
    ]
    self.assertFalse(
        app_screen_checker.matches_path(dumpsys_output,
                                        expected_view_hierarchy_path))

  def test_matches_path_one_level_deep(self):
    dumpsys_output = """
TASK
  ACTIVITY
    Some node we don't care
      Blah

    Some intermediate node
      View Hierarchy
        Hirohito
          Akihito
            Naruhito
              Aiko
            Fumihito
              Mako
              Kako
              Hisahito
          Masahito
"""
    expected_path = ['^Hirohito$', 'Akihito$', 'Fumihito$', 'Kako$']
    expected_view_hierarchy_path = [
        re.compile(regex) for regex in expected_path
    ]
    self.assertTrue(
        app_screen_checker.matches_path(
            dumpsys_output, expected_view_hierarchy_path, max_levels=3))

    # Also check that the view hierarchy is not found when searching only grand
    # children of TASK.
    expected_path = ['^Hirohito$', 'Akihito$', 'Fumihito$', 'Kako$']
    expected_view_hierarchy_path = [
        re.compile(regex) for regex in expected_path
    ]
    self.assertFalse(
        app_screen_checker.matches_path(
            dumpsys_output, expected_view_hierarchy_path, max_levels=2))


if __name__ == '__main__':
  absltest.main()
<file_sep># coding=utf-8
# Copyright 2021 DeepMind Technologies Limited.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

"""Definitions of exceptions used by AndroidEnv."""


class ReadObservationError(Exception):
  """When the environment is unable to obtain an observation from a simulator."""


class ObservationDecodingError(ReadObservationError):
  """When the environment is unable to decode the observation from a simulator."""


class PipeTimedOutError(ReadObservationError):
  """When the environment waited for too long for part of an observation."""


class CoordinatorError(Exception):
  """Error raised by the Coordinator."""


class TooManyRestartsError(CoordinatorError):
  """The number of restarts has exceeded _MAX_RESTART_TRIES."""


class NotAllowedError(Exception):
  """When the player does something that outside of the task scope."""


class PlayerExitedActivityError(NotAllowedError):
  """When the player quits the current Android activity."""


class PlayerExitedViewHierarchyError(NotAllowedError):
  """When the player quits the current Android app screen."""


class AdbControllerError(Exception):
  """Errors that can be raised by ADBController."""


class AdbControllerShellInitError(AdbControllerError):
  """Raised when an error occurred when initializing ADB shell."""


class AdbControllerPexpectError(AdbControllerError):
  """Raise when a problem with pexpect communication occurs."""


class AdbControllerDeviceTimeoutError(AdbControllerError):
  """Raised when a device takes too long to respond."""


class AdbControllerConnectionError(AdbControllerError):
  """Error connecting to tcpip address."""


class SimulatorCrashError(Exception):
  """Raised when an AndroidSimulator crashed."""


class ConsoleConnectionError(Exception):
  """Raised when cannot connect to the emulator console."""


class SendActionError(Exception):
  """Raised when action couldn't be sent successfully."""


class StepCommandError(Exception):
  """Raised when setup step interpreter cannot process a command."""


class WaitForAppScreenError(StepCommandError):
  """Raised when the wait_for_app_screen success check is not met."""


class CheckInstallError(StepCommandError):
  """Raised when the check_install success check is not met."""


class WaitForMessageError(StepCommandError):
  """Raised when the wait_for_message success check is not met."""
<file_sep># coding=utf-8
# Copyright 2021 DeepMind Technologies Limited.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

"""Tests for android_env.components.utils."""

from absl.testing import absltest
from absl.testing import parameterized
from android_env.components import utils
from android_env.proto import task_pb2
import numpy as np


class UtilsTest(parameterized.TestCase):

  @parameterized.parameters(
      ([0.5, 0.5], [320, 480], (160, 240)),
      ([0.25, 0.75], [320, 480], (80, 360)),
      ([0.0, 0.0], [320, 480], (0, 0)),
      ([1.0, 1.0], [320, 480], (319, 479)),
      )
  def test_touch_position_to_pixel_position(
      self, touch_pos, width_height, pixel_pos):
    self.assertEqual(utils.touch_position_to_pixel_position(
        np.array(touch_pos), width_height), pixel_pos)

  def test_transpose_pixels(self):
    image = np.reshape(np.array(range(12)), (3, 2, 2))
    expected = [[[0, 1], [4, 5], [8, 9]], [[2, 3], [6, 7], [10, 11]]]
    self.assertEqual(utils.transpose_pixels(image).shape, (2, 3, 2))
    self.assertTrue((utils.transpose_pixels(image) == expected).all())

  def test_orient_pixels(self):
    image = np.reshape(np.array(range(12)), (3, 2, 2))

    expected_90 = [[[8, 9], [4, 5], [0, 1]], [[10, 11], [6, 7], [2, 3]]]
    rot_90 = task_pb2.AdbCall.Rotate.Orientation.LANDSCAPE_90
    rotated = utils.orient_pixels(image, rot_90)
    self.assertEqual(rotated.shape, (2, 3, 2))
    self.assertTrue((rotated == expected_90).all())

    expected_180 = [[[10, 11], [8, 9]], [[6, 7], [4, 5]], [[2, 3], [0, 1]]]
    rot_180 = task_pb2.AdbCall.Rotate.Orientation.PORTRAIT_180
    rotated = utils.orient_pixels(image, rot_180)
    self.assertEqual(rotated.shape, (3, 2, 2))
    self.assertTrue((rotated == expected_180).all())

    expected_270 = [[[2, 3], [6, 7], [10, 11]], [[0, 1], [4, 5], [8, 9]]]
    rot_270 = task_pb2.AdbCall.Rotate.Orientation.LANDSCAPE_270
    rotated = utils.orient_pixels(image, rot_270)
    self.assertEqual(rotated.shape, (2, 3, 2))
    self.assertTrue((rotated == expected_270).all())

    rot_0 = task_pb2.AdbCall.Rotate.Orientation.PORTRAIT_0
    rotated = utils.orient_pixels(image, rot_0)
    self.assertEqual(rotated.shape, (3, 2, 2))
    self.assertTrue((rotated == image).all())

if __name__ == '__main__':
  absltest.main()
<file_sep># coding=utf-8
# Copyright 2021 DeepMind Technologies Limited.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

"""Tests for android_env.components.base_simulator."""

from typing import Optional, List

from absl.testing import absltest
from android_env.components import action_type
from android_env.components import adb_controller
from android_env.components import base_simulator
import mock
import numpy as np


class FakeSimulator(base_simulator.BaseSimulator):
  """A simulator that spits injected data."""

  def __init__(self, injected_adb_controller, **kwargs):
    self._injected_adb_controller = injected_adb_controller
    super().__init__(**kwargs)

  def adb_device_name(self) -> str:
    return 'FakeSimulator'

  def create_adb_controller(self) -> adb_controller.AdbController:
    return self._injected_adb_controller

  def send_action(self, action) -> None:
    pass

  def _restart_impl(self) -> None:
    pass

  def _launch_impl(self) -> None:
    pass

  def _get_observation(self) -> Optional[List[np.ndarray]]:
    return [np.ones(shape=(640, 480, 3)), self._timestamp]

  def set_timestamp(self, timestamp: int):
    self._timestamp = timestamp


class BaseSimulatorTest(absltest.TestCase):

  def setUp(self):
    super().setUp()
    self._adb_controller = self.enter_context(
        mock.patch.object(adb_controller, 'AdbController', autospec=True))
    tmp_dir = absltest.get_default_test_tmpdir()
    self._simulator = FakeSimulator(
        adb_path='/my/adb',
        adb_server_port=5037,
        tmp_dir=tmp_dir,
        prompt_regex='awesome>',
        injected_adb_controller=self._adb_controller)

  def test_adb_server_init(self):
    # The simulator init should start the adb server.
    self._adb_controller.init_server.assert_called_once()

  def test_device_name(self):
    self.assertNotEmpty(self._simulator.adb_device_name())

  def test_launch(self):
    # Before a successful launch(), screen_dimensions() should raise.
    self.assertRaises(AssertionError, self._simulator.screen_dimensions)
    # The simulator should launch and not crash.
    self._adb_controller.get_screen_dimensions.return_value = (640, 480)
    self._simulator.launch()
    # After a successful launch(), screen_dimensions() should return something.
    np.testing.assert_equal(self._simulator.screen_dimensions(), [640, 480])

  def test_launch_close(self):
    # The simulator should launch and not crash.
    self._adb_controller.get_screen_dimensions.return_value = (640, 480)
    self._simulator.launch()
    # Closing the simulator should also not crash.
    self._simulator.close()

  def test_get_observation(self):
    # The simulator should launch and not crash.
    self._adb_controller.get_screen_dimensions.return_value = (640, 480)
    self._simulator.launch()

    self._simulator.set_timestamp(123456)
    observation = self._simulator.get_observation()
    # Even though the FakeSimulator returns a 640x480x3 image, BaseSimulator
    # should add an extra layer for the last action.
    np.testing.assert_equal(observation['pixels'].shape, [640, 480, 3])
    # Because there was only a single step, the timestamp delta should be just
    # the timestamp returned by FakeSimulator.
    self.assertEqual(observation['timedelta'], 123456)

    # The orientation format should be a 4-dimension one-hot.
    np.testing.assert_equal(observation['orientation'].shape, [4])
    # Initially the orientation should not be set so we expect [0, 0, 0, 0].
    self.assertEqual(np.sum(observation['orientation']), 0)

    # After updating the device orientation, we do expect to get an actual
    # one-hot.
    self._adb_controller.get_orientation.return_value = '3'
    self._simulator.update_device_orientation()

    self._simulator.set_timestamp(123459)
    updated_observation = self._simulator.get_observation()
    np.testing.assert_equal(updated_observation['orientation'], [0, 0, 0, 1])
    # After setting the initial timestamp, we expect the timestamp delta to be
    # the difference between the current timestamp and the initial timestamp:
    # 123459 - 123456 = 3
    self.assertEqual(updated_observation['timedelta'], 3)

  def test_prepare_action(self):
    # The simulator should launch and not crash.
    self._adb_controller.get_screen_dimensions.return_value = (480, 640)
    self._simulator.launch()

    self._simulator.set_timestamp(123456)
    self.assertRaises(
        ValueError, self._simulator._prepare_action, {
            'action_type': np.array(action_type.ActionType.REPEAT),
            'touch_position': [0.0, 0.0]
        })
    self.assertEqual(
        self._simulator._prepare_action({
            'action_type': np.array(action_type.ActionType.LIFT),
            'touch_position': [0.0, 0.0]
        }), (0, 0, False))
    self.assertEqual(
        self._simulator._prepare_action({
            'action_type': np.array(action_type.ActionType.TOUCH),
            'touch_position': [0.0, 0.0]
        }), (0, 0, True))
    self.assertEqual(
        self._simulator._prepare_action({
            'action_type': np.array(action_type.ActionType.TOUCH),
            'touch_position': [0.5, 0.2]
        }), (320, 96, True))
    self.assertEqual(
        self._simulator._prepare_action({
            'action_type': np.array(action_type.ActionType.TOUCH),
            'touch_position': [1.0, 1.0]
        }), (639, 479, True))


if __name__ == '__main__':
  absltest.main()
<file_sep># AndroidEnv - Emulator Setup Guide

In this document we provide a step-by-step guide for creating a virtual Android
device with Android Studio. After creating an AVD
([Android Virtual Device](https://developer.android.com/studio/run/managing-avds))
you will be able to connect it to an AndroidEnv instance and you're ready to go.

To get started, you will need to download
[Android Studio](https://developer.android.com/studio) - an IDE widely used by
Android developers.

## Install an SDK Platform Package

Android Studio comes with the Android Software Development Toolkit (SDK) which,
among others, allows you to install different versions of Android. Click
on **Tools** > **SDK Manager** and select the SDK version that you would like to
use.

![Screenshot of 'android_studio_2'](images/android_studio/android_studio_2.png)

We recommend that you set the `Android SDK Location` to be in your home
directory (for example, on Linux the default one is `~/Android/Sdk`, while on
macOS - `~/Library/Android/sdk`). You can always find the SDK location in
Android Studio under **Preferences** > **Appearance & Behavior** >
**System Settings** > **Android SDKs** > _Android SDK Location_.

If you set the the custom `Android SDK Location`, make note of it - you will 
need it for connecting AndroidEnv to your AVD.

![Screenshot of 'android_studio_0'](images/android_studio/android_studio_0.png)

## Create an AVD

Now it is time to create a virtual device (AVD). Go to **Tools** >
**AVD Manager**.

![Screenshot of 'android_studio_1'](images/android_studio/android_studio_1.png)

In the pop-up window you will find an option to **Create Virtual Device**.

![Screenshot of 'android_studio_3'](images/android_studio/android_studio_3.png)

Configure the virtual device. You can select the model or choose from more
advanced settings (refer to the
[Android docs](https://developer.android.com/studio/run/managing-avds)
for step-by-step instructions).

![Screenshot of 'android_studio_4'](images/android_studio/android_studio_4.png)

Name your AVD and take note of this value. It will be neccessary for connecting
AndroidEnv to this virtual device.

![Screenshot of 'android_studio_6'](images/android_studio/android_studio_6.png)

Once you are done, you will see the new AVD show up in the **AVD Manager**.
Click on **View details** to inspect some of its properties.

![Screenshot of 'android_studio_8'](images/android_studio/android_studio_8.png)

Take note of the `AVD Path`. This value will be neccessary for connecting
AndroidEnv to this device. We recommend that you set this to be your home
directory (for instance, on Linux or macOS it may be  `~/.android/avd`).

![Screenshot of 'android_studio_9'](images/android_studio/android_studio_9.png)

## Ready to use

With SDK and AVD both set up, you are now ready to use this emulated device with
AndroidEnv. Don't forget to take note of the following three values: your AVD
name, the AVD path, and the SDK path. For example, on Linux they may be:

```
--avd_name=my_avd
--avd_package_path=~/.android/avd
--android_sdk_root=~/Android/Sdk
```

Next, once you have set up the AVD, follow the
[Task steps](instructions.md#the-task) in the
[Running the environment guide](instructions.md) to finish setting up
AndroidEnv.

However, if you want to just interact with the newly created device, click on
the run button next to your AVD in the **AVD Manager** in Android Studio (this
step is optional).

![Screenshot of 'android_studio_7'](images/android_studio/android_studio_7.png)

You will see an emulator window pop up. You can interact with it by clicking on
the screen.

![Screenshot of 'android_studio_10'](images/android_studio/android_studio_10.png)

There are many other features in Android Studio that let you customize your
device. For example, you can create custom images with pre-installed
applications or configured settings.
<file_sep># coding=utf-8
# Copyright 2021 DeepMind Technologies Limited.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

"""Tests for android_env.components.thread_function."""

from absl.testing import absltest
from android_env.components import thread_function


class ThreadFunctionTest(absltest.TestCase):

  def test_hello_world(self):

    class HelloFunction(thread_function.ThreadFunction):

      def main(self):
        v = self._read_value()
        if v:
          self._write_value('hello')
        else:
          self._write_value('world')

    hello_fn = HelloFunction(block_input=True, block_output=True, name='Hello')
    hello_fn.write(True, block=True)
    self.assertEqual(hello_fn.read(block=True), 'hello')
    hello_fn.write(False, block=True)
    self.assertEqual(hello_fn.read(block=True), 'world')
    hello_fn.kill()  # Not strictly necessary, but it's good hygiene.

  def test_class_with_state(self):
    """Derived class with internal state to change the thread output."""

    class HaveState(thread_function.ThreadFunction):

      def __init__(self, value, *args, **kwargs):
        self._value = value
        super().__init__(*args, **kwargs)

      def main(self):
        v = self._read_value()
        if v:
          self._write_value(v + self._value)

    have_state_fn = HaveState(
        value=123, block_input=True, block_output=True, name='HaveState')
    have_state_fn.write(3, block=True)
    self.assertEqual(have_state_fn.read(block=True), 126)
    have_state_fn.write(9, block=True)
    self.assertEqual(have_state_fn.read(block=True), 132)
    have_state_fn.kill()


if __name__ == '__main__':
  absltest.main()
<file_sep># coding=utf-8
# Copyright 2021 DeepMind Technologies Limited.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

"""Tests for android_env.components.adb_controller."""

import os
import subprocess
import time

from absl.testing import absltest
from absl.testing import parameterized
from android_env.components import adb_controller
from android_env.components import errors
from android_env.proto import task_pb2
import mock

# Timeout to be used by default in tests below. Set to a small value to avoid
# hanging on a failed test.
_TIMEOUT = 2


class AdbControllerTest(parameterized.TestCase):

  def setUp(self):
    super().setUp()
    self._mock_execute_command = self.enter_context(
        mock.patch.object(
            adb_controller.AdbController, '_execute_command', autospec=True))
    self._adb_controller = adb_controller.AdbController(
        adb_path='my_adb',
        device_name='awesome_device',
        adb_server_port=9999,
        shell_prompt='l33t>')

  @mock.patch.object(
      adb_controller.AdbController, '_wait_for_device', autospec=True)
  def test_set_touch_indicators(self, mock_wait_for_device):
    self._adb_controller.set_touch_indicators(
        show_touches=True, pointer_location=False, timeout=_TIMEOUT)
    mock_wait_for_device.assert_called_once()
    self._mock_execute_command.assert_has_calls([
        mock.call(self._adb_controller,
                  ['shell', 'settings', 'put', 'system', 'show_touches', '1'],
                  _TIMEOUT),
        mock.call(
            self._adb_controller,
            ['shell', 'settings', 'put', 'system', 'pointer_location', '0'],
            _TIMEOUT)
    ])

  def test_force_stop(self):
    self._adb_controller.force_stop('com.amazing.package', timeout=_TIMEOUT)
    self._mock_execute_command.assert_called_once_with(
        self._adb_controller,
        ['shell', 'am', 'force-stop', 'com.amazing.package'], _TIMEOUT)

  def test_clear_cache(self):
    self._adb_controller.clear_cache('com.amazing.package', timeout=_TIMEOUT)
    self._mock_execute_command.assert_called_with(
        self._adb_controller, ['shell', 'pm', 'clear', 'com.amazing.package'],
        _TIMEOUT)

  def test_grant_permissions(self):
    self._adb_controller.grant_permissions(
        'com.amazing.package', ['hey.READ', 'ho.WRITE'], timeout=_TIMEOUT)
    self._mock_execute_command.assert_has_calls([
        mock.call(self._adb_controller,
                  ['shell', 'pm', 'grant', 'com.amazing.package', 'hey.READ'],
                  _TIMEOUT),
        mock.call(self._adb_controller,
                  ['shell', 'pm', 'grant', 'com.amazing.package', 'ho.WRITE'],
                  _TIMEOUT),
    ])

  def test_start_activity(self):
    self._adb_controller.start_activity(
        'hello.world/hello.world.MainActivity', [], timeout=_TIMEOUT)
    self._adb_controller.start_activity(
        full_activity='hello.world/hello.world.MainActivity',
        extra_args=['Planet 1', 'Planet 2'],
        timeout=_TIMEOUT)
    self._mock_execute_command.assert_has_calls([
        mock.call(self._adb_controller, [
            'shell', 'am', 'start', '-S', '-n',
            'hello.world/hello.world.MainActivity'
        ], _TIMEOUT),
        mock.call(self._adb_controller, [
            'shell', 'am', 'start', '-S', '-n',
            'hello.world/hello.world.MainActivity', 'Planet 1', 'Planet 2'
        ], _TIMEOUT)
    ])

  def test_start_intent(self):
    self._adb_controller.start_intent(
        action='action',
        data_uri='data',
        package_name='my.package',
        timeout=_TIMEOUT)

    self._mock_execute_command.assert_called_once_with(
        self._adb_controller,
        ['shell', 'am', 'start', '-a', 'action', '-d', 'data', 'my.package'],
        _TIMEOUT)

  def test_broadcast(self):
    self._adb_controller.broadcast(
        'hello.world/hello.world.BroadcastReceiver',
        'android.intent.action.TEST', [],
        timeout=_TIMEOUT)
    self._adb_controller.broadcast(
        receiver='hello.world/hello.world.BroadcastReceiver',
        action='android.intent.action.TEST',
        extra_args=['--es', 'KEY', 'VALUE'],
        timeout=_TIMEOUT)
    self._mock_execute_command.assert_has_calls([
        mock.call(self._adb_controller, [
            'shell', 'am', 'broadcast', '-n',
            'hello.world/hello.world.BroadcastReceiver', '-a',
            'android.intent.action.TEST'
        ], _TIMEOUT),
        mock.call(self._adb_controller, [
            'shell', 'am', 'broadcast', '-n',
            'hello.world/hello.world.BroadcastReceiver', '-a',
            'android.intent.action.TEST', '--es', 'KEY', 'VALUE'
        ], _TIMEOUT)
    ])

  def test_setprop(self):
    self._adb_controller.setprop('myprop', 'true', timeout=_TIMEOUT)
    self._adb_controller.setprop('myotherprop', 'false', timeout=_TIMEOUT)
    self._mock_execute_command.assert_has_calls([
        mock.call(self._adb_controller, ['shell', 'setprop', 'myprop', 'true'],
                  _TIMEOUT),
        mock.call(self._adb_controller,
                  ['shell', 'setprop', 'myotherprop', 'false'], _TIMEOUT),
    ])

  def test_rotate_device(self):
    self._adb_controller.rotate_device(
        task_pb2.AdbCall.Rotate.Orientation.LANDSCAPE_90, timeout=_TIMEOUT)
    self._adb_controller.rotate_device(
        task_pb2.AdbCall.Rotate.Orientation.PORTRAIT_0, timeout=_TIMEOUT)
    self._adb_controller.rotate_device(
        task_pb2.AdbCall.Rotate.Orientation.LANDSCAPE_270, timeout=_TIMEOUT)
    self._adb_controller.rotate_device(
        task_pb2.AdbCall.Rotate.Orientation.PORTRAIT_180, timeout=_TIMEOUT)
    self._mock_execute_command.assert_has_calls([
        mock.call(
            self._adb_controller,
            args=['shell', 'settings', 'put', 'system', 'user_rotation', '1'],
            timeout=_TIMEOUT),
        mock.call(
            self._adb_controller,
            args=['shell', 'settings', 'put', 'system', 'user_rotation', '0'],
            timeout=_TIMEOUT),
        mock.call(
            self._adb_controller,
            args=['shell', 'settings', 'put', 'system', 'user_rotation', '3'],
            timeout=_TIMEOUT),
        mock.call(
            self._adb_controller,
            args=['shell', 'settings', 'put', 'system', 'user_rotation', '2'],
            timeout=_TIMEOUT)
    ])

  @mock.patch.object(
      adb_controller.AdbController, '_wait_for_device', autospec=True)
  def test_get_screen_dimensions_failed_wait(self, mock_wait_for_device):

    mock_wait_for_device.side_effect = errors.AdbControllerDeviceTimeoutError(
        'Time is up.')
    self.assertRaises(errors.AdbControllerDeviceTimeoutError,
                      self._adb_controller.get_screen_dimensions)
    mock_wait_for_device.assert_called_once()

  @mock.patch.object(
      adb_controller.AdbController, '_wait_for_device', autospec=True)
  def test_get_screen_dimensions_success(self, mock_wait_for_device):
    self._mock_execute_command.return_value = b'Physical size: 1280x800'
    screen_dimensions = self._adb_controller.get_screen_dimensions(
        timeout=_TIMEOUT)
    mock_wait_for_device.assert_called()
    self._mock_execute_command.assert_called_with(self._adb_controller,
                                                  ['shell', 'wm', 'size'],
                                                  _TIMEOUT)
    self.assertEqual(screen_dimensions, (800, 1280))

  def test_activity_dumpsys(self):
    package_name = 'com.world.hello'
    self._mock_execute_command.return_value = b'My awesome dumpsys output!!!'
    activity_dumpsys = self._adb_controller.get_activity_dumpsys(
        package_name, timeout=_TIMEOUT)
    self._mock_execute_command.assert_called_once_with(
        self._adb_controller,
        ['shell', 'dumpsys', 'activity', package_name, package_name], _TIMEOUT)
    # Compare activity_dumpsys to what we want. Notice that we expect a UTF-8
    # string, NOT bytes.
    self.assertEqual(activity_dumpsys, 'My awesome dumpsys output!!!')

  def test_input_tap(self):
    self._mock_execute_command.return_value = b''
    self._adb_controller.input_tap(123, 456, timeout=_TIMEOUT)
    self._mock_execute_command.assert_called_once_with(
        self._adb_controller, ['shell', 'input', 'tap', '123', '456'], _TIMEOUT)

  def test_input_text(self):
    self._mock_execute_command.return_value = b''
    self._adb_controller.input_text('my_text', timeout=_TIMEOUT)
    self._mock_execute_command.assert_called_once_with(
        self._adb_controller, ['shell', 'input', 'text', 'my_text'], _TIMEOUT)

  def test_input_key(self):
    self._mock_execute_command.return_value = b''
    self._adb_controller.input_key('KEYCODE_HOME', timeout=_TIMEOUT)
    self._adb_controller.input_key('KEYCODE_BACK', timeout=_TIMEOUT)
    self._adb_controller.input_key('KEYCODE_ENTER', timeout=_TIMEOUT)
    self._mock_execute_command.assert_has_calls([
        mock.call(self._adb_controller,
                  ['shell', 'input', 'keyevent', 'KEYCODE_HOME'], _TIMEOUT),
        mock.call(self._adb_controller,
                  ['shell', 'input', 'keyevent', 'KEYCODE_BACK'], _TIMEOUT),
        mock.call(self._adb_controller,
                  ['shell', 'input', 'keyevent', 'KEYCODE_ENTER'], _TIMEOUT),
    ])

    # A key code outside of the accepted codes should raise an exception.
    self.assertRaises(AssertionError, self._adb_controller.input_key,
                      'KEYCODE_0')

  def test_install_apk(self):
    self._mock_execute_command.return_value = b''
    # Passing an invalid path should raise an exception.
    self.assertRaises(AssertionError, self._adb_controller.install_apk, '')

    local_apk_path = os.path.join(absltest.get_default_test_tmpdir(),
                                  'my_app.apk')
    with open(local_apk_path, 'wb') as f:
      f.write(b'blah. whatever')
    self._adb_controller.install_apk(local_apk_path, timeout=_TIMEOUT)
    self._mock_execute_command.assert_has_calls([
        mock.call(self._adb_controller,
                  ['install', '-r', '-t', '-g', local_apk_path], _TIMEOUT),
    ])

  def test_start_accessibility_service(self):
    self._mock_execute_command.return_value = b''
    self._adb_controller.start_accessibility_service(
        'my.service', timeout=_TIMEOUT)
    self._mock_execute_command.assert_has_calls([
        mock.call(self._adb_controller, [
            'shell', 'settings', 'put', 'secure',
            'enabled_accessibility_services', 'my.service'
        ], _TIMEOUT)
    ])

  @mock.patch.object(
      adb_controller.AdbController, '_fetch_current_task_id', autospec=True)
  def test_start_screen_pinning_task_not_found(self,
                                               mock_fetch_current_task_id):
    self._mock_execute_command.return_value = b''
    mock_fetch_current_task_id.return_value = -1
    self._adb_controller.start_screen_pinning(
        'my.app.CoolActivity', timeout=_TIMEOUT)
    self._mock_execute_command.assert_not_called()

  @mock.patch.object(
      adb_controller.AdbController, '_fetch_current_task_id', autospec=True)
  def test_start_screen_pinning(self, mock_fetch_current_task_id):
    self._mock_execute_command.return_value = b''
    mock_fetch_current_task_id.return_value = 123
    self._adb_controller.start_screen_pinning(
        'my.app.CoolActivity', timeout=_TIMEOUT)
    self._mock_execute_command.assert_has_calls([
        mock.call(self._adb_controller, ['shell', 'am', 'task', 'lock', '123'],
                  _TIMEOUT)
    ])

  def test_fetch_current_task_id(self):
    full_activity_name = (
        'com.google.example.games.nostalgicracer/'
        'com.google.example.games.nostalgicracer.MainActivity')
    bad_task = (
        '  taskId=8: '
        'com.google.android.apps.maps/com.google.android.maps.MapsActivity '
        'bounds=[0,0][480,320] userId=0 visible=true '
        'topActivity=ComponentInfo{%s}' % full_activity_name)
    good_task_not_visible = (
        'taskId=49: %s bounds=[0,0][320,480] userId=0 visible=false more_stuff'
        % full_activity_name)
    good_task = (
        'taskId=50: %s bounds=[0,0][320,480] userId=0 visible=true more_stuff' %
        full_activity_name)
    stack_list = '\n'.join([bad_task, good_task_not_visible,
                            good_task]).encode('utf-8')
    self._mock_execute_command.return_value = stack_list

    self.assertEqual(
        50,
        self._adb_controller._fetch_current_task_id(
            full_activity_name, timeout=_TIMEOUT))

  def test_check_install_not_installed(self):
    self._mock_execute_command.return_value = b"""
package:foo
package:bar
package:baz
"""
    self.assertFalse(
        self._adb_controller.is_package_installed('faz', timeout=_TIMEOUT))

  def test_check_install_installed(self):
    self._mock_execute_command.return_value = b"""
package:foo
package:bar
package:baz
"""
    self.assertTrue(
        self._adb_controller.is_package_installed('baz', timeout=_TIMEOUT))

  @parameterized.parameters(
      (True, True, 'null*'),
      (True, False, 'immersive.status=*'),
      (False, True, 'immersive.navigation=*'),
      (False, False, 'immersive.full=*'),
      (None, None, 'immersive.full=*'),  # Defaults to hiding both.
  )
  def test_set_bar_visibility(self, navigation, status, expected):
    expected_output = b'Message.'
    self._mock_execute_command.return_value = expected_output

    self.assertEqual(
        expected_output,
        self._adb_controller.set_bar_visibility(
            navigation=navigation, status=status, timeout=_TIMEOUT))
    self._mock_execute_command.assert_has_calls([
        mock.call(
            self._adb_controller,
            ['shell', 'settings', 'put', 'global', 'policy_control', expected],
            _TIMEOUT),
    ])

  @mock.patch.object(time, 'sleep', autospec=True)
  def test_init_server(self, mock_sleep):
    self._adb_controller.init_server(timeout=_TIMEOUT)
    self._mock_execute_command.assert_called_once_with(self._adb_controller,
                                                       ['devices'], _TIMEOUT)
    mock_sleep.assert_called_once()


class AdbControllerInitTest(absltest.TestCase):

  def test_deletes_problem_env_vars(self):
    os.environ['ANDROID_HOME'] = '/usr/local/Android/Sdk'
    os.environ['ANDROID_ADB_SERVER_PORT'] = '1337'
    adb_controller.AdbController(
        adb_path='my_adb',
        device_name='awesome_device',
        adb_server_port=9999,
        shell_prompt='l33t>',
        default_timeout=_TIMEOUT)
    self.assertNotIn('ANDROID_HOME', os.environ)
    self.assertNotIn('ANDROID_ADB_SERVER_PORT', os.environ)


if __name__ == '__main__':
  absltest.main()
<file_sep># coding=utf-8
# Copyright 2021 DeepMind Technologies Limited.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

"""A class that manages an Android Emulator."""

from typing import Any, Dict, Optional, List

from absl import logging
from android_env.components import base_simulator
from android_env.components import emulator_console
from android_env.components import emulator_launcher
from android_env.components import errors
from android_env.proto import emulator_controller_pb2
from android_env.proto import emulator_controller_pb2_grpc

import grpc
import numpy as np
import portpicker


def is_existing_emulator_provided(launcher_args: Dict[str, Any]) -> bool:
  """Returns true if all necessary args were provided."""
  return bool(launcher_args.get('adb_port') and
              launcher_args.get('emulator_console_port') and
              launcher_args.get('grpc_port'))


class EmulatorSimulator(base_simulator.BaseSimulator):
  """Controls an Android Emulator."""

  def __init__(self,
               emulator_launcher_args: Dict[str, Any],
               emulator_console_args: Dict[str, Any],
               **kwargs):

    # If adb_port and console_port are both already provided,
    # we assume the emulator already exists and there's no need to launch.
    if is_existing_emulator_provided(emulator_launcher_args):
      self._existing_emulator_provided = True
      self._adb_port = emulator_launcher_args['adb_port']
      self._console_port = emulator_launcher_args['emulator_console_port']
      self._grpc_port = emulator_launcher_args['grpc_port']
      logging.info('Connecting to existing emulator "%r"', self._adb_port)
    else:
      self._existing_emulator_provided = False
      self._adb_port = portpicker.pick_unused_port()
      self._console_port = portpicker.pick_unused_port()
      self._grpc_port = emulator_launcher_args.get(
          'grpc_port', portpicker.pick_unused_port())

    super().__init__(**kwargs)

    self._emulator_stub = None
    self._image_format = None

    # Create EmulatorLauncher.
    emulator_launcher_args.update({
        'adb_port': self._adb_port,
        'adb_server_port': self._adb_server_port,
        'emulator_console_port': self._console_port,
        'local_tmp_dir': self._local_tmp_dir,
        'kvm_device': self._kvm_device,
        'grpc_port': self._grpc_port,
    })
    logging.info('emulator_launcher_args: %r', emulator_launcher_args)
    if not self._existing_emulator_provided:
      self._launcher = emulator_launcher.EmulatorLauncher(
          **emulator_launcher_args)

   # Prepare EmulatorConsole.
    emulator_console_args.update({
        'console_port': self._console_port,
        'tmp_dir': self._local_tmp_dir,
    })
    logging.info('emulator_console_args: %r', emulator_console_args)
    self._emulator_console_args = emulator_console_args
    self._console = None

  def adb_device_name(self) -> str:
    return 'emulator-%s' % (self._adb_port - 1)

  def _start_console(self) -> None:
    self._console = emulator_console.EmulatorConsole(
        **self._emulator_console_args)

  def _restart_impl(self) -> None:
    if self._console is not None:
      self._console.close()
    if not self._existing_emulator_provided:
      self._launcher.restart()
    if self._grpc_port < 0:
      self._start_console()

  def _launch_impl(self) -> None:
    try:
      if not self._existing_emulator_provided:
        self._launcher.launch()
      if self._grpc_port < 0:
        self._start_console()
    except errors.SimulatorCrashError:
      # If the simulator crashes on the initial launch, we try to restart once.
      self.restart()

  def _post_launch_setup(self):
    super()._post_launch_setup()
    if self._grpc_port >= 0:
      self._emulator_stub = self._create_emulator_stub()
      self._image_format = emulator_controller_pb2.ImageFormat(
          format=emulator_controller_pb2.ImageFormat.ImgFormat.RGB888,
          height=self._screen_dimensions[0],
          width=self._screen_dimensions[1],
      )

  def _create_emulator_stub(self, use_async: bool = False):
    """Returns a stub to the EmulatorController service."""
    port = f'localhost:{self._grpc_port}'
    options = [('grpc.max_send_message_length', -1),
               ('grpc.max_receive_message_length', -1)]
    if use_async:
      self._channel = grpc.aio.insecure_channel(port, options=options)
    else:
      self._channel = grpc.insecure_channel(port, options=options)
    logging.info('Added gRPC channel for the Emulator on port %s', port)
    return emulator_controller_pb2_grpc.EmulatorControllerStub(self._channel)

  def send_action(self, action: Dict[str, np.ndarray]) -> None:
    """Sends a touch event to the emulator."""
    if self._grpc_port < 0:
      # Use the Emulator Console
      assert self._console, 'Console has not been initialized yet.'
      action = self._prepare_action(action)
      self._console.send_mouse_action(*action)
    else:
      # Use gRPC connection
      assert self._emulator_stub, 'Emulator stub has not been initialized yet.'
      x, y, down = self._prepare_action(action)
      self._emulator_stub.sendTouch(
          emulator_controller_pb2.TouchEvent(touches=[
              emulator_controller_pb2.Touch(x=x, y=y, pressure=int(down))]))

  def _get_observation(self) -> Optional[List[np.ndarray]]:
    """Fetches the latest observation from the emulator."""
    if self._grpc_port < 0:
      assert self._console, 'Console has not been initialized yet.'
      return self._console.fetch_screenshot()
    else:
      assert self._emulator_stub, 'Emulator stub has not been initialized yet.'
      assert self._image_format, 'ImageFormat has not been initialized yet.'
      image_proto = self._emulator_stub.getScreenshot(self._image_format)
      h, w = image_proto.format.height, image_proto.format.width
      image = np.frombuffer(image_proto.image, dtype='uint8', count=h*w*3)
      image.shape = (h, w, 3)
      return [image, np.int64(image_proto.timestampUs)]

  def close(self):
    if self._console is not None:
      self._console.close()
    if hasattr(self, '_channel'):
      self._channel.close()
    if hasattr(self, '_emulator_stub'):
      del self._emulator_stub
    if hasattr(self, '_launcher') and not self._existing_emulator_provided:
      self._launcher.close()
    super().close()
<file_sep># coding=utf-8
# Copyright 2021 DeepMind Technologies Limited.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

"""Loads an interactive session where a human acts on behalf of an agent."""

import time
from typing import Any, Dict

from absl import app
from absl import flags
from absl import logging
import android_env
from android_env.components import action_type
from android_env.components import utils
from android_env.proto import task_pb2
import dm_env
import numpy as np
import pygame

# Simulator args.
flags.DEFINE_string('avd_name', None, 'Name of AVD to use.')
flags.DEFINE_string('android_avd_home', '~/.android/avd', 'Path to AVD.')
flags.DEFINE_string('android_sdk_root', '~/Android/Sdk', 'Path to SDK.')
flags.DEFINE_string('emulator_path',
                    '~/Android/Sdk/emulator/emulator', 'Path to emulator.')
flags.DEFINE_string('adb_path',
                    '~/Android/Sdk/platform-tools/adb', 'Path to ADB.')
flags.DEFINE_boolean('run_headless', True, 'Optionally turn off display.')

# Environment args.
flags.DEFINE_string('task_path', None, 'Path to task textproto file.')

# Pygame args.
flags.DEFINE_list('screen_size', '480,720', 'Screen width, height in pixels.')
flags.DEFINE_float('frame_rate', 1.0/30.0, 'Frame rate in seconds.')

FLAGS = flags.FLAGS


def _get_action_from_event(
    event: pygame.event.Event,
    screen: pygame.Surface,
    orientation: task_pb2.AdbCall.Rotate.Orientation) -> Dict[str, Any]:
  """Returns the current action by reading data from a pygame Event object."""

  act_type = action_type.ActionType.LIFT
  if event.type == pygame.MOUSEBUTTONDOWN:
    act_type = action_type.ActionType.TOUCH

  return {
      'action_type':
          np.array(act_type, dtype=np.int32),
      'touch_position':
          _scale_position(event.pos, screen, orientation),
  }


def _get_action_from_mouse(
    screen: pygame.Surface,
    orientation: task_pb2.AdbCall.Rotate.Orientation) -> Dict[str, Any]:
  """Returns the current action by reading data from the mouse."""

  act_type = action_type.ActionType.LIFT
  if pygame.mouse.get_pressed()[0]:
    act_type = action_type.ActionType.TOUCH

  return {
      'action_type':
          np.array(act_type, dtype=np.int32),
      'touch_position':
          _scale_position(pygame.mouse.get_pos(), screen, orientation),
  }


def _scale_position(
    position: np.ndarray,
    screen: pygame.Surface,
    orientation: task_pb2.AdbCall.Rotate.Orientation) -> np.ndarray:
  """AndroidEnv accepts mouse inputs as floats so we need to scale it."""

  scaled_pos = np.divide(position, screen.get_size(), dtype=np.float32)
  if orientation == task_pb2.AdbCall.Rotate.Orientation.LANDSCAPE_90:
    scaled_pos = scaled_pos[::-1]
    scaled_pos[0] = 1 - scaled_pos[0]
  return scaled_pos


def _accumulate_reward(
    timestep: dm_env.TimeStep,
    episode_return: float) -> float:
  """Accumulates rewards collected over the course of an episode."""

  if timestep.reward and timestep.reward != 0:
    logging.info('Reward: %s', timestep.reward)
    episode_return += timestep.reward

  if timestep.first():
    episode_return = 0
  elif timestep.last():
    logging.info('Episode return: %s', episode_return)

  return episode_return


def _render_pygame_frame(
    surface: pygame.Surface,
    screen: pygame.Surface,
    orientation: task_pb2.AdbCall.Rotate.Orientation,
    timestep: dm_env.TimeStep) -> None:
  """Displays latest observation on pygame surface."""

  frame = timestep.observation['pixels'][:, :, :3]  # (H x W x C) (RGB)
  frame = utils.transpose_pixels(frame)  # (W x H x C)
  frame = utils.orient_pixels(frame, orientation)

  pygame.surfarray.blit_array(surface, frame)
  pygame.transform.smoothscale(surface, screen.get_size(), screen)

  pygame.display.flip()


def main(_):

  pygame.init()
  pygame.display.set_caption('android_human_agent')

  with android_env.load(
      emulator_path=FLAGS.emulator_path,
      android_sdk_root=FLAGS.android_sdk_root,
      android_avd_home=FLAGS.android_avd_home,
      avd_name=FLAGS.avd_name,
      adb_path=FLAGS.adb_path,
      task_path=FLAGS.task_path,
      run_headless=FLAGS.run_headless) as env:

    # Reset environment.
    first_timestep = env.reset()
    orientation = np.argmax(first_timestep.observation['orientation'])

    # Create pygame canvas.
    screen_size = list(map(int, FLAGS.screen_size))  # (W x H)
    obs_shape = env.observation_spec()['pixels'].shape[:2]  # (H x W)

    if (orientation == task_pb2.AdbCall.Rotate.Orientation.LANDSCAPE_90 or
        orientation == task_pb2.AdbCall.Rotate.Orientation.LANDSCAPE_270):
      screen_size = screen_size[::-1]
      obs_shape = obs_shape[::-1]

    screen = pygame.display.set_mode(screen_size)  # takes (W x H)
    surface = pygame.Surface(obs_shape[::-1])  # takes (W x H)

    # Start game loop.
    prev_frame = time.time()
    episode_return = 0

    while True:
      if pygame.key.get_pressed()[pygame.K_ESCAPE]:
        return

      all_events = pygame.event.get()
      for event in all_events:
        if event.type == pygame.QUIT:
          return

      # Filter event queue for mouse click events.
      mouse_click_events = [
          event for event in all_events
          if event.type in [pygame.MOUSEBUTTONDOWN, pygame.MOUSEBUTTONUP]
      ]

      # Process all mouse click events.
      for event in mouse_click_events:
        action = _get_action_from_event(event, screen, orientation)
        timestep = env.step(action)
        episode_return = _accumulate_reward(timestep, episode_return)
        _render_pygame_frame(surface, screen, orientation, timestep)

      # Sample the current position of the mouse either way.
      action = _get_action_from_mouse(screen, orientation)
      timestep = env.step(action)
      episode_return = _accumulate_reward(timestep, episode_return)
      _render_pygame_frame(surface, screen, orientation, timestep)

      # Limit framerate.
      now = time.time()
      frame_time = now - prev_frame
      if frame_time < FLAGS.frame_rate:
        time.sleep(FLAGS.frame_rate - frame_time)
      prev_frame = now


if __name__ == '__main__':
  logging.set_verbosity('info')
  logging.set_stderrthreshold('info')
  flags.mark_flags_as_required(['avd_name', 'task_path'])
  app.run(main)
<file_sep># coding=utf-8
# Copyright 2021 DeepMind Technologies Limited.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

"""Coordinator handles interaction between internal components of AndroidEnv."""

import copy
import socket
import time
from typing import Any, Dict, Optional, Tuple

from absl import logging
from android_env.components import action_type as action_type_lib
from android_env.components import base_simulator
from android_env.components import errors
from android_env.components import specs
from android_env.components import task_manager as task_manager_lib
import dm_env
import numpy as np


class Coordinator():
  """Handles interaction between internal components of AndroidEnv."""

  def __init__(
      self,
      simulator: base_simulator.BaseSimulator,
      task_manager: task_manager_lib.TaskManager,
      step_timeout_sec: int = 10,
      max_steps_per_sec: float = 5.0,
      periodic_restart_time_min: float = 0.0,
      force_simulator_launch: bool = True,
  ):
    """Handles communication between AndroidEnv and its components.

    Args:
      simulator: A BaseSimulator instance.
      task_manager: The TaskManager, responsible for coordinating RL tasks.
      step_timeout_sec: Timeout in seconds between steps. If step is not called
        within that time, the episode will reset at the next step. Set to 0 to
        disable.
      max_steps_per_sec: Maximum steps per second. If the simulator is
        faster, the Coordinator will wait before returning an observation.
      periodic_restart_time_min: Time between periodic restarts in minutes.
        If > 0.0, will trigger a simulator restart at the end of the next
        episode once the time has been reached.
      force_simulator_launch: Forces the simulator to relaunch even if it is
        already launched.
    """
    self._simulator = simulator
    self._task_manager = task_manager
    self._step_timeout_sec = step_timeout_sec
    self._max_steps_per_sec = max_steps_per_sec
    self._periodic_restart_time_min = periodic_restart_time_min
    self._force_simulator_launch = force_simulator_launch

    # Logging settings.
    self._log_dict = {
        'total_steps': 0,
        'episode_steps': 0,
        'restart_count': 0,
        'restart_count_periodic': 0,
        'restart_count_setup_steps': 0,
        'restart_count_reset_steps': 0,
        'restart_count_simulator_launch': 0,
        'restart_count_simulator_reset': 0,
        'restart_count_execute_action': 0,
        'restart_count_fetch_observation': 0,
        'reset_count_step_timeout': 0,
    }

    # Initialize counters.
    self._should_restart = False
    self._latest_observation_time = None
    self._simulator_start_time = None

    self._restart_simulator()

  def action_spec(self) -> Dict[str, dm_env.specs.Array]:
    return specs.base_action_spec()

  def observation_spec(self) -> Dict[str, dm_env.specs.Array]:
    screen_dims = self._simulator.screen_dimensions()
    return specs.base_observation_spec(
        height=screen_dims[0], width=screen_dims[1])

  def task_extras_spec(self) -> Dict[str, dm_env.specs.Array]:
    return specs.base_task_extras_spec(task=self._task_manager.task())

  def reset_environment_state(self):
    """Resets the state of the simulation for a new RL episode.

    This involves resetting relevant counters and performing reset steps
    specific to the running task (e.g. restarting an application). A lift
    action is also performed to ensure that sequential touches are not
    interconnected between separate RL episodes.
    """

    # Restart the simulation if neccessary.
    if self._should_restart or self._should_periodic_restart():
      self._restart_simulator()

    # Reset counters.
    self._latest_observation_time = None
    for key in self._log_dict:
      if key.startswith('episode'):
        self._log_dict[key] = 0.0

    # Execute a lift action before resetting the task.
    self._send_action_to_simulator({
        'action_type': np.array(action_type_lib.ActionType.LIFT),
        'touch_position': np.array([0, 0]),
    })

    # Reset the task.
    try:
      self._task_manager.reset_task()
      self._simulator.update_device_orientation()
    except errors.StepCommandError:
      logging.exception('Failed to reset the task. Restarting simulator.')
      self._log_dict['restart_count_simulator_reset'] += 1
      self._should_restart = True

  def _should_periodic_restart(self) -> bool:
    """Checks if it is time to restart the simulator.

    If a periodic restart time was specified, the Coordinator will re-launch
    the simulator at regular time intervals. This helps to make sure that the
    simulator is not is a stale state even if the environment has been running
    for a significant amount of time.

    Returns:
      Boolean indicating if it is time to restart the simulator.
    """

    if self._periodic_restart_time_min and self._simulator_start_time:
      sim_alive_time = (time.time() - self._simulator_start_time) / 60.0
      logging.info('Simulator has been running for %f mins', sim_alive_time)
      if sim_alive_time > self._periodic_restart_time_min:
        logging.info('Maximum alive time reached. Restarting simulator.')
        self._log_dict['restart_count_periodic'] += 1
        return True
    return False

  def _restart_simulator(self, max_retries: int = 3):
    """Restarts the simulation.

    Closes and re-launches the system, restarting the simulator process and
    reinitializing the task in the newly started simulator.

    Args:
      max_retries: Number of times to attempt a restart before raising an error.
    """

    # Reset counters.
    self._should_restart = False

    # Attempt to restart the system a given number of times.
    num_tries = 1
    while True:
      if num_tries > max_retries:
        logging.error('Maximum number of restarts reached.')
        raise errors.TooManyRestartsError
      logging.info('Simulator launch attempt %d of %d', num_tries, max_retries)

      # Launch the simulator (will restart if already launched).
      try:
        if self._force_simulator_launch or not self._simulator.is_launched():
          self._task_manager.pause_task()
          self._simulator.launch()
          self._simulator_start_time = time.time()
        adb_controller = self._simulator.create_adb_controller()
      except errors.AdbControllerError:
        logging.error('Error launching the simulator.')
        self._log_dict['restart_count_simulator_launch'] += 1
        num_tries += 1
        continue

      # Start the task.
      try:
        self._task_manager.setup_task(adb_controller=adb_controller)
      except errors.StepCommandError:
        logging.error('Failed to set up the task. Restarting simulator.')
        self._log_dict['restart_count_setup_steps'] += 1
        num_tries += 1
        continue

      # Restart was successful.
      break

  def execute_action(
      self,
      action: Optional[Dict[str, np.ndarray]],
  ) -> Tuple[Optional[Dict[str, np.ndarray]], float, Dict[str, Any], bool]:
    """Executes the selected action and returns transition info.

    Args:
      action: Selected action to perform on the simulated Android device.
    Returns:
      observation: Pixel observations as displayed on the screen.
      reward: Total reward collected since the last call.
      extras: Task extras observed since the last call.
      episode_end: Boolean indicating if the RL episode should be terminated.
    """

    # Increment counters.
    self._task_manager.increment_steps()
    if action is not None:
      self._log_dict['total_steps'] += 1
      self._log_dict['episode_steps'] += 1

    # If a restart is neccessary, end the episode.
    if self._should_restart or self._check_timeout():
      return None, 0.0, {}, True

    # If the action is a TOUCH or LIFT, send it to the simulator.
    if (action is not None and
        action['action_type'].item() != action_type_lib.ActionType.REPEAT):
      self._send_action_to_simulator(action)

    # Sleep to maintain a steady interaction rate.
    self._wait_for_next_frame()

    # Read necessary transition information and return it to the agent.
    try:
      self._latest_observation_time = time.time()
      observation = self._simulator.get_observation()
      reward = self._task_manager.get_current_reward()
      task_extras = self._task_manager.get_current_extras()
      episode_end = self._task_manager.check_if_episode_ended()
      return observation, reward, task_extras, episode_end
    except (errors.ReadObservationError, socket.error):
      logging.exception('Unable to fetch observation. Restarting simulator.')
      self._log_dict['restart_count_fetch_observation'] += 1
      self._should_restart = True
      return None, 0.0, {}, True

  def _send_action_to_simulator(self, action: Dict[str, np.ndarray]) -> None:
    """Sends the selected action to the simulator.

    The simulator will interpret the action as a touchscreen event and perform
    it accordingly. The effect this action triggers in the Android OS will be
    determined by the currently running application.

    Args:
      action: action which will get interpreted as a touchscreen event.
    """

    try:
      self._simulator.send_action(action)
    except (socket.error, errors.SendActionError):
      logging.exception('Unable to execute action. Restarting simulator.')
      self._log_dict['restart_count_execute_action'] += 1
      self._should_restart = True

  def _check_timeout(self) -> bool:
    """Checks if timeout between steps have exceeded.

    If too much time has passed since the last step was performed, it is assumed
    that the simulation is in a bad state, so the Coordinator will re-launch
    the simulator to make sure interaction proceeds from a clean state.

    Returns:
      Boolean indicating if the step timeout limit has been reached.
    """

    if self._step_timeout_sec and self._latest_observation_time:
      time_since_last_obs = self._get_time_since_last_observation()
      if time_since_last_obs > self._step_timeout_sec:
        self._should_restart = True
        return True
    return False

  def _wait_for_next_frame(self) -> None:
    """Pauses the environment so that the interaction is around 1/FPS."""

    time_since_observation = self._get_time_since_last_observation()
    time_to_wait = 1. / self._max_steps_per_sec - time_since_observation
    if time_to_wait > 0.0:
      time.sleep(time_to_wait)

  def _get_time_since_last_observation(self) -> float:
    """Computes time passed since the last observation was fetched."""

    if self._latest_observation_time is not None:
      return time.time() - self._latest_observation_time
    else:
      return np.inf

  def get_logs(self) -> Dict[str, Any]:
    """Returns internal counter values."""

    log_dict = copy.deepcopy(self._log_dict)
    log_dict.update(self._task_manager.log_dict())
    return log_dict

  def close(self):
    """Cleans up the state of this Coordinator."""

    if hasattr(self, '_task_manager'):
      self._task_manager.close()
    if hasattr(self, '_simulator'):
      self._simulator.close()
<file_sep># coding=utf-8
# Copyright 2021 DeepMind Technologies Limited.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

"""A ThreadFunction that runs and parses adb dumpsys."""

import enum

from absl import logging

from android_env.components import app_screen_checker as screen_checker
from android_env.components import thread_function

AppScreenChecker = screen_checker.AppScreenChecker


class DumpsysThread(thread_function.ThreadFunction):
  """A class that executes dumpsys in a separate thread."""

  class Signal(enum.IntEnum):
    """Defines commands we can use to communicate with the dumpsys thread."""
    # To ask the thread to fetch dumpsys from Android.
    FETCH_DUMPSYS = 0
    # The user has left the activity that contains the AndroidEnv task.
    USER_EXITED_ACTIVITY = 1
    # The user exited the view hierarchy that we expect.
    USER_EXITED_VIEW_HIERARCHY = 2
    # App screen checker determined none of the errors above happened.
    OK = 3
    # App screen checker was not queried.
    DID_NOT_CHECK = 4

  def __init__(
      self,
      app_screen_checker: AppScreenChecker,
      check_frequency: int,
      max_failed_current_activity: int,
      block_input: bool,
      block_output: bool,
      name: str = 'dumpsys',
  ):
    """Initializes the dumpsys reader thread.

    This loops forever waiting for inputs from the main thread and outputting
    its analyses of the output of ADB dumpsys. These analyses are too expensive
    to be in the critical path of AndroidEnv::step() so we consume them async
    from this separate thread.

    Args:
      app_screen_checker: The class that actually determines if the current
          screen matches the expected screen.
      check_frequency: Integer. We only call dumpsys 1/check_frequency times in
          each iteration of the while loop below.
      max_failed_current_activity: Integer. We try to fetch the current activity
          but sometimes it fails. If it fails more than
          `max_failed_current_activity` consecutive times, we declare that the
          user has exited `expected_activity`.
      block_input: Whether to block this thread when reading its input queue.
      block_output: Whether to block this thread when writing to its output
        queue.
      name: Name of the thread.
    """

    self._app_screen_checker = app_screen_checker
    self._main_loop_counter = 0
    self._check_frequency = check_frequency
    self._max_failed_activity_extraction = max_failed_current_activity
    self._num_failed_activity_extraction = 0
    super().__init__(
        block_input=block_input, block_output=block_output, name=name)

  def main(self):
    v = self._read_value()
    if v != DumpsysThread.Signal.FETCH_DUMPSYS:
      self._write_value(DumpsysThread.Signal.DID_NOT_CHECK)
      return

    # Update and check loop_counter against check_frequency.
    self._main_loop_counter += 1
    if (self._check_frequency <= 0 or
        self._main_loop_counter < self._check_frequency):
      self._write_value(DumpsysThread.Signal.DID_NOT_CHECK)
      return
    self._main_loop_counter = 0

    outcome = self._app_screen_checker.matches_current_app_screen()

    # We were unable to determine the current activity.
    if outcome == AppScreenChecker.Outcome.FAILED_ACTIVITY_EXTRACTION:
      self._num_failed_activity_extraction += 1
      logging.info('self._num_failed_activity_extraction: %s',
                   self._num_failed_activity_extraction)
      if (self._num_failed_activity_extraction >=
          self._max_failed_activity_extraction):
        logging.error('Maximum number of failed activity extraction reached.')
        self._num_failed_activity_extraction = 0
        self._write_value(DumpsysThread.Signal.USER_EXITED_ACTIVITY)
        return
    else:
      self._num_failed_activity_extraction = 0

    # The current app screen matches all expectations.
    if (outcome == AppScreenChecker.Outcome.SUCCESS or
        outcome == AppScreenChecker.Outcome.EMPTY_EXPECTED_ACTIVITY):
      self._write_value(DumpsysThread.Signal.OK)
      return

    # Player has exited the app. Terminate the episode.
    elif outcome == AppScreenChecker.Outcome.UNEXPECTED_ACTIVITY:
      self._write_value(DumpsysThread.Signal.USER_EXITED_ACTIVITY)
      return

    # Player has exited the main game. Terminate the episode.
    elif outcome == AppScreenChecker.Outcome.UNEXPECTED_VIEW_HIERARCHY:
      self._write_value(DumpsysThread.Signal.USER_EXITED_VIEW_HIERARCHY)
      return
<file_sep># coding=utf-8
# Copyright 2021 DeepMind Technologies Limited.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

"""Base specs for AndroidEnv."""

from typing import Dict

from android_env.components import action_type
from android_env.proto import task_pb2
import dm_env
from dm_env import specs
import numpy as np


_PROTO_DTYPE_TO_NUMPY_DTYPE = {
    task_pb2.ArraySpec.DataType.FLOAT: np.float32,
    task_pb2.ArraySpec.DataType.DOUBLE: np.float64,
    task_pb2.ArraySpec.DataType.INT8: np.int8,
    task_pb2.ArraySpec.DataType.INT16: np.int16,
    task_pb2.ArraySpec.DataType.INT32: np.int32,
    task_pb2.ArraySpec.DataType.INT64: np.int64,
    task_pb2.ArraySpec.DataType.UINT8: np.uint8,
    task_pb2.ArraySpec.DataType.UINT16: np.uint16,
    task_pb2.ArraySpec.DataType.UINT32: np.uint32,
    task_pb2.ArraySpec.DataType.UINT64: np.uint64,
    task_pb2.ArraySpec.DataType.BOOL: np.bool_,
    task_pb2.ArraySpec.DataType.STRING_U1: np.dtype(('U1')),
    task_pb2.ArraySpec.DataType.STRING_U16: np.dtype(('<U16')),
    task_pb2.ArraySpec.DataType.STRING_U25: np.dtype(('<U25')),
    task_pb2.ArraySpec.DataType.STRING_U250: np.dtype(('<U250')),
}


def base_action_spec() -> Dict[str, specs.Array]:
  """Default action spec for AndroidEnv.

  Returns:
    action_type: An integer of type ActionType: TOUCH=0, LIFT=1, REPEAT=2
    touch_position: Position [x, y] of the touch action, where x, y are float
      values between 0.0 and 1.0 corresponding to the relative position on the
      screen. IGNORED when (action_type != ActionType.TOUCH).
  """

  return {
      'action_type':
          specs.DiscreteArray(
              num_values=len(action_type.ActionType),
              name='action_type'),
      'touch_position':
          specs.BoundedArray(
              shape=(2,),
              dtype=np.float32,
              minimum=[0.0, 0.0],
              maximum=[1.0, 1.0],
              name='touch_position'),
  }


def base_observation_spec(height: int, width: int) -> Dict[str, specs.Array]:
  """Default observation spec for AndroidEnv.

  Args:
    height: Height of the device screen in pixels.
    width: Width of the device screen in pixels.
  Returns:
    pixels: Spec for the RGB screenshot of the device. Has shape (H, W, 3)
    timedelta: Spec for time delta since the last observation (in microseconds).
    orientation: Spec for the latest orientation in a one-hot representation:
        [1, 0, 0, 0]: PORTRAIT  (0 degrees)
        [0, 1, 0, 0]: LANDSCAPE (90 degrees clockwise)
        [0, 0, 1, 0]: PORTRAIT  (180 degrees) ("upside down")
        [0, 0, 0, 1]: LANDSCAPE (270 degrees clockwise)
  """

  return {
      'pixels':
          specs.Array(
              shape=(height, width, 3),
              dtype=np.uint8,
              name='pixels'),
      'timedelta':
          specs.Array(shape=(), dtype=np.int64, name='timedelta'),
      'orientation':
          specs.Array(shape=np.array([4]), dtype=np.uint8, name='orientation'),
  }


def base_task_extras_spec(task: task_pb2.Task) -> Dict[str, dm_env.specs.Array]:
  """Task extras spec for AndroidEnv, as read from a task_pb2.Task."""

  return {
      spec.name: _convert_spec(spec)
      for spec in task.extras_spec
  }


def _convert_spec(array_spec: task_pb2.ArraySpec) -> specs.Array:
  """Converts ArraySpec proto to dm_env specs.Array."""

  return specs.Array(
      shape=array_spec.shape,
      dtype=_PROTO_DTYPE_TO_NUMPY_DTYPE[array_spec.dtype],
      name=array_spec.name)
<file_sep># coding=utf-8
# Copyright 2021 DeepMind Technologies Limited.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

"""Tests for android_env.components.coordinator."""

import time

from absl.testing import absltest
from android_env.components import action_type
from android_env.components import coordinator
from android_env.components import emulator_simulator
from android_env.components import errors
from android_env.components import task_manager
import mock
import numpy as np


class CoordinatorTest(absltest.TestCase):

  def setUp(self):
    super().setUp()
    self.addCleanup(mock.patch.stopall)  # Disable previous patches.

    self._simulator = mock.create_autospec(emulator_simulator.EmulatorSimulator)
    self._task_manager = mock.create_autospec(task_manager.TaskManager)

    self._coordinator = coordinator.Coordinator(
        simulator=self._simulator,
        task_manager=self._task_manager,
        step_timeout_sec=2,
        max_steps_per_sec=60,
        periodic_restart_time_min=0)

  def test_restart_simulator(self):
    self._coordinator._restart_simulator()

  def test_reset(self):
    self._coordinator.reset_environment_state()

  def test_process_action(self):
    self._simulator.get_observation.return_value = {'observation': 0}
    self._task_manager.get_current_reward.return_value = 10.0
    self._task_manager.get_current_extras.return_value = {'extra': [0.0]}
    self._task_manager.check_if_episode_ended.return_value = False
    obs, reward, task_extras, episode_end = self._coordinator.execute_action(
        action={'action_type': np.array(action_type.ActionType.LIFT)})
    self.assertDictEqual(obs, {'observation': 0})
    self.assertEqual(reward, 10.0)
    self.assertEqual(task_extras, {'extra': [0.0]})
    self.assertFalse(episode_end)

  def test_process_action_error(self):
    self._simulator.get_observation.side_effect = errors.ReadObservationError()
    self._task_manager.get_current_reward.return_value = 0.0
    self._task_manager.get_current_extras.return_value = {}
    self._task_manager.check_if_episode_ended.return_value = True
    obs, reward, task_extras, episode_end = self._coordinator.execute_action(
        action={'action_type': np.array(action_type.ActionType.LIFT)})
    self.assertTrue(self._coordinator._should_restart)
    self.assertIsNone(obs)
    self.assertEqual(reward, 0.0)
    self.assertEqual(task_extras, {})
    self.assertTrue(episode_end)

  def test_execute_action_touch(self):
    self._simulator.get_observation.return_value = {'observation': 0}
    self._simulator.send_action.return_value = True
    action = {'action_type': np.array(action_type.ActionType.TOUCH)}
    _ = self._coordinator.execute_action(action)
    self._simulator.send_action.assert_called_once_with(action)

  def test_execute_action_repeat(self):
    self._simulator.get_observation.return_value = {'observation': 0}
    self._simulator.send_action.return_value = True
    _ = self._coordinator.execute_action(
        {'action_type': np.array(action_type.ActionType.REPEAT)})
    self._simulator.send_action.assert_not_called()

  def test_execute_action_error(self):
    self._simulator.get_observation.return_value = {'observation': 0}
    self._simulator.send_action.side_effect = errors.SendActionError
    _ = self._coordinator.execute_action(
        {'action_type': np.array(action_type.ActionType.TOUCH)})
    self.assertTrue(self._coordinator._should_restart)

  def test_check_timeout_false(self):
    self._coordinator._latest_observation_time = time.time()
    timeout = self._coordinator._check_timeout()
    self.assertFalse(timeout)

  def test_check_timeout_true(self):
    self._coordinator._latest_observation_time = time.time()
    time.sleep(3)
    timeout = self._coordinator._check_timeout()
    self.assertTrue(timeout)

  def test_max_restarts_adb_error(self):
    # The method was called once at init.
    init_fn_call = self._simulator.create_adb_controller.call_count
    self._simulator.create_adb_controller.side_effect = (
        errors.AdbControllerError)
    self.assertRaises(errors.TooManyRestartsError,
                      self._coordinator._restart_simulator)
    # The method was called three more times when attempting to restart.
    self.assertEqual(init_fn_call + 3,
                     self._simulator.create_adb_controller.call_count)

  def test_max_restarts_setup_steps(self):
    init_fn_call = self._task_manager.setup_task.call_count
    self._task_manager.setup_task.side_effect = errors.StepCommandError
    self.assertRaises(errors.TooManyRestartsError,
                      self._coordinator._restart_simulator)
    # The method was called three more times when attempting to restart.
    self.assertEqual(init_fn_call + 3,
                     self._task_manager.setup_task.call_count)


if __name__ == '__main__':
  absltest.main()
<file_sep># coding=utf-8
# Copyright 2021 DeepMind Technologies Limited.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

"""Wraps the AndroidEnv environment to provide discrete actions."""

from typing import Optional, Sequence, Dict

from android_env.components import action_type
from android_env.wrappers import base_wrapper
import dm_env
from dm_env import specs
import numpy as np


NOISE_CLIP_VALUE = 0.4999


class DiscreteActionWrapper(base_wrapper.BaseWrapper):
  """AndroidEnv with discrete actions."""

  def __init__(self,
               env: dm_env.Environment,
               action_grid: Optional[Sequence[int]] = (10, 10),
               redundant_actions: bool = True,
               noise: float = 0.1):

    super().__init__(env)
    self._parent_action_spec = self._env.action_spec()
    self._assert_base_env()
    self._action_grid = action_grid  # [height, width]
    self._grid_size = np.product(self._action_grid)
    self._num_action_types = self._parent_action_spec['action_type'].num_values
    self._redundant_actions = redundant_actions
    self._noise = noise

  def _assert_base_env(self):
    """Checks that the wrapped env has the right action spec format."""

    assert len(self._parent_action_spec) == 2
    assert not self._parent_action_spec['action_type'].shape
    assert self._parent_action_spec['touch_position'].shape == (2,)

  @property
  def num_actions(self) -> int:
    """Number of discrete actions."""

    if self._redundant_actions:
      return np.product(self._action_grid) * self._num_action_types
    else:
      return np.product(self._action_grid) + self._num_action_types - 1

  def step(self, action: Dict[str, int]) -> dm_env.TimeStep:
    """Take a step in the base environment."""

    return self._env.step(self._process_action(action))

  def _process_action(self, action: Dict[str, int]) -> Dict[str, np.ndarray]:
    """Transforms action so that it agrees with AndroidEnv's action spec."""

    return {
        'action_type':
            np.array(self._get_action_type(action['action_id']),
                     dtype=self._parent_action_spec['action_type'].dtype),
        'touch_position':
            np.array(self._get_touch_position(action['action_id']),
                     dtype=self._parent_action_spec['touch_position'].dtype)
    }

  def _get_action_type(self, action_id: int) -> action_type.ActionType:
    """Compute action type corresponding to the given action_id.

    When `self._redundant_actions` == True the `grid_size` is "broadcast" over
    all the possible actions so you end up with `grid_size` discrete actions
    of type 0, `grid_size` discrete actions of type 1, etc. for all action
    types.

    When `self._redundant_actions` == False the first `grid_size` actions are
    reserved for "touch" and the rest are just added (NOT multiplied) to the
    total number of discrete actions (exactly one of LIFT and REPEAT).

    Args:
      action_id: A discrete action.
    Returns:
      action_type: The action_type of the action.
    """

    if self._redundant_actions:
      assert action_id < self._num_action_types * self._grid_size
      return action_id // self._grid_size

    else:
      assert action_id <= self._grid_size + 1
      if action_id < self._grid_size:
        return action_type.ActionType.TOUCH
      elif action_id == self._grid_size:
        return action_type.ActionType.LIFT
      else:
        return action_type.ActionType.REPEAT

  def _get_touch_position(self, action_id: int) -> Sequence[float]:
    """Compute the position corresponding to the given action_id.

    Note: in the touch_position (x, y) of an action, x corresponds to the
    horizontal axis (width), and y corresponds to the vertical axis (height)
    of the screen. BUT, the screen has dimensions (height, width), i.e. the
    first coordinate corresponds to y, and the second coordinate corresponds
    to x. Pay attention to this mismatch in the calculations below.

    Args:
      action_id: A discrete action.
    Returns:
      touch_position: The [0,1]x[0,1] coordinate of the action.
    """

    position_idx = action_id % self._grid_size

    x_pos_grid = position_idx % self._action_grid[1]  # WIDTH
    y_pos_grid = position_idx // self._action_grid[1]  # HEIGHT

    noise_x = np.random.normal(loc=0.0, scale=self._noise)
    noise_y = np.random.normal(loc=0.0, scale=self._noise)

    # Noise is clipped so that the action will strictly stay in the cell.
    noise_x = max(min(noise_x, NOISE_CLIP_VALUE), -NOISE_CLIP_VALUE)
    noise_y = max(min(noise_y, NOISE_CLIP_VALUE), -NOISE_CLIP_VALUE)

    x_pos = (x_pos_grid + 0.5 + noise_x) / self._action_grid[1]  # WIDTH
    y_pos = (y_pos_grid + 0.5 + noise_y) / self._action_grid[0]  # HEIGHT

    # Project action space to action_spec ranges. For the default case of
    # minimum = [0, 0] and maximum = [1, 1], this will not do anything.
    x_min, y_min = self._parent_action_spec['touch_position'].minimum
    x_max, y_max = self._parent_action_spec['touch_position'].maximum

    x_pos = x_min + x_pos * (x_max - x_min)
    y_pos = y_min + y_pos * (y_max - y_min)

    return [x_pos, y_pos]

  def action_spec(self) -> Dict[str, specs.Array]:
    """Action spec of the wrapped environment."""

    return {
        'action_id':
            specs.DiscreteArray(
                num_values=self.num_actions,
                name='action_id')
    }
<file_sep># coding=utf-8
# Copyright 2021 DeepMind Technologies Limited.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

"""Tests for android_env.components.emulator_console."""

import builtins
import os
import telnetlib
import time

from absl.testing import absltest
from android_env.components import emulator_console
from android_env.components import errors
from android_env.proto import raw_observation_pb2
import mock
import numpy as np


class EmulatorConsoleTest(absltest.TestCase):

  def setUp(self):
    super().setUp()
    self.addCleanup(mock.patch.stopall)  # Disable previous patches.

    # Create a mock file with a fake auth code.
    self._mock_auth_file = mock.MagicMock()
    self._mock_auth_file.__enter__ = mock.MagicMock(
        return_value=self._mock_auth_file)
    self._mock_auth_file.read.return_value = 'some_code_i_dont_care'

    # Create a mock file to hold the FIFO.
    self._mock_fifo_file = mock.MagicMock()
    self._mock_fifo_file.__enter__ = mock.MagicMock(
        return_value=self._mock_fifo_file)

    self._mock_open = mock.patch.object(builtins, 'open', autospec=True).start()

    def fake_open(fname, mode=''):
      """A closure that returns auth file the first time, then fifo."""
      del fname, mode
      fake_open.open_counter += 1
      if fake_open.open_counter == 1:
        return self._mock_auth_file
      return self._mock_fifo_file

    fake_open.open_counter = 0  # Function attribute ("static" local variable).
    self._mock_open.side_effect = fake_open

  @mock.patch.object(time, 'sleep', autospec=True)
  @mock.patch.object(
      telnetlib,
      'Telnet',
      side_effect=ConnectionRefusedError('oops'),
      autospec=True)
  def test_connection_error(self, telnet_connection, mock_sleep):
    self.assertRaises(
        errors.ConsoleConnectionError,
        emulator_console.EmulatorConsole,
        console_port=1234,
        auth_code='<PASSWORD>',
        tmp_dir=absltest.get_default_test_tmpdir())
    telnet_connection.assert_has_calls([mock.call('localhost', 1234)] * 3)
    mock_sleep.assert_called()

  def test_auth_error(self):
    telnet_connection = mock.create_autospec(telnetlib.Telnet)
    telnet_connection.read_until.return_value = 'whatever. I will not be used'
    telnet_connection.read_until.side_effect = EOFError('oh no')

    with mock.patch.object(
        telnetlib, 'Telnet', autospec=True,
        return_value=telnet_connection) as telnet_init:
      self.assertRaises(
          EOFError,
          emulator_console.EmulatorConsole,
          console_port=1234,
          auth_code='<PASSWORD>',
          tmp_dir=absltest.get_default_test_tmpdir())
      telnet_init.assert_called_once_with('localhost', 1234)

  @mock.patch.object(os.path, 'expanduser', autospec=True)
  def test_no_auth_should_look_in_home_folder(self, mock_expanduser):
    telnet_connection = mock.create_autospec(telnetlib.Telnet)
    telnet_connection.read_until.return_value = 'OK'

    with mock.patch.object(
        telnetlib, 'Telnet', autospec=True,
        return_value=telnet_connection) as telnet_init:
      console = emulator_console.EmulatorConsole(
          console_port=1234,
          auth_code=None,
          tmp_dir=absltest.get_default_test_tmpdir())
      telnet_init.assert_called_once_with('localhost', 1234)
      mock_expanduser.assert_called_once()
      console.close()

  @mock.patch.object(os, 'remove', autospec=True)
  @mock.patch.object(os.path, 'isfile', autospec=True)
  def test_existing_fifo_should_be_deleted(self, mock_isfile, mock_remove):
    telnet_connection = mock.create_autospec(telnetlib.Telnet)
    telnet_connection.read_until.return_value = 'OK'

    # Pretend that a fifo with that name already exists so that we verify that
    # EmulatorConsole is deleting and recreating it.
    mock_isfile.return_value = True

    self._mock_fifo_file.read.return_value = b''

    with mock.patch.object(
        telnetlib, 'Telnet', autospec=True,
        return_value=telnet_connection) as telnet_init:
      console = emulator_console.EmulatorConsole(
          console_port=1234,
          auth_code=None,
          tmp_dir=absltest.get_default_test_tmpdir())
      telnet_init.assert_called_once_with('localhost', 1234)
      mock_isfile.assert_called_once()
      mock_remove.assert_called_once()

      console.close()

  @mock.patch.object(os, 'remove', autospec=True)
  @mock.patch.object(os, 'close', autospec=True)
  def test_send_mouse_action(self, mock_close, mock_remove):
    telnet_connection = mock.create_autospec(telnetlib.Telnet)
    telnet_connection.read_until.return_value = 'OK'

    with mock.patch.object(
        telnetlib, 'Telnet', autospec=True,
        return_value=telnet_connection) as telnet_init:
      console = emulator_console.EmulatorConsole(
          console_port=1234,
          auth_code=None,
          tmp_dir=absltest.get_default_test_tmpdir())
      telnet_init.assert_called_once_with('localhost', 1234)

    console.send_mouse_action(7, 8, True)
    telnet_connection.write.assert_called()

    # Close the console.
    console.close()

    # Because fetch_screenshot() was not called, the pipe was not actually
    # opened so we don't expect it to be closed.
    mock_close.assert_has_calls([])
    mock_remove.assert_has_calls([])

  def test_fetch_screenshot_timedout(self):
    """Ensures that we get an exception if `fetch_screenshot` takes >20s."""
    telnet_connection = mock.create_autospec(telnetlib.Telnet)
    telnet_connection.read_until.return_value = 'OK'
    pipe_read_timeout = 1.0

    def stuck_read():
      """A .read() call that takes a long time to return something."""
      time.sleep(pipe_read_timeout + 5)
      return b'hello there'

    self._mock_fifo_file.read.side_effect = stuck_read

    with mock.patch.object(
        telnetlib, 'Telnet', autospec=True,
        return_value=telnet_connection) as telnet_init:
      console = emulator_console.EmulatorConsole(
          console_port=1234,
          auth_code=None,
          tmp_dir=absltest.get_default_test_tmpdir(),
          pipe_read_timeout_sec=pipe_read_timeout)
      telnet_init.assert_called_once_with('localhost', 1234)

    self.assertRaises(errors.PipeTimedOutError, console.fetch_screenshot)

    console.close()

  def test_fetch_screenshot_io_error(self):
    telnet_connection = mock.create_autospec(telnetlib.Telnet)
    telnet_connection.read_until.return_value = 'OK'

    # Create a fake flat image with 4 channels.
    fake_img = np.array(list(range(10)) * 4, dtype=np.uint8)
    fake_raw_obs = raw_observation_pb2.RawObservation()
    fake_raw_obs.screen.data = fake_img.tobytes()
    fake_raw_obs.screen.height = 5
    fake_raw_obs.screen.width = 2
    fake_raw_obs.screen.num_channels = 4
    fake_raw_obs.timestamp_us = 123456789

    # Setup fifo as thread starts reading in the constructor.
    def io_error_read():
      io_error_read.counter += 1
      if io_error_read.counter % 2 == 1:
        return fake_raw_obs.SerializeToString()
      raise IOError('Nooo, f.read() crashed!')

    io_error_read.counter = 0
    self._mock_fifo_file.read.side_effect = io_error_read

    with mock.patch.object(
        telnetlib, 'Telnet', autospec=True,
        return_value=telnet_connection) as telnet_init:
      console = emulator_console.EmulatorConsole(
          console_port=1234,
          auth_code=None,
          tmp_dir=absltest.get_default_test_tmpdir())
      telnet_init.assert_called_once_with('localhost', 1234)

    self.assertRaises(IOError, console.fetch_screenshot)

    console.close()

  def test_fetch_screenshot_decoding_error(self):
    telnet_connection = mock.create_autospec(telnetlib.Telnet)
    telnet_connection.read_until.return_value = 'OK'

    # Setup fifo as thread starts reading in the constructor.
    def bad_proto_read():
      bad_proto_read.counter += 1
      if bad_proto_read.counter % 2 == 1:
        return b'I am definitely not a RawObservation!'
      else:
        return b''

    bad_proto_read.counter = 0
    self._mock_fifo_file.read.side_effect = bad_proto_read

    with mock.patch.object(
        telnetlib, 'Telnet', autospec=True,
        return_value=telnet_connection) as telnet_init:
      console = emulator_console.EmulatorConsole(
          console_port=1234,
          auth_code=None,
          tmp_dir=absltest.get_default_test_tmpdir())
      telnet_init.assert_called_once_with('localhost', 1234)

    self.assertRaises(errors.ObservationDecodingError, console.fetch_screenshot)

    console.close()

  def test_fetch_screenshot_ok(self):
    telnet_connection = mock.create_autospec(telnetlib.Telnet)
    telnet_connection.read_until.return_value = 'OK'

    # Create a fake flat image with 4 channels.
    fake_img = np.array(list(range(10)) * 4, dtype=np.uint8)
    fake_raw_obs = raw_observation_pb2.RawObservation()
    fake_raw_obs.screen.data = fake_img.tobytes()
    fake_raw_obs.screen.height = 5
    fake_raw_obs.screen.width = 2
    fake_raw_obs.screen.num_channels = 4
    fake_raw_obs.timestamp_us = 123456789

    # Setup fifo as thread starts reading in the constructor.
    def good_read():
      good_read.counter += 1
      if good_read.counter % 2 == 1:
        return fake_raw_obs.SerializeToString()
      else:
        return b''

    good_read.counter = 0
    self._mock_fifo_file.read.side_effect = good_read

    with mock.patch.object(
        telnetlib, 'Telnet', autospec=True,
        return_value=telnet_connection) as telnet_init:
      console = emulator_console.EmulatorConsole(
          console_port=1234,
          auth_code=None,
          tmp_dir=absltest.get_default_test_tmpdir())
      telnet_init.assert_called_once_with('localhost', 1234)

    observation = console.fetch_screenshot()

    reference_img = fake_img.reshape((5, 2, 4))
    reference_img = np.delete(reference_img, 3, 2)
    np.testing.assert_equal(reference_img, observation[0])
    self.assertEqual(observation[1], 123456789)

    # Close the console.
    console.close()


if __name__ == '__main__':
  absltest.main()
<file_sep># AndroidEnv - Tasks

With AndroidEnv we provide a mechanism for easily defining RL tasks for the
agent to learn. This includes various types of information such as what app/game
it should train on, what rewards the environment returns, or the start state
distribution and the episode end criteria.

## Task structure

A *task* definition is captured in the form of a `Task()` proto message. These
are most easily created by writing a `.textproto` file, then parsing it into a
proto message. In this section you can find a detailed description about the
types of information that make up a task, and an example demonstrating exactly
how to put these into code.

<details>
  <summary>Expand this tab to view the main types of information captured in these messages: </summary>

*   `id`: An ID used to identify the task.

*   `setup_steps`: These are steps the environment will perform right after
    launching the simulator. Possible steps include:

    *   `install_apk`: Installs an application from a specified path to the APK
        file.
    *   `start_activity`: Launches the requested app/activity.
    *   `rotate`: Sets the orientation of the device (landscape/portrait).

*   `reset_steps`: These are steps the environment will perform right at the
    beginning of a new RL episode. Possible steps include:

    *   `force_stop`: Stops a given app.
    *   `start_activity`: Launches the requested app/activity.
    *   `start_screen_pinning`: Restricts the agent's interaction to a
        particular activity through
        [screen pinning](https://support.google.com/android/answer/9455138?hl=en),
        meaning the agent will not be able to quit the given app.
    *   `clear_cache`: Clears the cache of a given app.

*   `success_conditions`: For each success condition defined, the environment
    will make sure that these conditions were met after finishing `setup_steps`
    and `reset_steps`. They might include conditions such as:

    *   `check_install`: Makes sure that the request app was successfully
        installed.
    *   `wait_for_app_screen`: Waits until the request app was successfully
        launched.

*   `expected_app_screen`: If this value is set to a particular activity, the
    environment will periodically check if the agent is still interacting with
    said activity, making sure it has not accidentally quit the application we
    want it to be training on.

*   `max_duration_sec`: Puts a time limit on the episodes, triggering an episode
    reset if the current episode has lasted too long.

*   `max_duration_steps`: Puts a step limit on the episodes, triggering an
    episode reset once the agent has reached the specified limit.

*   `log_parsing_config`: If the environment is parsing logcat messages, this
    field will determine what information it should listen for using regular
    expressions.

    *   `filters`: The environment filters log messages for these labels which
        signify that such messages were meant to be parsed by AndroidEnv.
    *   `log_regexps`: Once a log message was identified as relevant using the
        filters, the environment parses its contents using these regular
        expressions. For example, an application might be sending log messages
        of the form `reward: 1.0`, then the task will capture this info using
        the regexp `^[Rr]eward: ([-+]?[0-9]*\\.?[0-9]*)$`.

*   `extras_spec`: Determines the type and shape of extras exposed by the task.
    Extras are usually parsed from logcat messages.

</details>

<details>
  <summary>Expand this tab to see what an example `.textproto` file might look like in practice:</summary>

```python
id: "classic_2048"
name: "Classic 2048 - Default"
description: "Slide numbered tiles on a grid to combine them to create a tile with the number 2048"
package_name: "com.tpcstld.twozerogame"
full_activity_name: "com.tpcstld.twozerogame/com.tpcstld.twozerogame.MainActivity"

# Perform these upon launching the environment
setup_steps: [
  {
    # Install the 2048 app
    adb_call: {
      install_apk: {
        filesystem: {
          path: path/to/classic_2048.apk
        }
      }
    }
    # Check if it was installed correctly
    success_condition: {
      check_install: {
        package_name: "com.tpcstld.twozerogame"
        timeout_sec: 10.0
      }
    }
  },
  # Orient the screen in portait mode
  { adb_call: { rotate: { orientation: PORTRAIT_0 } } }
]

# Perform these upon episode resets
reset_steps: [

  # Stop the 2048 app
  { adb_call: { force_stop: { package_name: "com.tpcstld.twozerogame" } } },
  { adb_call: { clear_cache: { package_name: "com.tpcstld.twozerogame" } } },

  # Start the 2048 app
  {
    adb_call: {
      start_activity: {
        full_activity: "com.tpcstld.twozerogame/com.tpcstld.twozerogame.MainActivity"
        extra_args: [
            "--ez", '"RL_TASK_ENABLED"', '"true"',
            "--es", '"RL_TASK_GAME_CONFIG"', '"{}"'
        ]
      }
    }

    # Wait until the app has launched successfully
    success_condition: {
      wait_for_app_screen: {
        app_screen: {
          activity: "com.tpcstld.twozerogame/com.tpcstld.twozerogame.MainActivity"
          view_hierarchy_path: [
          ]
        }
        timeout_sec: 10.0
      }
      num_retries: 10
    }
  },

  # Make sure the agent cannot quit the 2048 app
  {
    adb_call: {
      start_screen_pinning: {
        full_activity: "com.tpcstld.twozerogame/com.tpcstld.twozerogame.MainActivity"
      }
    }
  }
]

# Periodically check if the agent has accidentally quit the app
expected_app_screen: {
  activity: "com.tpcstld.twozerogame/com.tpcstld.twozerogame.MainActivity"
  view_hierarchy_path: []
}

max_num_steps: 500

# Capture expected format of log messages
log_parsing_config: {
  filters: ["AndroidRLTask:V"]
  log_regexps: {
    score: "^[Ss]core: ([-+]?[0-9]*\\.?[0-9]*)$"
    reward: "^[Rr]eward: ([-+]?[0-9]*\\.?[0-9]*)$"
    episode_end: "^episode[ _]end$"
    extra: "^extra: (?P<name>[^ ]*)[ ]?(?P<extra>.*)$"
    json_extra: "^json_extra: (?P<json_extra>.*)$"
  }
}

# Capture expected shape and type of extras
extras_spec: [
  # Grid representing the state of the board.
  { name: "grid" shape: [4, 4], dtype: INT32},
  # Direction of the last swipe action that prompted that change in the state.
  # 0: up, 1: right, 2: down, 3: left
  { name: "direction" shape: [1], dtype: INT32 }
]

```

</details>

## Log messages and custom APKs

You might have noticed that tasks often rely on log messages exposed by the
Android system, which AndroidEnv can intercept and translate into items such as
rewards, episode end signals or task extras.

One way to define rewards is by using
`log_parsing_config.LogRegexps.RewardEvent` messages in the task proto. These
consist of a regular expression and a numeric value indicating the intended
reward. If the regexp is matched in any of the lines of the logcat stream, the
agent will receive the given reward. It is also possible to have multiple of
these RewardEvents, allowing us to give rewards for different log messages. The
same applies for episode end signals: logcat messages that match the regexps
defined in `log_parsing_config.LogRegexps.episode_end` will trigger an episode
reset.

Of course, applications might not send suitable messages by default, so in order
to have access to such messages, we often add them to the apps' source code to
match our expectations. For example, in the case of the 2048 app, we find in the
game's source code the exact lines where the score is computed, and add a line
to log this value in the format that is expected by the textproto (or
conversely, make sure the textproto matches the format you specified here). For
example:

```java
// Make sure thet LOG_FILTER matches 'filters' in the textproto
public static final String LOG_FILTER = "AndroidRLTask";

// Make sure that the corresponding part of 'log_regexps' will match this string
Log.i(LOG_FILTER, String.format(Locale.ENGLISH, "reward: %r", reward_value))
```

You can take a look at example APKs extended with log messages in the example
tasks (see the section below).

## Example tasks

Along with the environment implementation we provide a set of example task
definitions. These were chosen so that they would demonstrate the large variety
of different challenges (e.g. app navigtion, puzzle games, time-reactive games,
adventure games, card games...) and corresponding interfaces (e.g. button
pressing, swiping, drag-and-drop...) available in AndroidEnv. You can find a
list and detailed description of each of these tasks in
[example_tasks.md](example_tasks.md).
<file_sep># coding=utf-8
# Copyright 2021 DeepMind Technologies Limited.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

"""Wraps the AndroidEnv environment to rescale the observations."""

from typing import Optional, Sequence, Dict

from android_env.wrappers import base_wrapper
import dm_env
from dm_env import specs
import numpy as np
from PIL import Image


# Taken from https://pillow.readthedocs.io/en/3.2.x/reference/Image.html#PIL.Image.Image.convert
#
# This array maps an RGB image to a grayscale image using the ITU-R 709
# specification which is good for computer displays and HDTV.
RGB_TO_GRAYSCALE_COEFFICIENTS = [0.2126, 0.7152, 0.0722]


class ImageRescaleWrapper(base_wrapper.BaseWrapper):
  """AndroidEnv with rescaled observations."""

  def __init__(
      self,
      env: dm_env.Environment,
      zoom_factors: Optional[Sequence[float]] = (0.5, 0.5),
      grayscale: bool = False):
    super().__init__(env)
    assert 'pixels' in self._env.observation_spec()
    assert self._env.observation_spec()['pixels'].shape[-1] in [1, 3], (
        'Number of pixel channels should be 1 or 3.')
    self._grayscale = grayscale
    if zoom_factors is None:
      zoom_factors = (1.0, 1.0)
    # We only zoom the width and height of each layer, and we explicitly do not
    # want to zoom the number of channels so we just multiply it by 1.0.
    self._zoom_factors = tuple(zoom_factors) + (1.0,)

    # Save the raw image for making videos, for example.
    self._raw_pixels = None

  @property
  def raw_pixels(self):
    return self._raw_pixels

  def _process_timestep(self, timestep: dm_env.TimeStep) -> dm_env.TimeStep:
    observation = timestep.observation
    self._raw_pixels = observation['pixels'].copy()
    processed_observation = observation.copy()
    processed_observation['pixels'] = self._process_pixels(
        observation['pixels'])
    return timestep._replace(observation=processed_observation)

  def _process_pixels(self, raw_observation: np.ndarray) -> np.ndarray:
    # We expect `raw_observation` to have shape (W, H, 3) - 3 for RGB
    new_shape = np.array(
        self._zoom_factors[0:2] * np.array(raw_observation.shape[0:2]),
        dtype=np.int)[::-1]
    if self._grayscale:
      # When self._grayscale == True, we squash the RGB into a single layer
      image = np.dot(raw_observation, RGB_TO_GRAYSCALE_COEFFICIENTS)
    else:
      image = raw_observation
    return self._resize_image_array(image, new_shape)

  def _resize_image_array(
      self,
      grayscale_or_rbg_array: np.ndarray,
      new_shape: Sequence[int]) -> np.ndarray:
    """Resize color or grayscale/action_layer array to new_shape."""
    assert np.array(new_shape).ndim == 1
    assert len(new_shape) == 2
    resized_array = np.array(Image.fromarray(
        grayscale_or_rbg_array.astype('uint8')).resize(new_shape))
    if resized_array.ndim == 2:
      return np.expand_dims(resized_array, axis=-1)
    return resized_array

  def reset(self) -> dm_env.TimeStep:
    timestep = self._env.reset()
    return self._process_timestep(timestep)

  def step(self, action) -> dm_env.TimeStep:
    timestep = self._env.step(action)
    return self._process_timestep(timestep)

  def observation_spec(self) -> Dict[str, specs.Array]:
    parent_spec = self._env.observation_spec().copy()
    out_shape = np.multiply(parent_spec['pixels'].shape,
                            self._zoom_factors).astype(np.int32)
    if self._grayscale:
      # In grayscale mode we want the output shape to be [W, H, 1]
      out_shape[-1] = 1
    parent_spec['pixels'] = specs.Array(
        shape=out_shape,
        dtype=parent_spec['pixels'].dtype,
        name=parent_spec['pixels'].name)
    return parent_spec
<file_sep># coding=utf-8
# Copyright 2021 DeepMind Technologies Limited.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

"""Tests for android_env.components.dumpsys_thread."""

from absl.testing import absltest
from android_env.components import app_screen_checker as screen_checker
from android_env.components import dumpsys_thread

import mock


class DumpsysThreadTest(absltest.TestCase):

  def setUp(self):
    super().setUp()
    self._dumpsys_thread = dumpsys_thread.DumpsysThread(
        app_screen_checker=mock.create_autospec(
            screen_checker.AppScreenChecker),
        check_frequency=1,
        max_failed_current_activity=5,
        block_input=True,
        block_output=True)

  def test_unexpected_activity(self):
    outcome = screen_checker.AppScreenChecker.Outcome.UNEXPECTED_ACTIVITY
    self._dumpsys_thread._app_screen_checker.matches_current_app_screen.return_value = outcome
    self._dumpsys_thread.write(
        dumpsys_thread.DumpsysThread.Signal.FETCH_DUMPSYS)
    v = self._dumpsys_thread.read(block=True)
    expected = dumpsys_thread.DumpsysThread.Signal.USER_EXITED_ACTIVITY
    self.assertEqual(expected, v)

  def test_unexpected_view_hierarchy(self):
    outcome = screen_checker.AppScreenChecker.Outcome.UNEXPECTED_VIEW_HIERARCHY
    self._dumpsys_thread._app_screen_checker.matches_current_app_screen.return_value = outcome
    self._dumpsys_thread.write(
        dumpsys_thread.DumpsysThread.Signal.FETCH_DUMPSYS)
    v = self._dumpsys_thread.read(block=True)
    expected = dumpsys_thread.DumpsysThread.Signal.USER_EXITED_VIEW_HIERARCHY
    self.assertEqual(expected, v)

  def test_success(self):
    outcome = screen_checker.AppScreenChecker.Outcome.SUCCESS
    self._dumpsys_thread._app_screen_checker.matches_current_app_screen.return_value = outcome
    self._dumpsys_thread.write(
        dumpsys_thread.DumpsysThread.Signal.FETCH_DUMPSYS)
    v = self._dumpsys_thread.read(block=True)
    expected = dumpsys_thread.DumpsysThread.Signal.OK
    self.assertEqual(expected, v)

  def test_not_requested(self):
    self._dumpsys_thread.write('wrong_signal')
    v = self._dumpsys_thread.read(block=True)
    expected = dumpsys_thread.DumpsysThread.Signal.DID_NOT_CHECK
    self.assertEqual(expected, v)

  def test_skipped(self):
    outcome = screen_checker.AppScreenChecker.Outcome.SUCCESS
    self._dumpsys_thread._app_screen_checker.matches_current_app_screen.return_value = outcome
    self._dumpsys_thread._check_frequency = 5
    self._dumpsys_thread._main_loop_counter = 0

    for _ in range(4):
      self._dumpsys_thread.write(
          dumpsys_thread.DumpsysThread.Signal.FETCH_DUMPSYS)
      v = self._dumpsys_thread.read(block=True)
      expected = dumpsys_thread.DumpsysThread.Signal.DID_NOT_CHECK
      self.assertEqual(expected, v)

    self._dumpsys_thread.write(
        dumpsys_thread.DumpsysThread.Signal.FETCH_DUMPSYS)
    v = self._dumpsys_thread.read(block=True)
    expected = dumpsys_thread.DumpsysThread.Signal.OK
    self.assertEqual(expected, v)


if __name__ == '__main__':
  absltest.main()
<file_sep># coding=utf-8
# Copyright 2021 DeepMind Technologies Limited.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

"""Base class for AndroidEnv wrappers."""

from typing import Any, Dict
import dm_env
from dm_env import specs


class BaseWrapper(dm_env.Environment):
  """AndroidEnv wrapper."""

  def __init__(self, env):
    self._env = env

  def reset(self) -> dm_env.TimeStep:
    self._reset_state()
    timestep = self._process_timestep(self._env.reset())
    return timestep

  def step(self, action: Any) -> dm_env.TimeStep:
    action = self._process_action(action)
    return self._process_timestep(self._env.step(action))

  def _reset_state(self):
    pass

  def _process_action(self, action: Any) -> Any:
    return action

  def _process_timestep(self, timestep: dm_env.TimeStep) -> dm_env.TimeStep:
    return timestep

  def observation_spec(self) -> Dict[str, specs.Array]:
    return self._env.observation_spec()

  def action_spec(self) -> Dict[str, specs.Array]:
    return self._env.action_spec()

  def _wrapper_logs(self) -> Dict[str, Any]:
    """Add wrapper specific logging here."""
    return {}

  def android_logs(self) -> Dict[str, Any]:
    info = self._env.android_logs()
    info.update(self._wrapper_logs())
    return info

  @property
  def raw_env(self):
    """Recursively unwrap until we reach the true 'raw' env."""
    wrapped = self._env
    if hasattr(wrapped, 'raw_env'):
      return wrapped.raw_env
    return wrapped

  def __getattr__(self, attr):
    """Delegate attribute access to underlying environment."""
    return getattr(self._env, attr)

  def close(self):
    self._env.close()
<file_sep># coding=utf-8
# Copyright 2021 DeepMind Technologies Limited.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

"""Unit tests for AndroidEnv."""

from absl.testing import absltest
from android_env import environment
from android_env.components import coordinator as coordinator_lib
import dm_env
import mock
import numpy as np


class AndroidEnvTest(absltest.TestCase):

  def test_specs(self):
    coordinator = mock.create_autospec(coordinator_lib.Coordinator)
    coordinator.action_spec.return_value = {
        'action_type':
            dm_env.specs.DiscreteArray(num_values=3),
        'touch_position':
            dm_env.specs.BoundedArray(
                shape=(2,), dtype=np.float32, minimum=0.0, maximum=1.0),
    }
    coordinator.observation_spec.return_value = {
        'pixels': dm_env.specs.Array(shape=(123, 456, 3), dtype=np.uint8),
        'timedelta': dm_env.specs.Array(shape=(), dtype=np.int64),
        'orientation': dm_env.specs.Array(shape=(4,), dtype=np.uint8),
    }
    coordinator.task_extras_spec.return_value = {
        'click': dm_env.specs.Array(shape=(), dtype=np.int64),
    }
    env = environment.AndroidEnv(coordinator)

    # Check action spec.
    self.assertNotEmpty(env.action_spec())
    self.assertIn('action_type', env.action_spec())
    self.assertIsInstance(env.action_spec()['action_type'],
                          dm_env.specs.DiscreteArray)
    self.assertIn('touch_position', env.action_spec())
    self.assertIsInstance(env.action_spec()['touch_position'],
                          dm_env.specs.BoundedArray)

    # Check observation spec.
    self.assertNotEmpty(env.observation_spec())
    self.assertIn('pixels', env.observation_spec())
    self.assertIsInstance(env.observation_spec()['pixels'], dm_env.specs.Array)
    # The `pixels` entry in the observation spec should match the screen size of
    # the simulator with three color channels (RGB).
    self.assertEqual(env.observation_spec()['pixels'].shape, (123, 456, 3))
    self.assertIn('timedelta', env.observation_spec())
    self.assertIsInstance(env.observation_spec()['timedelta'],
                          dm_env.specs.Array)
    # The `timedelta` should be a scalar.
    self.assertEqual(env.observation_spec()['timedelta'].shape, ())
    self.assertIn('orientation', env.observation_spec())
    # The `orientation` should be a one-hot vector with four dimensions.
    self.assertIsInstance(env.observation_spec()['orientation'],
                          dm_env.specs.Array)
    self.assertEqual(env.observation_spec()['orientation'].shape, (4,))

    # Check extras spec.
    self.assertNotEmpty(env.task_extras_spec())
    self.assertIn('click', env.task_extras_spec())
    self.assertEqual(env.task_extras_spec()['click'].shape, ())
    self.assertEqual(env.task_extras_spec()['click'].dtype, np.int64)

  def test_reset_and_step(self):
    coordinator = mock.create_autospec(coordinator_lib.Coordinator)
    coordinator.action_spec.return_value = {
        'action_type':
            dm_env.specs.DiscreteArray(num_values=3),
        'touch_position':
            dm_env.specs.BoundedArray(
                shape=(2,), dtype=np.float32, minimum=0.0, maximum=1.0),
    }
    coordinator.observation_spec.return_value = {
        'pixels': dm_env.specs.Array(shape=(123, 456, 3), dtype=np.uint8),
        'timedelta': dm_env.specs.Array(shape=(), dtype=np.int64),
        'orientation': dm_env.specs.Array(shape=(4,), dtype=np.uint8),
    }
    coordinator.task_extras_spec.return_value = {
        'click': dm_env.specs.Array(shape=(1,), dtype=np.int64),
    }
    env = environment.AndroidEnv(coordinator)
    coordinator.execute_action.return_value = (
        {
            'pixels': np.random.rand(987, 654, 3),
            'timedelta': 123456,
            'orientation': np.array((1, 0, 0, 0)),
        },  # Observation
        0.0,  # Reward.
        {
            'click': np.array([[246]], dtype=np.int64)
        },  # Task extras.
        False  # Episode ended?
    )

    ts = env.reset()
    self.assertIsInstance(ts, dm_env.TimeStep)
    # After a `reset()` the TimeStep should follow some expectations.
    self.assertTrue(ts.first())
    self.assertEqual(ts.reward, 0.0)
    self.assertEqual(ts.discount, 0.0)
    obs = ts.observation
    self.assertIn('pixels', obs)
    self.assertEqual(obs['pixels'].shape, (987, 654, 3))
    self.assertIn('timedelta', obs)
    self.assertEqual(obs['timedelta'], 123456)
    self.assertIn('orientation', obs)
    self.assertEqual(obs['orientation'].shape, (4,))
    np.testing.assert_equal(obs['orientation'], (1, 0, 0, 0))

    # Extras should also be provided.
    extras = env.task_extras()
    self.assertIn('click', extras)
    self.assertEqual(extras['click'], np.array([246], dtype=np.int64))

    coordinator.get_logs.return_value = {
        'my_measurement': 135,
    }

    # Step again in the environment and check expectations again.
    ts = env.step({'action_type': 1, 'touch_position': (10, 20)})
    self.assertIsInstance(ts, dm_env.TimeStep)
    # The StepType now should NOT be FIRST.
    self.assertFalse(ts.first())
    self.assertEqual(ts.reward, 0.0)
    self.assertEqual(ts.discount, 0.0)
    obs = ts.observation
    self.assertIn('pixels', obs)
    self.assertEqual(obs['pixels'].shape, (987, 654, 3))
    self.assertIn('timedelta', obs)
    self.assertEqual(obs['timedelta'], 123456)
    self.assertIn('orientation', obs)
    self.assertEqual(obs['orientation'].shape, (4,))
    np.testing.assert_equal(obs['orientation'], (1, 0, 0, 0))

    # Extras should still be provided.
    extras = env.task_extras()
    self.assertIn('click', extras)
    self.assertEqual(extras['click'], np.array([246], dtype=np.int64))

    # At this point these methods and properties should return something.
    self.assertNotEmpty(env.android_logs())
    self.assertNotEmpty(env.raw_observation)
    self.assertNotEmpty(env.raw_action)

if __name__ == '__main__':
  absltest.main()
<file_sep># coding=utf-8
# Copyright 2021 DeepMind Technologies Limited.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

"""Tests for loader."""

import builtins
import os
from absl.testing import absltest
from android_env import environment
from android_env import loader
from android_env.components import coordinator as coordinator_lib
from android_env.components import emulator_simulator
from android_env.components import task_manager as task_manager_lib
from android_env.proto import task_pb2
import mock


class LoaderTest(absltest.TestCase):

  @mock.patch.object(task_manager_lib, 'TaskManager', autospec=True)
  @mock.patch.object(emulator_simulator, 'EmulatorSimulator', autospec=True)
  @mock.patch.object(coordinator_lib, 'Coordinator', autospec=True)
  @mock.patch.object(builtins, 'open', autospec=True)
  def test_load(self, mock_open, coordinator, simulator, task_manager):

    mock_open.return_value.__enter__ = mock_open
    mock_open.return_value.read.return_value = ''

    env = loader.load(
        task_path='some/path/',
        avd_name='my_avd',
        android_avd_home='~/.android/avd',
        android_sdk_root='~/Android/Sdk',
        emulator_path='~/Android/Sdk/emulator/emulator',
        adb_path='~/Android/Sdk/platform-tools/adb',
        run_headless=False,
    )

    self.assertIsInstance(env, environment.AndroidEnv)
    simulator.assert_called_with(
        emulator_launcher_args=dict(
            avd_name='my_avd',
            android_avd_home=os.path.expanduser('~/.android/avd'),
            android_sdk_root=os.path.expanduser('~/Android/Sdk'),
            emulator_path=os.path.expanduser('~/Android/Sdk/emulator/emulator'),
            run_headless=False,
            gpu_mode='swiftshader_indirect',
            grpc_port=-1),
        emulator_console_args={},
        adb_path=os.path.expanduser('~/Android/Sdk/platform-tools/adb'),
        adb_server_port=5037,
        prompt_regex=r'\w*:\/ \$',
    )
    coordinator.assert_called_with(
        simulator.return_value,
        task_manager.return_value,
    )

  @mock.patch.object(task_manager_lib, 'TaskManager', autospec=True)
  @mock.patch.object(emulator_simulator, 'EmulatorSimulator', autospec=True)
  @mock.patch.object(coordinator_lib, 'Coordinator', autospec=True)
  @mock.patch.object(builtins, 'open', autospec=True)
  def test_task(self, mock_open, coordinator, simulator, task_manager):

    del coordinator, simulator
    mock_open.return_value.__enter__ = mock_open
    mock_open.return_value.read.return_value = r'''
id: "fake_task"
name: "Fake Task"
description: "Task for testing loader."
max_duration_sec: 0
'''

    env = loader.load(
        task_path='some/path/',
        avd_name='my_avd',
    )

    expected_task = task_pb2.Task()
    expected_task.id = 'fake_task'
    expected_task.name = 'Fake Task'
    expected_task.description = 'Task for testing loader.'
    expected_task.max_duration_sec = 0

    task_manager.assert_called_with(expected_task)
    assert isinstance(env, environment.AndroidEnv)


if __name__ == '__main__':
  absltest.main()
<file_sep># coding=utf-8
# Copyright 2021 DeepMind Technologies Limited.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

"""Android environment implementation."""

from typing import Any, Dict
from absl import logging
from android_env.components import coordinator as coordinator_lib
import dm_env
import numpy as np


class AndroidEnv(dm_env.Environment):
  """An RL environment that interacts with Android apps."""

  def __init__(self, coordinator: coordinator_lib.Coordinator):
    """Initializes the state of this AndroidEnv object."""

    self._coordinator = coordinator
    self._latest_action = {}
    self._latest_observation = {}
    self._latest_extras = {}
    self._reset_next_step = True

    logging.info('Action spec: %s', self.action_spec())
    logging.info('Observation spec: %s', self.observation_spec())
    logging.info('Task extras spec: %s', self.task_extras_spec())

  def action_spec(self) -> Dict[str, dm_env.specs.Array]:
    return self._coordinator.action_spec()

  def observation_spec(self) -> Dict[str, dm_env.specs.Array]:
    return self._coordinator.observation_spec()

  def task_extras_spec(self) -> Dict[str, dm_env.specs.Array]:
    return self._coordinator.task_extras_spec()

  @property
  def raw_action(self):
    return self._latest_action

  @property
  def raw_observation(self):
    return self._latest_observation

  def android_logs(self) -> Dict[str, Any]:
    return self._coordinator.get_logs()

  def reset(self) -> dm_env.TimeStep:
    """Resets the environment for a new RL episode."""

    logging.info('Resetting AndroidEnv...')

    # Reset state of the environment.
    self._coordinator.reset_environment_state()

    # Execute selected action (None when resetting).
    obs, _, extras, _ = self._coordinator.execute_action(action=None)

    # Process relevant information.
    if obs is not None:
      self._latest_observation = obs.copy()
    self._latest_extras = extras.copy()
    self._latest_action = {}
    self._reset_next_step = False

    logging.info('Done resetting AndroidEnv.')
    logging.info('************* NEW EPISODE *************')

    return dm_env.TimeStep(
        step_type=dm_env.StepType.FIRST,
        observation=self._latest_observation,
        reward=0.0,
        discount=0.0)

  def step(self, action: Dict[str, np.ndarray]) -> dm_env.TimeStep:
    """Takes a step in the environment."""

    # Check if it's time to reset the episode.
    if self._reset_next_step:
      return self.reset()

    # Execute selected action.
    obs, reward, extras, episode_end = self._coordinator.execute_action(action)

    # Process relevant information.
    if obs is not None:
      self._latest_observation = obs.copy()
    self._latest_extras = extras.copy()
    self._latest_action = action.copy()
    self._reset_next_step = episode_end

    # Return timestep with reward and observation just computed.
    if episode_end:
      return dm_env.termination(
          observation=self._latest_observation, reward=reward)
    else:
      return dm_env.transition(
          observation=self._latest_observation, reward=reward, discount=0.0)

  def task_extras(self, latest_only: bool = True) -> Dict[str, np.ndarray]:
    """Returns latest task extras."""

    task_extras = {}
    for key, spec in self.task_extras_spec().items():
      if key in self._latest_extras:
        extra_values = self._latest_extras[key].astype(spec.dtype)
        for extra in extra_values:
          spec.validate(extra)
        task_extras[key] = extra_values[-1] if latest_only else extra_values
    return task_extras

  def close(self) -> None:
    """Cleans up running processes, threads and local files."""
    logging.info('Cleaning up AndroidEnv...')
    if hasattr(self, '_coordinator'):
      self._coordinator.close()
    logging.info('Done cleaning up AndroidEnv.')

  def __del__(self) -> None:
    self.close()
