Update CodeFileEditAtomicFlow.py
Browse files- CodeFileEditAtomicFlow.py +29 -0
CodeFileEditAtomicFlow.py
CHANGED
|
@@ -19,6 +19,11 @@ class CodeFileEditAtomicFlow(AtomicFlow):
|
|
| 19 |
- `temp_code_file_location`: str
|
| 20 |
"""
|
| 21 |
def _generate_content(self, code_lib_location, code_str) -> str:
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 22 |
content = (
|
| 23 |
"# The below code will be appended to " +
|
| 24 |
code_lib_location + "\n"
|
|
@@ -33,11 +38,22 @@ class CodeFileEditAtomicFlow(AtomicFlow):
|
|
| 33 |
return content
|
| 34 |
|
| 35 |
def _generate_temp_file_location(self, code_lib_location):
|
|
|
|
|
|
|
|
|
|
|
|
|
| 36 |
directory = os.path.dirname(code_lib_location)
|
| 37 |
ret = os.path.join(directory, 'temp.py')
|
| 38 |
return ret
|
| 39 |
|
| 40 |
def _write_code_content_to_file(self, file_location, content: str):
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 41 |
try:
|
| 42 |
with open(file_location, "w") as file:
|
| 43 |
file.write(content)
|
|
@@ -48,6 +64,10 @@ class CodeFileEditAtomicFlow(AtomicFlow):
|
|
| 48 |
return False, str(e), file_location
|
| 49 |
|
| 50 |
def _check_input(self, input_data: Dict[str, Any]):
|
|
|
|
|
|
|
|
|
|
|
|
|
| 51 |
assert "code" in input_data, "code is not passed to CodeFileEditAtomicFlow"
|
| 52 |
assert "language_of_code" in input_data, "language_of_code is not passed to CodeFileEditAtomicFlow"
|
| 53 |
assert "memory_files" in input_data, "memory_files is not passed to CodeFileEditAtomicFlow"
|
|
@@ -59,6 +79,11 @@ class CodeFileEditAtomicFlow(AtomicFlow):
|
|
| 59 |
assert language_of_code.lower() == 'python', "sorry!! only writing python code is supported."
|
| 60 |
|
| 61 |
def _generate_input_to_writer(self, input_data: Dict[str, Any]):
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 62 |
code_str = input_data['code']
|
| 63 |
code_lib_location = input_data["memory_files"]["code_library"]
|
| 64 |
content_to_write = self._generate_content(code_lib_location, code_str)
|
|
@@ -69,6 +94,10 @@ class CodeFileEditAtomicFlow(AtomicFlow):
|
|
| 69 |
self,
|
| 70 |
input_data: Dict[str, Any]
|
| 71 |
):
|
|
|
|
|
|
|
|
|
|
|
|
|
| 72 |
self._check_input(input_data)
|
| 73 |
|
| 74 |
# ~~~ Getting input data to the file editor ~~~
|
|
|
|
| 19 |
- `temp_code_file_location`: str
|
| 20 |
"""
|
| 21 |
def _generate_content(self, code_lib_location, code_str) -> str:
|
| 22 |
+
"""This function generates the content to be written to the temp file.
|
| 23 |
+
:param code_lib_location (str): The location of the code library file.
|
| 24 |
+
:param code_str (str): The code to be written to the temp file.
|
| 25 |
+
:return: (str) The content to be written to the temp file.
|
| 26 |
+
"""
|
| 27 |
content = (
|
| 28 |
"# The below code will be appended to " +
|
| 29 |
code_lib_location + "\n"
|
|
|
|
| 38 |
return content
|
| 39 |
|
| 40 |
def _generate_temp_file_location(self, code_lib_location):
|
| 41 |
+
"""This function generates the location of the temp file.
|
| 42 |
+
:param code_lib_location (str): The location of the code library file.
|
| 43 |
+
:return: (str) The location of the temp file.
|
| 44 |
+
"""
|
| 45 |
directory = os.path.dirname(code_lib_location)
|
| 46 |
ret = os.path.join(directory, 'temp.py')
|
| 47 |
return ret
|
| 48 |
|
| 49 |
def _write_code_content_to_file(self, file_location, content: str):
|
| 50 |
+
"""This function writes the content to the temp file.
|
| 51 |
+
:param file_location (str): The location of the temp file.
|
| 52 |
+
:param content (str): The content to be written to the temp file.
|
| 53 |
+
:return: (bool, str, str) The first value is a boolean indicating if the write was successful.
|
| 54 |
+
The second value is a string containing the output of the write operation.
|
| 55 |
+
The third value is a string containing the location of the temp file.
|
| 56 |
+
"""
|
| 57 |
try:
|
| 58 |
with open(file_location, "w") as file:
|
| 59 |
file.write(content)
|
|
|
|
| 64 |
return False, str(e), file_location
|
| 65 |
|
| 66 |
def _check_input(self, input_data: Dict[str, Any]):
|
| 67 |
+
"""This function checks if the input data is valid.
|
| 68 |
+
:param input_data (Dict[str, Any]): The input data.
|
| 69 |
+
:return: None
|
| 70 |
+
"""
|
| 71 |
assert "code" in input_data, "code is not passed to CodeFileEditAtomicFlow"
|
| 72 |
assert "language_of_code" in input_data, "language_of_code is not passed to CodeFileEditAtomicFlow"
|
| 73 |
assert "memory_files" in input_data, "memory_files is not passed to CodeFileEditAtomicFlow"
|
|
|
|
| 79 |
assert language_of_code.lower() == 'python', "sorry!! only writing python code is supported."
|
| 80 |
|
| 81 |
def _generate_input_to_writer(self, input_data: Dict[str, Any]):
|
| 82 |
+
"""This function generates the input to the writer function.
|
| 83 |
+
:param input_data (Dict[str, Any]): The input data.
|
| 84 |
+
:return: (str, str) The first value is a string containing the content to be written to the temp file.
|
| 85 |
+
The second value is a string containing the location of the temp file.
|
| 86 |
+
"""
|
| 87 |
code_str = input_data['code']
|
| 88 |
code_lib_location = input_data["memory_files"]["code_library"]
|
| 89 |
content_to_write = self._generate_content(code_lib_location, code_str)
|
|
|
|
| 94 |
self,
|
| 95 |
input_data: Dict[str, Any]
|
| 96 |
):
|
| 97 |
+
"""This function runs the atomic flow.
|
| 98 |
+
:param input_data (Dict[str, Any]): The input data.
|
| 99 |
+
:return: Dict[str, Any] The output data.
|
| 100 |
+
"""
|
| 101 |
self._check_input(input_data)
|
| 102 |
|
| 103 |
# ~~~ Getting input data to the file editor ~~~
|