Memory System

def add_values_from_dict(self, values: Dict[str, Any]) -> None:
    """
    Add fields to the memory item.
    :param values: The values of the fields.
    """
    for key, value in values.items():
        self.set_value(key, value)

def filter(self, keys: List[str] = []) -> None:
    """
    Fetch the memory item.
    :param keys: The keys to fetch.
    :return: The filtered memory item.
    """

    return {key: value for key, value in self.to_dict().items() if key in keys}

def from_dict(self, data: Dict[str, str]) -> None:
    """
    Convert the dictionary to a MemoryItem.
    :param data: The dictionary.
    """
    for key, value in data.items():
        self.set_value(key, value)

def get_value(self, key: str) -> Optional[str]:
    """
    Get the value of the field.
    :param key: The key of the field.
    :return: The value of the field.
    """

    return getattr(self, key, None)

def get_values(self, keys: List[str]) -> dict:
    """
    Get the values of the fields.
    :param keys: The keys of the fields.
    :return: The values of the fields.
    """
    return {key: self.get_value(key) for key in keys}

def set_value(self, key: str, value: str) -> None:
    """
    Add a field to the memory item.
    :param key: The key of the field.
    :param value: The value of the field.
    """
    setattr(self, key, value)

    if key not in self._memory_attributes:
        self._memory_attributes.append(key)

def to_dict(self) -> Dict[str, str]:
    """
    Convert the MemoryItem to a dictionary.
    :return: The dictionary.
    """

    return {
        key: value
        for key, value in self.__dict__.items()
        if key in self._memory_attributes
    }

def to_json(self) -> str:
    """
    Convert the memory item to a JSON string.
    :return: The JSON string.
    """
    return json.dumps(self.to_dict())

def add_memory_item(self, memory_item: MemoryItem) -> None:
    """
    Add a memory item to the memory.
    :param memory_item: The memory item to add.
    """
    self._content.append(memory_item)

def clear(self) -> None:
    """
    Clear the memory.
    """
    self._content = []

def delete_memory_item(self, step: int) -> None:
    """
    Delete a memory item from the memory.
    :param step: The step of the memory item to delete.
    """
    self._content = [item for item in self._content if item.step != step]

def filter_memory_from_keys(self, keys: List[str]) -> List[Dict[str, str]]:
    """
    Filter the memory from the keys. If an item does not have the key, the key will be ignored.
    :param keys: The keys to filter.
    :return: The filtered memory.
    """
    return [item.filter(keys) for item in self._content]

def filter_memory_from_steps(self, steps: List[int]) -> List[Dict[str, str]]:
    """
    Filter the memory from the steps.
    :param steps: The steps to filter.
    :return: The filtered memory.
    """
    return [item.to_dict() for item in self._content if item.step in steps]

def from_list_of_dicts(self, data: List[Dict[str, str]]) -> None:
    """
    Convert the list of dictionaries to the memory.
    :param data: The list of dictionaries.
    """
    self._content = []
    for item in data:
        memory_item = MemoryItem()
        memory_item.from_dict(item)
        self._content.append(memory_item)

def get_latest_item(self) -> MemoryItem:
    """
    Get the latest memory item.
    :return: The latest memory item.
    """
    if self.length == 0:
        return None
    return self._content[-1]

def is_empty(self) -> bool:
    """
    Check if the memory is empty.
    :return: The boolean value indicating if the memory is empty.
    """
    return self.length == 0

def load(self, content: List[MemoryItem]) -> None:
    """
    Load the data from the memory.
    :param content: The content to load.
    """
    self._content = content

def to_json(self) -> str:
    """
    Convert the memory to a JSON string.
    :return: The JSON string.
    """

    return json.dumps(
        [item.to_dict() for item in self._content if item is not None]
    )

def to_list_of_dicts(self) -> List[Dict[str, str]]:
    """
    Convert the memory to a list of dictionaries.
    :return: The list of dictionaries.
    """
    return [item.to_dict() for item in self._content]

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.add_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.add_values_from_dict({"text": data})
        memory.add_memory_item(data_memory)
    else:
        print(f"Warning: Unsupported data type: {type(data)} when adding data.")

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 = utils.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.add_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_from_dict(
    self, blackboard_dict: Dict[str, List[Dict[str, str]]]
) -> None:
    """
    Convert the dictionary to the blackboard.
    :param blackboard_dict: The dictionary.
    """
    self.questions.from_list_of_dicts(blackboard_dict.get("questions", []))
    self.requests.from_list_of_dicts(blackboard_dict.get("requests", []))
    self.trajectories.from_list_of_dicts(blackboard_dict.get("trajectories", []))
    self.screenshots.from_list_of_dicts(blackboard_dict.get("screenshots", []))

def blackboard_to_dict(self) -> Dict[str, List[Dict[str, str]]]:
    """
    Convert the blackboard to a dictionary.
    :return: The blackboard in the dictionary format.
    """
    blackboard_dict = {
        "questions": self.questions.to_list_of_dicts(),
        "requests": self.requests.to_list_of_dicts(),
        "trajectories": self.trajectories.to_list_of_dicts(),
        "screenshots": self.screenshots.to_list_of_dicts(),
    }

    return blackboard_dict

def blackboard_to_json(self) -> str:
    """
    Convert the blackboard to a JSON string.
    :return: The JSON string.
    """
    return json.dumps(self.blackboard_to_dict())

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 Completed Previously:]"
        )
        + 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()

def add_memory_item(self, memory_item: MemoryItem) -> None:
    """
    Add a memory item to the memory.
    :param memory_item: The memory item to add.
    """
    self._content.append(memory_item)

def clear(self) -> None:
    """
    Clear the memory.
    """
    self._content = []

def delete_memory_item(self, step: int) -> None:
    """
    Delete a memory item from the memory.
    :param step: The step of the memory item to delete.
    """
    self._content = [item for item in self._content if item.step != step]

def filter_memory_from_keys(self, keys: List[str]) -> List[Dict[str, str]]:
    """
    Filter the memory from the keys. If an item does not have the key, the key will be ignored.
    :param keys: The keys to filter.
    :return: The filtered memory.
    """
    return [item.filter(keys) for item in self._content]

def filter_memory_from_steps(self, steps: List[int]) -> List[Dict[str, str]]:
    """
    Filter the memory from the steps.
    :param steps: The steps to filter.
    :return: The filtered memory.
    """
    return [item.to_dict() for item in self._content if item.step in steps]

def from_list_of_dicts(self, data: List[Dict[str, str]]) -> None:
    """
    Convert the list of dictionaries to the memory.
    :param data: The list of dictionaries.
    """
    self._content = []
    for item in data:
        memory_item = MemoryItem()
        memory_item.from_dict(item)
        self._content.append(memory_item)

def get_latest_item(self) -> MemoryItem:
    """
    Get the latest memory item.
    :return: The latest memory item.
    """
    if self.length == 0:
        return None
    return self._content[-1]

def is_empty(self) -> bool:
    """
    Check if the memory is empty.
    :return: The boolean value indicating if the memory is empty.
    """
    return self.length == 0

def load(self, content: List[MemoryItem]) -> None:
    """
    Load the data from the memory.
    :param content: The content to load.
    """
    self._content = content

def to_json(self) -> str:
    """
    Convert the memory to a JSON string.
    :return: The JSON string.
    """

    return json.dumps(
        [item.to_dict() for item in self._content if item is not None]
    )

def to_list_of_dicts(self) -> List[Dict[str, str]]:
    """
    Convert the memory to a list of dictionaries.
    :return: The list of dictionaries.
    """
    return [item.to_dict() for item in self._content]

def add_values_from_dict(self, values: Dict[str, Any]) -> None:
    """
    Add fields to the memory item.
    :param values: The values of the fields.
    """
    for key, value in values.items():
        self.set_value(key, value)

def filter(self, keys: List[str] = []) -> None:
    """
    Fetch the memory item.
    :param keys: The keys to fetch.
    :return: The filtered memory item.
    """

    return {key: value for key, value in self.to_dict().items() if key in keys}

def from_dict(self, data: Dict[str, str]) -> None:
    """
    Convert the dictionary to a MemoryItem.
    :param data: The dictionary.
    """
    for key, value in data.items():
        self.set_value(key, value)

def get_value(self, key: str) -> Optional[str]:
    """
    Get the value of the field.
    :param key: The key of the field.
    :return: The value of the field.
    """

    return getattr(self, key, None)

def get_values(self, keys: List[str]) -> dict:
    """
    Get the values of the fields.
    :param keys: The keys of the fields.
    :return: The values of the fields.
    """
    return {key: self.get_value(key) for key in keys}

def set_value(self, key: str, value: str) -> None:
    """
    Add a field to the memory item.
    :param key: The key of the field.
    :param value: The value of the field.
    """
    setattr(self, key, value)

    if key not in self._memory_attributes:
        self._memory_attributes.append(key)

def to_dict(self) -> Dict[str, str]:
    """
    Convert the MemoryItem to a dictionary.
    :return: The dictionary.
    """

    return {
        key: value
        for key, value in self.__dict__.items()
        if key in self._memory_attributes
    }

def to_json(self) -> str:
    """
    Convert the memory item to a JSON string.
    :return: The JSON string.
    """
    return json.dumps(self.to_dict())

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.add_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.add_values_from_dict({"text": data})
        memory.add_memory_item(data_memory)
    else:
        print(f"Warning: Unsupported data type: {type(data)} when adding data.")

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 = utils.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.add_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)