Upload 6 files

PlanFileEditAtomicFlow.py

@@ -14,8 +14,22 @@ class PlanFileEditAtomicFlow(AtomicFlow):
     *Output Interface*:
     - `plan_editor_output`: str
     - `temp_plan_file_location`: str
+
+    *Configuration Parameters*:
+    - `input_interface`: The input interface of the atomic flow.
+    - `output_interface`: The output interface of the atomic flow.
+
     """
     def _generate_content(self, plan_file_location, plan_str) -> str:
+        """
+        This function generates the content to be written to the plan file.
+        :param plan_file_location: The location of the plan file.
+        :type plan_file_location: str
+        :param plan_str: The plan to be written to the plan file.
+        :type plan_str: str
+        :return: The content to be written to the plan file.
+        :rtype: str
+        """
         content = (
                 "The below plan will be written to " +
                 plan_file_location + "\n"
@@ -30,11 +44,27 @@ class PlanFileEditAtomicFlow(AtomicFlow):
         return content
 
     def _generate_temp_file_location(self, plan_file_location):
+        """
+        This function generates the location of the temp plan file.
+        :param plan_file_location: The location of the plan file.
+        :type plan_file_location: str
+        :return: The location of the temp plan file.
+        :rtype: str
+        """
         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):
+        """
+        This function writes the plan content to the plan file.
+        :param file_location: The location of the plan file.
+        :type file_location: str
+        :param content: The content to be written to the plan file.
+        :type content: str
+        :return: The result of writing the plan content to the plan file.
+        :rtype: Tuple[bool, str, str]
+        """
         try:
             with open(file_location, "w") as file:
                 file.write(content)
@@ -45,6 +75,14 @@ class PlanFileEditAtomicFlow(AtomicFlow):
             return False, str(e), file_location
 
     def _check_input(self, input_data: Dict[str, Any]):
+        """
+        This function checks if the input data is valid.
+        :param input_data: The input data.
+        :type input_data: Dict[str, Any]
+        :return: None
+        :rtype: None
+        :raises AssertionError: If the input data is invalid.
+        """
         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"]
@@ -53,13 +91,18 @@ class PlanFileEditAtomicFlow(AtomicFlow):
 
     def _generate_input_to_writer(self, input_data: Dict[str, Any]):
         """
-        sometimes the plan generator will return an arrary of indexed plans, like
+        sometimes the plan generator will return an array of indexed plans, like
         [
             "1. Extend the code library with a function named 'import_libraries'. This function should import necessary libraries for the task...",
             "2. Extend the code library with a function named 'fetch_stock_prices'. This function should take two inputs: 'company_code' and 'duration'...",
             "3. Investigate the issue with importing the 'fetch_stock_prices' function from the library..."
         ]
-        In this case we need to prase this format accorrdingly.
+        In this case we need to parse this format accordingly.
+
+        :param input_data: The input data.
+        :type input_data: Dict[str, Any]
+        :return: The content to be written to the plan file and the location of the temp plan file.
+        :rtype: Tuple[str, str]
         """
         
         plan = input_data['plan'] if "plan" in input_data else input_data['new_plan']
@@ -78,6 +121,13 @@ class PlanFileEditAtomicFlow(AtomicFlow):
             self,
             input_data: Dict[str, Any]
     ):
+        """
+        This function runs the atomic flow.
+        :param input_data: The input data.
+        :type input_data: Dict[str, Any]
+        :return: The output data.
+        :rtype: Dict[str, Any]
+        """
         self._check_input(input_data)
 
         # ~~~ Getting input data to the file editor ~~~

README.md

@@ -2,6 +2,8 @@
 
 * [PlanFileEditAtomicFlow](#PlanFileEditAtomicFlow)
   * [PlanFileEditAtomicFlow](#PlanFileEditAtomicFlow.PlanFileEditAtomicFlow)
+    * [run](#PlanFileEditAtomicFlow.PlanFileEditAtomicFlow.run)
+* [run](#run)
 * [\_\_init\_\_](#__init__)
 
 <a id="PlanFileEditAtomicFlow"></a>
@@ -27,6 +29,32 @@ to the user.
 - `plan_editor_output`: str
 - `temp_plan_file_location`: str
 
+*Configuration Parameters*:
+- `input_interface`: The input interface of the atomic flow.
+- `output_interface`: The output interface of the atomic flow.
+
+<a id="PlanFileEditAtomicFlow.PlanFileEditAtomicFlow.run"></a>
+
+#### run
+
+```python
+def run(input_data: Dict[str, Any])
+```
+
+This function runs the atomic flow.
+
+**Arguments**:
+
+- `input_data` (`Dict[str, Any]`): The input data.
+
+**Returns**:
+
+`Dict[str, Any]`: The output data.
+
+<a id="run"></a>
+
+# run
+
 <a id="__init__"></a>
 
 # \_\_init\_\_