view asyncio_threads/stop_token/main.py @ 278:8d560f50ed4c

Improve infinite canvas interactions and browser chrome Render Lucide icons directly with Raylib, add searchable icon browsing, robust text editing, entity lifecycle animations, z-order-safe input, semantic themes, and animated editable browser controls. Document rendering, pinning, context, and component extension for future agents. Co-authored-by: Copilot <[email protected]> Copilot-Session: f68442b1-fa8f-46a0-9689-81710613bbd4
author MrJuneJune <me@mrjunejune.com>
date Mon, 17 Aug 2026 22:16:14 -0700
parents 46daba6e3cf4
children
line wrap: on
line source

from typing import Generator
from typing import List

# [Hello There, "I am <EN", "D> Something"]
#     |

def truncate_stream(text_stream: Generator[str, None, None], stop_token: str) -> Generator[str, None, None]:
    buffer = ""
    for chunk in text_stream:
        can_be_prefix = False
        buffer += chunk
        if stop_token not in buffer:
            for end in range(1, len(stop_token)):
                if stop_token[:end] in buffer[-1 * len(stop_token):]:
                    can_be_prefix = True
            if not can_be_prefix:
                yield chunk
                buffer = ""
        else: 
            pos = buffer.find(stop_token)
            yield buffer[:pos]
            return

def stream_to_list(stream: Generator[str, None, None]) -> List[str]:
    return [chunk for chunk in stream]

def list_to_stream(chunks: List[str]) -> Generator[str, None, None]:
    for chunk in chunks:
        yield chunk


print(
    stream_to_list(
        truncate_stream(list_to_stream(["Hello there ", "I'm doing great today.<END>", "Thanks for asking.", "How are you"]), "<END>")
    )
) 

print(
    stream_to_list(
        truncate_stream(list_to_stream(["Hello there ", "I'm doing great today.<E", "N", "D>", "Thanks for asking.", "How are you"]), "<END>")
    )
)