PlanFileEditAtomicFlow.py

import subprocess
import webbrowser
from typing import Dict, Any
from flows.base_flows.atomic import AtomicFlow
import os


class PlanFileEditAtomicFlow(AtomicFlow):
    def _generate_content(self, plan_file_location, plan_str) -> str:
        content = (
                "The below plan will be written to " +
                plan_file_location + "\n"
                "Edit the plan directly or provide your thoughts down below if you have any suggestions.\n"
                "When you are done editing plan and providing feedback, save file and close the current VSCode session to continue.\n"
                "###########\n"
                "Plan:\n" +
                plan_str +
                "\n############\n"
                "Thoughts:"
        )
        return content

    def _generate_temp_file_location(self, plan_file_location):
        directory = os.path.dirname(plan_file_location)
        ret = os.path.join(directory, 'temp_plan.txt')
        return ret

    def _write_plan_content_to_file(self, file_location, content: str):
        try:
            with open(file_location, "w") as file:
                file.write(content)

            return True, f"Plan written to {file_location}", file_location

        except Exception as e:
            return False, str(e), file_location

    def _check_input(self, input_data: Dict[str, Any]):
        assert any(item in input_data for item in ["plan", "new_plan"]), "plan or new_plan is not passed to PlanFileEditAtomicFlow"
        assert "plan_file_location" in input_data, "plan_file_location not passed to PlanFileEditAtomicFlow"
        plan_file_loc = input_data["plan_file_location"]
        assert os.path.exists(plan_file_loc), f"{plan_file_loc} does not exist"
        assert os.path.isfile(plan_file_loc), f"{plan_file_loc} is not a file"

    def _generate_input_to_writer(self, input_data: Dict[str, Any]):
        plan_str = input_data['plan'] if "plan" in input_data else input_data['new_plan']
        plan_file_location = input_data["plan_file_location"]
        content_to_write = self._generate_content(plan_file_location, plan_str)
        file_location_to_write = self._generate_temp_file_location(plan_file_location)
        return content_to_write, file_location_to_write

    def run(
            self,
            input_data: Dict[str, Any]
    ):
        self._check_input(input_data)

        # ~~~ Getting input data to the file editor ~~~
        content_to_write, file_location_to_write = self._generate_input_to_writer(input_data)

        # ~~~ Calling the writer function ~~~
        result, plan_editor_output, temp_file_location = self._write_plan_content_to_file(
            file_location_to_write, content_to_write)

        # ~~~ Generating return variables ~~~
        response = {}
        response["plan_editor_output"] = plan_editor_output
        response["temp_plan_file_location"] = temp_file_location
        return response