Repository: Avaiga/taipy
File: doc/gui/examples/controls/chat-streaming.py (Python)

Code before:
    "né": "births",
    "dead": "deaths",
    "mort": "deaths",
}
user_agent = "https://taipy.io/demo"

# the list of messages
messages: list[tuple[str, str, str]] = []  # (Message id, message, sender)

# the users
users = [
    ["wikipedia", Icon("https://www.wikipedia.org/static/apple-touch/wikipedia.png", "Wikipedia")],
    ["taipy", Icon("https://docs.taipy.io/en/latest/assets/images/favicon.png", "Taipy")],
]


# user state initialization
def on_init(state: State):
    # Do not share the message list with other users
    state.messages = []


# add an image if one is present in the wikipedia returned json
def add_image_to_message(state: State, idx: int, text: str, image_url: str):
    msg_content: str = state.messages[idx][1]
    pos = msg_content.find(text)
    if pos > -1:
        msg_content = msg_content[: pos + len(text)] + f"\n\n![{text}]({image_url})" + msg_content[pos + len(text) :]
        set_message(state, msg_content, idx)


# invoked by update_message through a thread
def update_message_with_image(gui: Gui, state_id: str, message_idx: int, text: str, image: dict):
    if src := image.get("source"):
        time.sleep(0.2) # to show the streaming effect
        invoke_callback(
            gui,
            state_id,
            add_image_to_message,
            [message_idx, text, src],
        )


# invoked by request_wikipedia
def update_message(state: State, json, event_type: str, for_date: str, idx: int):
    if isinstance(json, dict):
        # response initial message
        set_message(state, f"{event_type} for {for_date}: \n", idx)

        for event in json.get(event_type, []):
            time.sleep(0.2) # to show the streaming effect

Review comment (suggestion):
```suggestion
    if (pos := msg_content.find(text)) > -1:
```
may be less readable. Your choice.

Code after:
    "né": "births",
    "dead": "deaths",
    "mort": "deaths",
}
user_agent = "https://taipy.io/demo"

# The list of messages
messages: list[tuple[str, str, str]] = []  # (Message id, message, sender)

# The two users of this app
users = [
    ["wikipedia", Icon("https://www.wikipedia.org/static/apple-touch/wikipedia.png", "Wikipedia")],
    ["taipy", Icon("https://docs.taipy.io/en/latest/assets/images/favicon.png", "Taipy")],
]


# Initialize the user state
def on_init(state: State):
    #  Messages are for this user only
    state.messages = []


# Add the image if there is one in the Wikipedia returned data
def add_image_to_message(state: State, idx: int, text: str, image_url: str):
    msg_content: str = state.messages[idx][1]
    if (pos := msg_content.find(text)) > -1:
        msg_content = msg_content[: pos + len(text)] + f"\n\n![{text}]({image_url})" + msg_content[pos + len(text) :]
        set_message(state, msg_content, idx)


# Invoked by update_message through a thread
def update_message_with_image(gui: Gui, state_id: str, message_idx: int, text: str, image: dict):
    if src := image.get("source"):
        time.sleep(0.2) # Apply the typewriter effect
        invoke_callback(
            gui,
            state_id,
            add_image_to_message,
            [message_idx, text, src],
        )


# Invoked by query_wikipedia()
def update_message(state: State, json, event_type: str, for_date: str, idx: int):
    if isinstance(json, dict):
        # Initial response content
        set_message(state, f"{event_type} for {for_date}:\n", idx)

        for event in json.get(event_type, []):
            time.sleep(0.2) # Apply the typewriter effect
            # Update response text