|
from typing import Dict, Any |
|
|
|
from flow_modules.Tachi67.InterpreterFlowModule import InterpreterAtomicFlow |
|
|
|
|
|
class CodeFileEditorAtomicFlow(InterpreterAtomicFlow): |
|
|
|
def __init__(self): |
|
super().__init__() |
|
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} |
|
|