Update CodeFileEditAtomicFlow.py
Browse files
CodeFileEditAtomicFlow.py
CHANGED
@@ -4,7 +4,7 @@ from typing import Dict, Any
|
|
4 |
from flows.base_flows.atomic import AtomicFlow
|
5 |
import os
|
6 |
|
7 |
-
class
|
8 |
def _write_code_to_temp_file(self, code_str, language_of_code, code_lib_location):
|
9 |
directory = os.path.dirname(code_lib_location)
|
10 |
if language_of_code.lower() == 'python':
|
@@ -26,10 +26,11 @@ class CodeFileEditorAtomicFlow(AtomicFlow):
|
|
26 |
)
|
27 |
with open(temp_file_location, "w") as file:
|
28 |
file.write(content)
|
29 |
-
|
|
|
30 |
|
31 |
except Exception as e:
|
32 |
-
return False, str(e), temp_file_location
|
33 |
|
34 |
def _check_input(self, input_data: Dict[str, Any]):
|
35 |
assert "code" in input_data, "code is not passed to CodeFileEditAtomicFlow"
|
@@ -48,7 +49,7 @@ class CodeFileEditorAtomicFlow(AtomicFlow):
|
|
48 |
code_str = input_data['code']
|
49 |
language_of_code = input_data["language_of_code"]
|
50 |
code_lib_location = input_data["memory_files"]["code_library"]
|
51 |
-
result, code_editor_output, temp_file_location = self._write_code_to_temp_file(code_str, language_of_code, code_lib_location)
|
52 |
if result:
|
53 |
try:
|
54 |
subprocess.run(["code", temp_file_location], timeout=10)
|
@@ -57,4 +58,5 @@ class CodeFileEditorAtomicFlow(AtomicFlow):
|
|
57 |
response = {}
|
58 |
response["code_editor_output"] = code_editor_output
|
59 |
response["temp_code_file_location"] = temp_file_location
|
|
|
60 |
return response
|
|
|
4 |
from flows.base_flows.atomic import AtomicFlow
|
5 |
import os
|
6 |
|
7 |
+
class CodeFileEditAtomicFlow(AtomicFlow):
|
8 |
def _write_code_to_temp_file(self, code_str, language_of_code, code_lib_location):
|
9 |
directory = os.path.dirname(code_lib_location)
|
10 |
if language_of_code.lower() == 'python':
|
|
|
26 |
)
|
27 |
with open(temp_file_location, "w") as file:
|
28 |
file.write(content)
|
29 |
+
file_written_timestamp = os.path.getmtime(temp_file_location)
|
30 |
+
return True, f"Code written to {temp_file_location}", temp_file_location, file_written_timestamp
|
31 |
|
32 |
except Exception as e:
|
33 |
+
return False, str(e), temp_file_location, 0
|
34 |
|
35 |
def _check_input(self, input_data: Dict[str, Any]):
|
36 |
assert "code" in input_data, "code is not passed to CodeFileEditAtomicFlow"
|
|
|
49 |
code_str = input_data['code']
|
50 |
language_of_code = input_data["language_of_code"]
|
51 |
code_lib_location = input_data["memory_files"]["code_library"]
|
52 |
+
result, code_editor_output, temp_file_location, file_written_timestamp = self._write_code_to_temp_file(code_str, language_of_code, code_lib_location)
|
53 |
if result:
|
54 |
try:
|
55 |
subprocess.run(["code", temp_file_location], timeout=10)
|
|
|
58 |
response = {}
|
59 |
response["code_editor_output"] = code_editor_output
|
60 |
response["temp_code_file_location"] = temp_file_location
|
61 |
+
response["temp_code_file_written_timestamp"] = file_written_timestamp
|
62 |
return response
|