| from typing import Dict, Any |
|
|
| from flow_modules.Tachi67.InterpreterFlowModule import InterpreterAtomicFlow |
|
|
| class PlanFileEditorAtomicFlow(InterpreterAtomicFlow): |
|
|
| def _process_interperter_inputs(self, input_data: Dict[str, Any]): |
| input_data['language'] = 'python' |
| file_location = input_data['file_location'] |
| plan_str = input_data['plan'] |
| input_data['code'] = f""" |
| import os |
| file_location = {repr(file_location)} |
| plan_str = {repr(plan_str)} |
| if os.path.isdir(file_location): |
| file_location = os.path.join(file_location, 'plan.txt') |
| with open(file_location, 'w') as file: |
| file.write(plan_str) |
| print('plan written to' + file_location)""" |
|
|
| def run( |
| self, |
| input_data: Dict[str, Any]): |
| self._process_interperter_inputs(input_data) |
| self._process_input_data(input_data) |
| response = self._call() |
| return {"plan_writer_output": response} |
