| 7 | |
| 8 | |
| 9 | class Interpreter: |
| 10 | def __init__(self, status_queue: Queue): |
| 11 | # MP Queue to put current status of execution in while processes commands. |
| 12 | # It helps us reflect the current status on the UI. |
| 13 | self.status_queue = status_queue |
| 14 | |
| 15 | def process_commands(self, json_commands: list[dict[str, Any]]) -> bool: |
| 16 | """ |
| 17 | Reads a list of JSON commands and runs the corresponding function call as specified in context.txt |
| 18 | :param json_commands: List of JSON Objects with format as described in context.txt |
| 19 | :return: True for successful execution, False for exception while interpreting or executing. |
| 20 | """ |
| 21 | for command in json_commands: |
| 22 | success = self.process_command(command) |
| 23 | if not success: |
| 24 | return False # End early and return |
| 25 | return True |
| 26 | |
| 27 | def process_command(self, json_command: dict[str, Any]) -> bool: |
| 28 | """ |
| 29 | Reads the passed in JSON object and extracts relevant details. Format is specified in context.txt. |
| 30 | After interpretation, it proceeds to execute the appropriate function call. |
| 31 | |
| 32 | :return: True for successful execution, False for exception while interpreting or executing. |
| 33 | """ |
| 34 | function_name = json_command['function'] |
| 35 | parameters = json_command.get('parameters', {}) |
| 36 | human_readable_justification = json_command.get('human_readable_justification') |
| 37 | print(f'Now performing - {function_name} - {parameters} - {human_readable_justification}') |
| 38 | self.status_queue.put(human_readable_justification) |
| 39 | try: |
| 40 | self.execute_function(function_name, parameters) |
| 41 | return True |
| 42 | except Exception as e: |
| 43 | print(f'\nError:\nWe are having a problem executing this step - {type(e)} - {e}') |
| 44 | print(f'This was the json we received from the LLM: {json.dumps(json_command, indent=2)}') |
| 45 | print(f'This is what we extracted:') |
| 46 | print(f'\t function_name:{function_name}') |
| 47 | print(f'\t parameters:{parameters}') |
| 48 | |
| 49 | return False |
| 50 | |
| 51 | def execute_function(self, function_name: str, parameters: dict[str, Any]) -> None: |
| 52 | """ |
| 53 | We are expecting only two types of function calls below |
| 54 | 1. time.sleep() - to wait for web pages, applications, and other things to load. |
| 55 | 2. pyautogui calls to interact with system's mouse and keyboard. |
| 56 | """ |
| 57 | # Sometimes pyautogui needs warming up i.e. sometimes first call isn't executed hence padding a random call here |
| 58 | pyautogui.press("command", interval=0.2) |
| 59 | |
| 60 | if function_name == "sleep" and parameters.get("secs"): |
| 61 | sleep(parameters.get("secs")) |
| 62 | elif hasattr(pyautogui, function_name): |
| 63 | # Execute the corresponding pyautogui function i.e. Keyboard or Mouse commands. |
| 64 | function_to_call = getattr(pyautogui, function_name) |
| 65 | |
| 66 | # Special handling for the 'write' function |