|
|
import os |
|
|
from typing import Dict, Any |
|
|
from flow_modules.Tachi67.CodeFileEditFlowModule import CodeFileEditAtomicFlow |
|
|
|
|
|
class RunCodeFileEditAtomicFlow(CodeFileEditAtomicFlow): |
|
|
|
|
|
def _generate_content(self, code_str, comment_sign) -> str: |
|
|
if comment_sign == "<!-- -->": |
|
|
instruction = ( |
|
|
"<!-- The below code will be executed.\n" |
|
|
"Change the code if you consider necessary.\n" |
|
|
"Shut this vscode session completely to start running the code. -->\n" |
|
|
) |
|
|
else: |
|
|
instruction = ( |
|
|
f"{comment_sign} The below code will be executed.\n" |
|
|
f"{comment_sign} Change the code if you consider necessary.\n" |
|
|
f"{comment_sign} Shut this vscode session completely to start running the code.\n" |
|
|
) |
|
|
content = ( |
|
|
instruction + |
|
|
"###########\n" |
|
|
"# Code:\n" + |
|
|
code_str + |
|
|
"\n############\n" |
|
|
) |
|
|
return content |
|
|
|
|
|
def _generate_temp_file_location(self, code_lib_location, file_extension): |
|
|
directory = os.path.dirname(code_lib_location) |
|
|
ret = os.path.join(directory, 'run_code_temp'+file_extension) |
|
|
return ret |
|
|
|
|
|
def _check_input(self, input_data: Dict[str, Any]): |
|
|
assert "code" in input_data, "code is not passed to RunCodeFileEditAtomicFlow" |
|
|
assert "memory_files" in input_data, "memory_files is not passed to RunCodeFileEditAtomicFlow" |
|
|
assert "language" in input_data, "language is not passed to RunCodeFileEditAtomicFlow" |
|
|
|
|
|
def _generate_file_extension_and_comment_sign(self, language: str): |
|
|
details = { |
|
|
"python": {"extension": ".py", "comment": "#"}, |
|
|
"bash": {"extension": ".sh", "comment": "#"}, |
|
|
"shell": {"extension": ".sh", "comment": "#"}, |
|
|
"javascript": {"extension": ".js", "comment": "//"}, |
|
|
"html": {"extension": ".html", "comment": "<!-- -->"}, |
|
|
"applescript": {"extension": ".scpt", "comment": "--"}, |
|
|
"r": {"extension": ".r", "comment": "#"}, |
|
|
"powershell": {"extension": ".ps1", "comment": "#"} |
|
|
} |
|
|
if language.lower() not in details: |
|
|
raise NotImplemented |
|
|
return details.get(language.lower()) |
|
|
|
|
|
def _generate_input_to_writer(self, input_data: Dict[str, Any]): |
|
|
code_str = input_data['code'] |
|
|
code_lib_location = input_data["memory_files"]["code_library"] |
|
|
lang_details = self._generate_file_extension_and_comment_sign(input_data["language"]) |
|
|
content_to_write = self._generate_content(code_str, lang_details["comment"]) |
|
|
file_location_to_write = self._generate_temp_file_location(code_lib_location, lang_details["extension"]) |
|
|
return content_to_write, file_location_to_write |
|
|
|
|
|
def run( |
|
|
self, |
|
|
input_data: Dict[str, Any] |
|
|
): |
|
|
self._check_input(input_data) |
|
|
|
|
|
|
|
|
content_to_write, file_location_to_write = self._generate_input_to_writer(input_data) |
|
|
|
|
|
|
|
|
_, code_editor_output, temp_file_location, _ = self._write_code_content_to_file( |
|
|
file_location_to_write, content_to_write) |
|
|
|
|
|
|
|
|
|
|
|
response = {} |
|
|
response["code_editor_output"] = code_editor_output |
|
|
response["temp_code_file_location"] = temp_file_location |
|
|
return response |