Source code for langchain_community.callbacks.streamlit.mutable_expander

from __future__ import annotations

from enum import Enum
from typing import TYPE_CHECKING, Any, Dict, List, NamedTuple, Optional

if TYPE_CHECKING:
    from streamlit.delta_generator import DeltaGenerator
    from streamlit.type_util import SupportsStr


class ChildType(Enum):
    """子类型的枚举器。"""

    MARKDOWN = "MARKDOWN"
    EXCEPTION = "EXCEPTION"


class ChildRecord(NamedTuple):
    """作为一个命名元组的子记录。"""

    type: ChildType
    kwargs: Dict[str, Any]
    dg: DeltaGenerator


class MutableExpander:
    """Streamlit可展开器，可以重命名并动态展开/折叠。"""

    def __init__(self, parent_container: DeltaGenerator, label: str, expanded: bool):
        """创建一个新的MutableExpander。

参数
----------
parent_container
    MutableExpander将被创建在其中的`st.container`。

    当标签更改时，MutableExpander会透明地删除并重新创建其底层的`st.expander`实例，并使用`parent_container`来确保在屏幕上的相同位置重新创建此底层的expander。
label
    expander的初始标签。
expanded
    expander的初始`expanded`值。
"""
        self._label = label
        self._expanded = expanded
        self._parent_cursor = parent_container.empty()
        self._container = self._parent_cursor.expander(label, expanded)
        self._child_records: List[ChildRecord] = []

    @property
    def label(self) -> str:
        """扩展器的标签字符串。"""
        return self._label

    @property
    def expanded(self) -> bool:
        """如果扩展器是使用`expanded=True`创建的，则为True。"""
        return self._expanded

    def clear(self) -> None:
        """完全删除容器及其内容。清空的容器无法被重复使用。
"""
        self._container = self._parent_cursor.empty()
        self._child_records.clear()

    def append_copy(self, other: MutableExpander) -> None:
        """将另一个MutableExpander的子项的副本附加到此MutableExpander。
"""
        other_records = other._child_records.copy()
        for record in other_records:
            self._create_child(record.type, record.kwargs)

    def update(
        self, *, new_label: Optional[str] = None, new_expanded: Optional[bool] = None
    ) -> None:
        """更改展开器的标签和展开状态"""
        if new_label is None:
            new_label = self._label
        if new_expanded is None:
            new_expanded = self._expanded

        if self._label == new_label and self._expanded == new_expanded:
            # No change!
            return

        self._label = new_label
        self._expanded = new_expanded
        self._container = self._parent_cursor.expander(new_label, new_expanded)

        prev_records = self._child_records
        self._child_records = []

        # Replay all children into the new container
        for record in prev_records:
            self._create_child(record.type, record.kwargs)

    def markdown(
        self,
        body: SupportsStr,
        unsafe_allow_html: bool = False,
        *,
        help: Optional[str] = None,
        index: Optional[int] = None,
    ) -> int:
        """向容器添加一个Markdown元素并返回其索引。"""
        kwargs = {"body": body, "unsafe_allow_html": unsafe_allow_html, "help": help}
        new_dg = self._get_dg(index).markdown(**kwargs)
        record = ChildRecord(ChildType.MARKDOWN, kwargs, new_dg)
        return self._add_record(record, index)

    def exception(
        self, exception: BaseException, *, index: Optional[int] = None
    ) -> int:
        """将一个异常元素添加到容器中，并返回其索引。"""
        kwargs = {"exception": exception}
        new_dg = self._get_dg(index).exception(**kwargs)
        record = ChildRecord(ChildType.EXCEPTION, kwargs, new_dg)
        return self._add_record(record, index)

    def _create_child(self, type: ChildType, kwargs: Dict[str, Any]) -> None:
        """使用给定参数创建一个新的子节点"""
        if type == ChildType.MARKDOWN:
            self.markdown(**kwargs)
        elif type == ChildType.EXCEPTION:
            self.exception(**kwargs)
        else:
            raise RuntimeError(f"Unexpected child type {type}")

    def _add_record(self, record: ChildRecord, index: Optional[int]) -> int:
        """将ChildRecord添加到self._children中。如果指定了`index`，则替换该索引处的现有记录。否则，将记录追加到列表的末尾。
返回添加记录的索引。
"""
        if index is not None:
            # Replace existing child
            self._child_records[index] = record
            return index

        # Append new child
        self._child_records.append(record)
        return len(self._child_records) - 1

    def _get_dg(self, index: Optional[int]) -> DeltaGenerator:
        if index is not None:
            # Existing index: reuse child's DeltaGenerator
            assert 0 <= index < len(self._child_records), f"Bad index: {index}"
            return self._child_records[index].dg

        # No index: use container's DeltaGenerator
        return self._container