From 347dc0d726bf6d49b7e2a48bc7934a5a9e454bb9 Mon Sep 17 00:00:00 2001
From: moon <shawmakesmagic@gmail.com>
Date: Wed, 26 Jul 2023 13:33:16 -0700
Subject: [PATCH] Add formatted string options

---
 README.md        |  33 ++--
 easytask/main.py | 385 ++++++++++++++++++++++++++++++++++++++++-------
 easytask/test.py |  83 ++++++++++
 3 files changed, 438 insertions(+), 63 deletions(-)

diff --git a/README.md b/README.md
index dc75380..1fdadf9 100644
--- a/README.md
+++ b/README.md
@@ -182,17 +182,6 @@ This will create a new task with the goal "Write README.md file", and then mark
     print(steps)
     ```
 
-**`create_step(goal: str, steps: list, plan: str) -> list`**
-
-    Adds a new step to the given steps based on the goal, and updates the plan. Returns the updated steps.
-
-    *Example:*
-
-    ```python
-    steps = create_step("New step for the project", steps, "Plan for the project")
-    print(steps)
-    ```
-
 **`update_step(task: Union[dict, int, str], step: dict) -> dict`**
 
     Updates the specified step of the specified task. The task can be specified as a dictionary (as returned by `create_task`), an integer ID, or a string ID.
@@ -234,6 +223,28 @@ This will create a new task with the goal "Write README.md file", and then mark
     cancel_step(task, "Step to cancel")
     ```
 
+**`get_task_as_formatted_string(task: dict, include_plan: bool = True, include_status: bool = True, include_steps: bool = True) -> str`**
+
+    Returns a string representation of the task, including the plan, status, and steps based on the arguments provided.
+
+    *Example:*
+
+    ```python
+    task_string = get_task_as_formatted_string(task, include_plan=True, include_status=True, include_steps=True)
+    print(task_string)
+    ```
+
+**`list_tasks_as_formatted_string() -> str`**
+
+    Retrieves and formats a list of all current tasks. Returns a string containing details of all current tasks.
+
+    *Example:*
+
+    ```python
+    tasks_string = list_tasks_as_formatted_string()
+    print(tasks_string)
+    ```
+
 # Contributions Welcome
 
 If you like this library and want to contribute in any way, please feel free to submit a PR and I will review it. Please note that the goal here is simplicity and accesibility, using common language and few dependencies.
diff --git a/easytask/main.py b/easytask/main.py
index 31df4f8..0224d5d 100644
--- a/easytask/main.py
+++ b/easytask/main.py
@@ -53,7 +53,18 @@
 debug = os.environ.get("DEBUG", False)
 
 
-def create_task(goal, plan=None, steps=None):
+def create_task(goal, plan=None, steps=None, model="gpt-3.5-turbo-0613"):
+    """Create a task and store it in memory.
+
+    Args:
+        goal (str): The goal of the task.
+        plan (str, optional): A plan to accomplish the task. Defaults to None.
+        steps (list or dict, optional): Steps needed to complete the task. Defaults to None.
+        model (str, optional): The OpenAI model to use for AI operations. Defaults to 'gpt-3.5-turbo-0613'.
+
+    Returns:
+        None
+    """
     if plan is None:
         log("Creating plan for goal: {}".format(goal), log=debug)
         plan = create_plan(goal)
@@ -63,6 +74,7 @@ def create_task(goal, plan=None, steps=None):
             functions=[step_creation_function],
             function_call="create_steps",
             debug=debug,
+            model=model,
         )
 
         steps = response["arguments"]["steps"]
@@ -72,8 +84,8 @@ def create_task(goal, plan=None, steps=None):
         steps = json.dumps(steps)
 
     # get timestamp
-    created_at = datetime.now()
-    updated_at = datetime.now()
+    created_at = datetime.timestamp(datetime.now())
+    updated_at = datetime.timestamp(datetime.now())
 
     task = {
         "created_at": created_at,
@@ -90,6 +102,14 @@ def create_task(goal, plan=None, steps=None):
 
 
 def list_tasks(status="in_progress"):
+    """List all tasks with the given status.
+
+    Args:
+        status (str, optional): The status of the tasks to retrieve. Defaults to 'in_progress'.
+
+    Returns:
+        list: A list of tasks with the given status.
+    """
     debug = os.environ.get("DEBUG", False)
     memories = get_memories(
         "task", filter_metadata={"status": status}, include_embeddings=False
@@ -99,11 +119,19 @@ def list_tasks(status="in_progress"):
 
 
 def search_tasks(search_term, status="in_progress"):
-    # return tasks whose goal is most relevant to the search term
+    """Search for tasks related to a given search term.
+
+    Args:
+        search_term (str): The search term to use.
+        status (str, optional): The status of the tasks to retrieve. Defaults to 'in_progress'.
+
+    Returns:
+        list: A list of tasks related to the search term.
+    """
     memories = search_memory(
         "task",
         search_term,
-        filter_metadata={"status": "in_progress"},
+        filter_metadata={"status": status},
         include_embeddings=False,
         include_distances=False,
     )
@@ -112,6 +140,14 @@ def search_tasks(search_term, status="in_progress"):
 
 
 def get_task_id(task):
+    """Get the ID of a task.
+
+    Args:
+        task (dict or int or str): The task to get the ID of.
+
+    Returns:
+        str: The ID of the task.
+    """
     if isinstance(task, dict):
         return task["id"]
     elif isinstance(task, int):
@@ -121,23 +157,77 @@ def get_task_id(task):
 
 
 def delete_task(task):
+    """Delete a task.
+
+    Args:
+        task (dict or int or str): The task to delete.
+
+    Returns:
+        dict: The response from the memory deletion operation.
+    """
     log("Deleting task: {}".format(task), log=debug)
     return delete_memory("task", get_task_id(task))
 
 
 def finish_task(task):
+    """Mark a task as complete.
+
+    Args:
+        task (dict or int or str): The task to finish.
+
+    Returns:
+        dict: The response from the memory update operation.
+    """
     log("Finishing task: {}".format(task), log=debug)
-    return update_memory("task", get_task_id(task), metadata={"status": "complete"})
+    updated_at = datetime.timestamp(datetime.now())
+
+    memory = get_memory("task", get_task_id(task))
+
+    metadata = memory["metadata"]
+    metadata["status"] = "complete"
+    metadata["updated_at"] = updated_at
+
+    return update_memory(
+        "task",
+        get_task_id(task),
+        metadata=metadata,
+    )
 
 
 def cancel_task(task):
+    """Cancel a task.
+
+    Args:
+        task (dict or int or str): The task to cancel.
+
+    Returns:
+        dict: The response from the memory update operation.
+    """
     log("Cancelling task: {}".format(task), log=debug)
-    return update_memory("task", get_task_id(task), metadata={"status": "cancelled"})
+    updated_at = datetime.timestamp(datetime.now())
+
+    memory = get_memory("task", get_task_id(task))
+
+    metadata = memory["metadata"]
+    metadata["status"] = "cancelled"
+    metadata["updated_at"] = updated_at
+
+    return update_memory(
+        "task",
+        get_task_id(task),
+        metadata=metadata,
+    )
 
 
-# Get Last Created Task
 def get_last_created_task():
-    # return the task with the most recent created_at date
+    """
+    Get the most recently created task.
+
+    Returns
+    -------
+    dict or None
+        The task with the most recent created_at date. If no tasks are found, None is returned.
+    """
     tasks = get_memories("task", include_embeddings=False)
     sorted_tasks = sorted(
         tasks, key=lambda x: x["metadata"]["created_at"], reverse=True
@@ -146,9 +236,15 @@ def get_last_created_task():
     return sorted_tasks[0] if sorted_tasks else None
 
 
-# Get Last Updated Task
 def get_last_updated_task():
-    # return the task with the most recent updated_at date
+    """
+    Get the most recently updated task.
+
+    Returns
+    -------
+    dict or None
+        The task with the most recent updated_at date. If no tasks are found, None is returned.
+    """
     tasks = get_memories("task", include_embeddings=False)
     sorted_tasks = sorted(
         tasks, key=lambda x: x["metadata"]["updated_at"], reverse=True
@@ -158,6 +254,14 @@ def get_last_updated_task():
 
 
 def get_current_task():
+    """
+    Get the current active task.
+
+    Returns
+    -------
+    dict or None
+        The task marked as the current active task. If no current task is found, None is returned.
+    """
     memory = get_memories(
         "task", filter_metadata={"current": True}, include_embeddings=False
     )
@@ -170,6 +274,14 @@ def get_current_task():
 
 
 def set_current_task(task):
+    """Set a task as the current task.
+
+    Args:
+        task (dict or int or str): The task to be set as current.
+
+    Returns:
+        dict: The response from the memory update operation.
+    """
     task_id = get_task_id
 
     memories = get_memories(
@@ -177,94 +289,263 @@ def set_current_task(task):
     )
 
     for memory in memories:
-        update_memory("task", memory["id"], metadata={"current": False})
+        metadata = memory["metadata"]
+        metadata["current"] = False
+        update_memory("task", memory["id"], metadata=metadata)
     log("Setting current task: {}".format(task), log=debug)
-    return update_memory("task", task_id, metadata={"current": True})
+    metadata = memory["metadata"]
+    metadata["current"] = True
+    return update_memory("task", task_id, metadata=metadata)
+
+
+def create_plan(goal, model="gpt-3.5-turbo-0613"):
+    """Create a plan for the goal using OpenAI API.
 
+    Args:
+        goal (str): The goal for which to create the plan.
+        model (str, optional): The OpenAI model to use for AI operations. Defaults to 'gpt-3.5-turbo-0613'.
 
-# Plans
-def create_plan(goal):
+    Returns:
+        str: The generated plan.
+    """
     response = openai_text_call(
-        compose_prompt(planning_prompt, {"goal": goal}), debug=debug
+        compose_prompt(planning_prompt, {"goal": goal}), debug=debug, model=model
     )
     return response["text"]
 
 
 def update_plan(task, plan):
+    """Update the plan for a task.
+
+    Args:
+        task (dict or int or str): The task for which to update the plan.
+        plan (str): The new plan.
+
+    Returns:
+        None
+    """
     task_id = get_task_id(task)
     log("Updating plan for task: {}".format(task), log=debug)
-    update_memory("task", task_id, metadata={"plan": plan})
+    memory = get_memory("task", task_id)
+    metadata = memory["metadata"]
+    metadata["plan"] = plan
+    metadata["updated_at"] = datetime.timestamp(datetime.now())
+    update_memory("task", task_id, metadata=metadata)
+
 
+def create_steps(goal, plan, model="gpt-3.5-turbo-0613"):
+    """Create a series of steps based on the plan and the goal using OpenAI API.
 
-# Create Steps
-def create_steps(goal, plan):
+    Args:
+        goal (str): The goal for which to create the steps.
+        plan (str): The plan based on which to create the steps.
+        model (str, optional): The OpenAI model to use for AI operations. Defaults to 'gpt-3.5-turbo-0613'.
+
+    Returns:
+        list: The generated steps.
+    """
     log("Creating steps for goal: {}".format(goal), log=debug)
+    updated_at = datetime.timestamp(datetime.now())
     response = openai_function_call(
-        text=compose_prompt(step_creation_prompt, {"goal": goal, "plan": plan}),
+        text=compose_prompt(
+            step_creation_prompt, {"goal": goal, "plan": plan, "updated_at": updated_at}
+        ),
         functions=[step_creation_function],
         function_call="create_steps",
         debug=debug,
+        model=model,
     )
     return response["arguments"]["steps"]
 
 
-# Create Step
-def create_step(goal, steps, plan):
-    steps.append({"content": goal, "completed": False})
-    update_plan(goal, steps, plan)
-    log("Creating step for goal: {}".format(goal), log=debug)
-    return steps
-
-
-# Update Step
 def update_step(task, step):
+    """
+    Update a step in a task.
+
+    Parameters
+    ----------
+    task : dict
+        The task in which the step is to be updated.
+    step : dict
+        The step which is to be updated.
+
+    Returns
+    -------
+    dict
+        The updated task after updating the step.
+    """
     task_id = get_task_id(task)
     task = get_memory("task", task_id)
-    steps = task["metadata"]["steps"]
-    steps = json.loads(steps)
+    metadata = task["metadata"]
+    metadata["steps"] = json.loads(metadata["steps"])
 
-    for s in steps:
+    for s in metadata["steps"]:
         if s["content"] == step["content"]:
             s["completed"] = step["completed"]
 
-    steps = json.dumps(steps)
-    log("Updating step for task: {}\nSteps are: {}".format(task, steps), log=debug)
-    return update_memory("task", task_id, metadata={"steps": steps})
+    metadata["steps"] = json.dumps(metadata["steps"])
+    metadata["updated_at"] = datetime.timestamp(datetime.now())
+    log("Updating step for task: {}\nSteps are: {}".format(task, metadata["steps"]), log=debug)
+    return update_memory("task", task_id, metadata=metadata)
 
 
-# Add Step
 def add_step(task, step):
+    """
+    Add a step to a task.
+
+    Parameters
+    ----------
+    task : dict
+        The task to which the step is to be added.
+    step : str
+        The step which is to be added to the task.
+
+    Returns
+    -------
+    dict
+        The updated task after adding the step.
+    """
     task_id = get_task_id(task)
     task = get_memory("task", task_id)
-    steps = task["metadata"]["steps"]
-    steps = json.loads(steps)
+    metadata = task["metadata"]
+    steps = json.loads(metadata["steps"])
     steps.append({"content": step, "completed": False})
-    steps = json.dumps(steps)
-    log("Adding step for task: {}\nSteps are: {}".format(task, steps), log=debug)
-    return update_memory("task", task_id, metadata={"steps": steps})
+    metadata["steps"] = json.dumps(steps)
+    log("Adding step for task: {}\nSteps are: {}".format(task, metadata["steps"]), log=debug)
+    metadata["updated_at"] = datetime.timestamp(datetime.now())
+    return update_memory("task", task_id, metadata=metadata)
 
 
-# Finish Step
 def finish_step(task, step):
+    """
+    Mark a step in a task as completed.
+
+    Parameters
+    ----------
+    task : dict
+        The task containing the step to be marked as completed.
+    step : str
+        The step which is to be marked as completed.
+
+    Returns
+    -------
+    dict
+        The updated task after marking the step as completed.
+    """
     task_id = get_task_id(task)
     task = get_memory("task", task_id)
-    steps = task["metadata"]["steps"]
-    steps = json.loads(steps)
+    metadata = task["metadata"]
+    steps = json.loads(metadata["steps"])
     for s in steps:
         if s["content"] == step:
             s["completed"] = True
-    steps = json.dumps(steps)
-    log("Finishing step for task: {}\nSteps are: {}".format(task, steps), log=debug)
-    return update_memory("task", task_id, metadata={"steps": steps})
+    metadata["steps"] = json.dumps(steps)
+    log("Finishing step for task: {}\nSteps are: {}".format(task, metadata["steps"]), log=debug)
+    metadata["updated_at"] = datetime.timestamp(datetime.now())
+    return update_memory("task", task_id, metadata=metadata)
 
 
-# Cancel Step
 def cancel_step(task, step):
+    """
+    Remove a step from a task.
+
+    Parameters
+    ----------
+    task : dict
+        The task from which the step is to be removed.
+    step : str
+        The step which is to be removed from the task.
+
+    Returns
+    -------
+    dict
+        The updated task after removing the step.
+    """
     task_id = get_task_id(task)
     task = get_memory("task", task_id)
-    steps = task["metadata"]["steps"]
+    metadata = task["metadata"]
+    steps = metadata["steps"]
     steps = json.loads(steps)
     steps = [s for s in steps if s["content"] != step]
-    steps = json.dumps(steps)
-    log("Cancelling step for task: {}\nSteps are: {}".format(task, steps), log=debug)
-    return update_memory("task", task_id, metadata={"steps": steps})
+    metadata["steps"] = json.dumps(steps)
+    metadata["updated_at"] = datetime.timestamp(datetime.now())
+    log("Cancelling step for task: {}\nSteps are: {}".format(task, metadata["steps"]), log=debug)
+    return update_memory("task", task_id, metadata=metadata)
+
+
+def get_next_step(task):
+    """
+    This function will get the task, unpack the string to a dict from the
+    "steps" object in task["metadata"], find the first step in the list
+    that hasn't been completed and return it
+    """
+
+    # First, get the steps from the task metadata
+    steps = json.loads(task["metadata"]["steps"])
+
+    # Loop through each step to find the first one that isn't completed
+    for step in steps:
+        if not step["completed"]:
+            # If a step is not completed, return it
+            return step
+
+    # If all steps are completed, return None
+    return None
+
+
+def get_task_as_formatted_string(
+    task, include_plan=True, include_status=True, include_steps=True
+):
+    """
+    This function will return a string representation of the task,
+    including the plan, status and steps based on the arguments provided
+    """
+
+    # Define an empty list to store the task details
+    task_details = []
+
+    # Append each detail to the list based on the arguments
+    if include_plan:
+        task_details.append("Plan: {}".format(task["metadata"]["plan"]))
+
+    if include_status:
+        task_details.append("Status: {}".format(task["metadata"]["status"]))
+
+    if include_steps:
+        # For the steps, since it's a list, we need to format each step separately
+        steps = json.loads(task["metadata"]["steps"])
+        formatted_steps = ", ".join(
+            [
+                "{}: {}".format(
+                    step["content"],
+                    "Completed" if step["completed"] else "Not completed",
+                )
+                for step in steps
+            ]
+        )
+        task_details.append("Steps: {}".format(formatted_steps))
+
+    # Finally, join all the task details into a single string and return it
+    return "\n".join(task_details)
+
+
+def list_tasks_as_formatted_string():
+    """
+    Retrieve and format a list of all current tasks.
+
+    Returns:
+        str: Formatted string containing details of all current tasks.
+    """
+
+    # Get all tasks
+    tasks = list_tasks()
+
+    # Define an empty list to store the task details
+    task_details = []
+
+    # Loop through each task and append the formatted string to the list
+    for task in tasks:
+        task_details.append(get_task_as_formatted_string(task))
+
+    # Finally, join all the task details into a single string and return it
+    return "\n".join(task_details)
diff --git a/easytask/test.py b/easytask/test.py
index 2b36e73..6b82459 100644
--- a/easytask/test.py
+++ b/easytask/test.py
@@ -15,6 +15,7 @@
     finish_step,
     cancel_step,
 )
+from easytask.main import get_next_step, get_task_as_formatted_string, list_tasks_as_formatted_string
 
 goal = "Make a balogna sandwich"
 
@@ -154,6 +155,8 @@ def test_update_plan():
     plan = "New plan"
     update_plan(task, plan)
     updated_task = get_memory("task", task["id"])
+    print('updated_task')
+    print(updated_task)
     assert updated_task["metadata"]["plan"] == plan
     teardown()
 
@@ -198,3 +201,83 @@ def test_cancel_step():
     step_data = next((s for s in updated_steps if s["content"] == step), None)
     assert step_data == None
     teardown()
+
+
+def test_get_next_step():
+    task = {
+        "metadata": {
+            "steps": json.dumps(
+                [
+                    {"content": "Step 1", "completed": False},
+                    {"content": "Step 2", "completed": False},
+                ]
+            )
+        }
+    }
+
+    next_step = get_next_step(task)
+    assert next_step == {"content": "Step 1", "completed": False}
+
+
+def test_get_task_as_formatted_string():
+    task = {
+        "metadata": {
+            "plan": "Plan 1",
+            "status": "in_progress",
+            "steps": json.dumps(
+                [
+                    {"content": "Step 1", "completed": False},
+                    {"content": "Step 2", "completed": False},
+                ]
+            ),
+        }
+    }
+
+    task_string = get_task_as_formatted_string(task)
+    expected_string = (
+        "Plan: Plan 1\n"
+        "Status: in_progress\n"
+        "Steps: Step 1: Not completed, Step 2: Not completed"
+    )
+    assert task_string == expected_string
+
+
+def test_list_tasks_as_formatted_string():
+    # Setup: Create multiple tasks
+    tasks = [
+        {
+            "plan": "Plan 1",
+            "status": "in_progress",
+            "steps": json.dumps([
+                {"content": "Step 1", "completed": False},
+                {"content": "Step 2", "completed": False}
+            ])
+        },
+        {
+            "plan": "Plan 2",
+            "status": "Completed",
+            "steps": json.dumps([
+                {"content": "Step 3", "completed": True},
+                {"content": "Step 4", "completed": True}
+            ])
+        },
+    ]
+    for task in tasks:
+        create_task("some goal", task["plan"], task["steps"])
+
+    tasks = get_memories("task", unique=False)
+
+    # Execute
+    tasks_string = list_tasks_as_formatted_string()
+
+    # Check if tasks_string includes all tasks
+    for task in tasks:
+        task_string = get_task_as_formatted_string(task)
+        print("*** task_string is")
+        print(task_string)
+        print("***  tasks_string")
+        print(tasks_string)
+        assert task_string in tasks_string
+
+    # Teardown: Remove all tasks
+    wipe_category("task")