File size: 4,048 Bytes
428252c 7d04631 428252c 93b2a9f 428252c 93b2a9f 428252c 93b2a9f 428252c 93b2a9f 428252c 93b2a9f 428252c 93b2a9f 428252c |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 |
import os
import time
from typing import Dict, Any
import subprocess
from flow_modules.aiflows.InterpreterFlowModule import InterpreterAtomicFlow
class ExecuteCodeAtomicFlow(InterpreterAtomicFlow):
"""This class inherits from InterpreterAtomicFlow and is used to execute code in a file.
It opens up the file in VSCode and waits for the user to save and close the file. Once the file is saved and closed, it reads the code
from the file and executes it. It then returns the output of the execution.
*Input Interface*:
- temp_code_file_location: The location of the file containing the code to be executed.
- language: The language of the code to be executed.
*Output Interface*:
- interpreter_output: The output of the execution of the code.
- code_ran: The code that was executed.
*Configuration Parameters*:
- `input_interface`: The input interface of the atomic flow.
- `output_interface`: The output interface of the atomic flow.
"""
def _prepare_code(self, input_data: Dict[str, Any]):
"""
This method reads the code from the file and stores it in the input_data dictionary.
:param input_data: The input_data dictionary.
:type input_data: Dict[str, Any]
:return: None
"""
file_location = input_data["temp_code_file_location"]
start_marker = "# Code:\n"
end_marker = "############"
code_started = False
code_str = ""
with open(file_location, 'r') as file:
for line in file:
if line.strip() == start_marker.strip():
code_started = True
continue
if line.strip() == end_marker.strip():
break
if code_started:
code_str += line
input_data["code"] = code_str
def _check_input(self, input_data: Dict[str, Any]):
"""
This method checks if the input_data dictionary contains the required keys.
:param input_data: The input_data dictionary.
:type input_data: Dict[str, Any]
:raises AssertionError: If the input_data dictionary does not contain the required keys.
:return: None
"""
assert "temp_code_file_location" in input_data, "temp_code_file_location not passed to ExecuteCodeAtomicFlow"
assert "language" in input_data, "language not passed to ExecuteCodeAtomicFlow"
def _delete_file(self, file_location):
"""
This method deletes the file at the given location.
:param file_location: The location of the file to be deleted.
:type file_location: str
:return: None
"""
if os.path.exists(file_location):
os.remove(file_location)
def _open_file_and_wait_for_upd(self, file_location):
"""
This method opens the file at the given location in VSCode and waits for the user to save the file.
:param file_location: The location of the file to be opened.
:type file_location: str
:return: None
"""
process = subprocess.Popen(["code", "--wait", file_location])
while True:
if process.poll() is not None:
break
time.sleep(1)
def run(
self,
input_data: Dict[str, Any]):
"""
This method is called when the atomic flow is called.
:param input_data: The input_data dictionary.
:type input_data: Dict[str, Any]
:return: The output of the execution of the code.
:rtype: Dict[str, Any]
"""
self._check_input(input_data)
file_loc = input_data["temp_code_file_location"]
self._open_file_and_wait_for_upd(file_loc)
self._prepare_code(input_data)
self._process_input_data(input_data)
execution_output = self._call()
self._delete_file(file_loc)
response = {"interpreter_output": execution_output, "code_ran": input_data['code']}
return response
|