view asyncio_threads/stop_token/main.py @ 275:78699f810817

Add Qwen3-VL and WebRTC dictation services Add Bazel targets for the CUDA-backed Qwen3-VL server and a local WebRTC faster-whisper dictation service. Co-authored-by: Copilot <[email protected]> Copilot-Session: e3d8cb06-6c95-4ae0-9757-651d3796ab00
author MrJuneJune <me@mrjunejune.com>
date Mon, 17 Aug 2026 10:58:47 -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>")
    )
)