Agent Blackboard

def add_data(
    self, data: Union[MemoryItem, Dict[str, str], str], memory: Memory
) -> None:
    """
    Add the data to the a memory in the blackboard.
    :param data: The data to be added. It can be a dictionary or a MemoryItem or a string.
    :param memory: The memory to add the data to.
    """

    if isinstance(data, dict):
        data_memory = MemoryItem()
        data_memory.set_values_from_dict(data)
        memory.add_memory_item(data_memory)
    elif isinstance(data, MemoryItem):
        memory.add_memory_item(data)
    elif isinstance(data, str):
        data_memory = MemoryItem()
        data_memory.set_values_from_dict({"text": data})
        memory.add_memory_item(data_memory)

def add_image(
    self,
    screenshot_path: str = "",
    metadata: Optional[Dict[str, str]] = None,
) -> None:
    """
    Add the image to the blackboard.
    :param screenshot_path: The path of the image.
    :param metadata: The metadata of the image.
    """

    if os.path.exists(screenshot_path):

        screenshot_str = PhotographerFacade().encode_image_from_path(
            screenshot_path
        )
    else:
        print(f"Screenshot path {screenshot_path} does not exist.")
        screenshot_str = ""

    image_memory_item = ImageMemoryItem()
    image_memory_item.set_values_from_dict(
        {
            ImageMemoryItemNames.METADATA: metadata.get(
                ImageMemoryItemNames.METADATA
            ),
            ImageMemoryItemNames.IMAGE_PATH: screenshot_path,
            ImageMemoryItemNames.IMAGE_STR: screenshot_str,
        }
    )

    self.screenshots.add_memory_item(image_memory_item)

def add_questions(self, questions: Union[MemoryItem, Dict[str, str]]) -> None:
    """
    Add the data to the blackboard.
    :param questions: The data to be added. It can be a dictionary or a MemoryItem or a string.
    """

    self.add_data(questions, self.questions)

def add_requests(self, requests: Union[MemoryItem, Dict[str, str]]) -> None:
    """
    Add the data to the blackboard.
    :param requests: The data to be added. It can be a dictionary or a MemoryItem or a string.
    """

    self.add_data(requests, self.requests)

def add_trajectories(self, trajectories: Union[MemoryItem, Dict[str, str]]) -> None:
    """
    Add the data to the blackboard.
    :param trajectories: The data to be added. It can be a dictionary or a MemoryItem or a string.
    """

    self.add_data(trajectories, self.trajectories)

def blackboard_to_prompt(self) -> List[str]:
    """
    Convert the blackboard to a prompt.
    :return: The prompt.
    """
    prefix = [
        {
            "type": "text",
            "text": "[Blackboard:]",
        }
    ]

    blackboard_prompt = (
        prefix
        + self.texts_to_prompt(self.questions, "[Questions & Answers:]")
        + self.texts_to_prompt(self.requests, "[Request History:]")
        + self.texts_to_prompt(self.trajectories, "[Step Trajectories:]")
        + self.screenshots_to_prompt()
    )

    return blackboard_prompt

def clear(self) -> None:
    """
    Clear the blackboard.
    """
    self.questions.clear()
    self.requests.clear()
    self.trajectories.clear()
    self.screenshots.clear()

def is_empty(self) -> bool:
    """
    Check if the blackboard is empty.
    :return: True if the blackboard is empty, False otherwise.
    """
    return (
        self.questions.is_empty()
        and self.requests.is_empty()
        and self.trajectories.is_empty()
        and self.screenshots.is_empty()
    )

def load_questions(self, file_path: str, last_k=-1) -> None:
    """
    Load the data from a file.
    :param file_path: The path of the file.
    :param last_k: The number of lines to read from the end of the file. If -1, read all lines.
    """
    qa_list = self.read_json_file(file_path, last_k)
    for qa in qa_list:
        self.add_questions(qa)

def questions_to_json(self) -> str:
    """
    Convert the data to a dictionary.
    :return: The data in the dictionary format.
    """
    return self.questions.to_json()

@staticmethod
def read_json_file(file_path: str, last_k=-1) -> Dict[str, str]:
    """
    Read the json file.
    :param file_path: The path of the file.
    :param last_k: The number of lines to read from the end of the file. If -1, read all lines.
    :return: The data in the file.
    """

    data_list = []

    # Check if the file exists
    if os.path.exists(file_path):
        # Open the file and read the lines
        with open(file_path, "r", encoding="utf-8") as file:
            lines = file.readlines()

        # If last_k is not -1, only read the last k lines
        if last_k != -1:
            lines = lines[-last_k:]

        # Parse the lines as JSON
        for line in lines:
            try:
                data = json.loads(line.strip())
                data_list.append(data)
            except json.JSONDecodeError:
                print(f"Warning: Unable to parse line as JSON: {line}")

    return data_list

def requests_to_json(self) -> str:
    """
    Convert the data to a dictionary.
    :return: The data in the dictionary format.
    """
    return self.requests.to_json()

def screenshots_to_json(self) -> str:
    """
    Convert the images to a dictionary.
    :return: The images in the dictionary format.
    """
    return self.screenshots.to_json()

def screenshots_to_prompt(self) -> List[str]:
    """
    Convert the images to a prompt.
    :return: The prompt.
    """

    user_content = []
    for screenshot_dict in self.screenshots.list_content:
        user_content.append(
            {
                "type": "text",
                "text": json.dumps(
                    screenshot_dict.get(ImageMemoryItemNames.METADATA, "")
                ),
            }
        )
        user_content.append(
            {
                "type": "image_url",
                "image_url": {
                    "url": screenshot_dict.get(ImageMemoryItemNames.IMAGE_STR, "")
                },
            }
        )

    return user_content

def texts_to_prompt(self, memory: Memory, prefix: str) -> List[str]:
    """
    Convert the data to a prompt.
    :return: The prompt.
    """

    user_content = [
        {"type": "text", "text": f"{prefix}\n {json.dumps(memory.list_content)}"}
    ]

    return user_content

def trajectories_to_json(self) -> str:
    """
    Convert the data to a dictionary.
    :return: The data in the dictionary format.
    """
    return self.trajectories.to_json()