| 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 | |