| from typing import Dict, Any |
|
|
| from flow_modules.Tachi67.InterpreterFlowModule import InterpreterAtomicFlow |
|
|
|
|
| class CodeFileEditorAtomicFlow(InterpreterAtomicFlow): |
|
|
| def __init__(self, **kwargs): |
| super().__init__(**kwargs) |
| self.file_type = { |
| "python": "py", |
| "bash": "sh", |
| "r": "r", |
| "applescript": "applescript", |
| "html": "html", |
| "javascript": "js", |
| "shell": "sh", |
| "powershell": "ps1" |
| } |
| def _process_interperter_inputs(self, input_data: Dict[str, Any]): |
| language_of_code = input_data['language_of_code'].lower() |
| file_extension = self.file_type[language_of_code] |
| file_location = input_data['file_location'] |
| code_to_write = input_data['code'] |
| input_data['code'] = f""" |
| import os |
| file_location = {repr(file_location)} |
| code_to_write = {repr(code_to_write)} |
| if os.path.isdir(file_location): |
| file_location = os.path.join(file_location, 'library.{file_extension}') |
| with open(file_location, 'a') as file: |
| file.write('\\n' + code_to_write) |
| print('code written to' + file_location)""" |
| input_data['language'] = 'python' |
|
|
| def run( |
| self, |
| input_data: Dict[str, Any]): |
| self._process_interperter_inputs(input_data) |
| self._process_input_data(input_data) |
| response = self._call() |
| return {"code_writer_output": response} |
|
|