Merge pull request #31 from mengqianX/LLMPolicy

LLm policy

kea/device_state.py

@@ -10,7 +10,7 @@
     from .device import Device
 
 from .utils import md5, deprecated
-from .input_event import SearchEvent, SetTextAndSearchEvent, TouchEvent, LongTouchEvent, ScrollEvent, SetTextEvent, KeyEvent
+from .input_event import SearchEvent, SetTextAndSearchEvent, TouchEvent, LongTouchEvent, ScrollEvent, SetTextEvent, KeyEvent, UIEvent
 
 
 class DeviceState(object):
@@ -805,3 +805,108 @@ def is_view_exist(self, view_dict):
             ):
                 return view
         return None
+
+    def get_state_screen(self): 
+        return self.screenshot_path
+
+    def get_view_desc(self, view):
+        content_description = self.__safe_dict_get(view, 'content_description', default='')
+        view_text = self.__safe_dict_get(view, 'text', default='')
+        scrollable = self.__safe_dict_get(view, 'scrollable')
+        view_desc = f'view'
+        if scrollable:
+            view_desc = f'scrollable view'
+        if content_description:
+            view_desc += f' "{content_description}"'
+        if view_text:
+            view_text = view_text.replace('\n', '  ')
+            view_text = f'{view_text[:20]}...' if len(view_text) > 20 else view_text
+            view_desc += f' with text "{view_text}"'
+        return view_desc
+
+    def get_described_actions(self):
+        """
+        Get a text description of current state
+        """
+        enabled_view_ids = []
+        for view_dict in self.views:
+            # exclude navigation bar if exists
+            if self.__safe_dict_get(view_dict, 'visible') and \
+                self.__safe_dict_get(view_dict, 'resource_id') not in \
+               ['android:id/navigationBarBackground',
+                'android:id/statusBarBackground']:
+                enabled_view_ids.append(view_dict['temp_id'])
+
+        view_descs = []
+        available_actions = []
+        for view_id in enabled_view_ids:
+            view = self.views[view_id]
+            clickable = self._get_self_ancestors_property(view, 'clickable')
+            scrollable = self.__safe_dict_get(view, 'scrollable')
+            checkable = self._get_self_ancestors_property(view, 'checkable')
+            long_clickable = self._get_self_ancestors_property(view, 'long_clickable')
+            editable = self.__safe_dict_get(view, 'editable')
+            actionable = clickable or scrollable or checkable or long_clickable or editable
+            checked = self.__safe_dict_get(view, 'checked')
+            selected = self.__safe_dict_get(view, 'selected')
+            content_description = self.__safe_dict_get(view, 'content_description', default='')
+            view_text = self.__safe_dict_get(view, 'text', default='')
+            if not content_description and not view_text and not scrollable:  # actionable?
+                continue
+
+            view_status = ''
+            if editable:
+                view_status += 'editable '
+            if checked or selected:
+                view_status += 'checked '
+            view_desc = f'- a {view_status}view'
+            if content_description:
+                content_description = content_description.replace('\n', '  ')
+                content_description = f'{content_description[:20]}...' if len(content_description) > 20 else content_description
+                view_desc += f' "{content_description}"'
+            if view_text:
+                view_text = view_text.replace('\n', '  ')
+                view_text = f'{view_text[:20]}...' if len(view_text) > 20 else view_text
+                view_desc += f' with text "{view_text}"'
+            if actionable:
+                view_actions = []
+                if editable:
+                    view_actions.append(f'edit ({len(available_actions)})')
+                    available_actions.append(SetTextEvent(view=view, text='HelloWorld'))
+                if clickable or checkable:
+                    view_actions.append(f'click ({len(available_actions)})')
+                    available_actions.append(TouchEvent(view=view))
+                # if checkable:
+                #     view_actions.append(f'check/uncheck ({len(available_actions)})')
+                #     available_actions.append(TouchEvent(view=view))
+                # if long_clickable:
+                #     view_actions.append(f'long click ({len(available_actions)})')
+                #     available_actions.append(LongTouchEvent(view=view))
+                if scrollable:
+                    view_actions.append(f'scroll up ({len(available_actions)})')
+                    available_actions.append(ScrollEvent(view=view, direction='UP'))
+                    view_actions.append(f'scroll down ({len(available_actions)})')
+                    available_actions.append(ScrollEvent(view=view, direction='DOWN'))
+                view_actions_str = ', '.join(view_actions)
+                view_desc += f' that can {view_actions_str}'
+            view_descs.append(view_desc)
+        view_descs.append(f'- a key to go back ({len(available_actions)})')
+        available_actions.append(KeyEvent(name='BACK'))
+        state_desc = 'The current state has the following UI views and corresponding actions, with action id in parentheses:\n '
+        state_desc += ';\n '.join(view_descs)
+        return state_desc, available_actions
+
+    def get_action_desc(self, action):
+        desc = action.event_type
+        if isinstance(action, KeyEvent):
+            desc = f'- go {action.name.lower()}'
+        if isinstance(action, UIEvent):
+            action_name = action.event_type
+            if isinstance(action, LongTouchEvent):
+                action_name = 'long click'
+            elif isinstance(action, SetTextEvent):
+                action_name = f'enter "{action.text}" into'
+            elif isinstance(action, ScrollEvent):
+                action_name = f'scroll {action.direction.lower()}'
+            desc = f'- {action_name} {self.get_view_desc(action.view)}'
+        return desc

kea/input_manager.py

@@ -2,6 +2,8 @@
 import logging
 import time
 
+from .similarity import Similarity
+
 from .input_event import EventLog
 from .input_policy import (
     GuidedPolicy,
@@ -10,6 +12,8 @@
     KeaInputPolicy,
     RandomPolicy,
     POLICY_NONE,
+    POLICY_LLM,
+    LLMPolicy
 )
 
 DEFAULT_POLICY = POLICY_RANDOM
@@ -18,6 +22,7 @@
 DEFAULT_EVENT_COUNT = 100000000
 DEFAULT_TIMEOUT = 3600
 DEFAULT_DEVICE_SERIAL = "emulator-5554"
+DEFAULT_UI_TARPIT_NUM = 2
 
 class UnknownInputException(Exception):
     pass
@@ -80,6 +85,7 @@ def __init__(
         self.number_of_events_that_restart_app = number_of_events_that_restart_app
         self.generate_utg = generate_utg
         self.policy = self.get_input_policy(device, app, master)
+        self.sim_calculator = Similarity(DEFAULT_UI_TARPIT_NUM)
 
     def get_input_policy(self, device, app, master):
         if self.policy_name == POLICY_NONE:
@@ -94,6 +100,8 @@ def get_input_policy(self, device, app, master):
             )
         elif self.policy_name == POLICY_RANDOM:
             input_policy = RandomPolicy(device, app, random_input=self.random_input, kea=self.kea, number_of_events_that_restart_app = self.number_of_events_that_restart_app, clear_and_restart_app_data_after_100_events=True, generate_utg = self.generate_utg)
+        elif self.policy_name == POLICY_LLM:
+            input_policy = LLMPolicy(device, app, random_input=self.random_input, kea=self.kea, number_of_events_that_restart_app = self.number_of_events_that_restart_app, clear_and_restart_app_data_after_100_events=True, generate_utg = self.generate_utg)
         else:
             self.logger.warning(
                 "No valid input policy specified. Using policy \"none\"."