流式输入

在之前的指南中，我们介绍了如何从事件处理程序中流式传输输出序列。Gradio 还允许您将用户摄像头中的图像或麦克风中的音频块流式传输到您的事件处理程序中。这可以用于使用 Gradio 创建实时对象检测应用程序或对话式聊天应用程序。

目前，gr.Image 和 gr.Audio 组件通过 stream 事件支持输入流式传输。让我们创建一个最简单的流式应用程序，它只是返回未修改的网络摄像头流。

import gradio as gr

with gr.Blocks() as demo:
    with gr.Row():
        with gr.Column():
            input_img = gr.Image(label="Input", sources="webcam")
        with gr.Column():
            output_img = gr.Image(label="Output")
        input_img.stream(lambda s: s, input_img, output_img, time_limit=15, stream_every=0.1, concurrency_limit=30)

if __name__ == "__main__":

    demo.launch()

试试看！当用户开始录制时，流式传输事件被触发。在后台，网络摄像头将每 0.1 秒拍摄一张照片并将其发送到服务器。然后服务器将返回该图像。

stream 事件有两个独特的关键字参数

time_limit - 这是 gradio 服务器将花费在处理事件上的时间量。媒体流本质上是无界的，因此设置时间限制非常重要，以防止一个用户占用 Gradio 队列。时间限制仅计算处理流所花费的时间，而不计算在队列中等待所花费的时间。输入图像底部显示的橙色条表示剩余时间。当时间限制到期时，用户将自动重新加入队列。
stream_every - 这是流捕获输入并将其发送到服务器的频率（以秒为单位）。对于像图像检测或处理这样的演示，希望设置一个较小的值以获得“实时”效果。对于像语音转录这样的演示，较高的值很有用，这样转录算法可以更好地理解所说的内容。

一个真实的图像演示

让我们创建一个演示，用户可以选择一个过滤器来应用于他们的网络摄像头流。用户可以从边缘检测过滤器、卡通过滤器或简单地垂直翻转流中进行选择。

import gradio as gr
import numpy as np
import cv2

def transform_cv2(frame, transform):
    if transform == "cartoon":
        # prepare color
        img_color = cv2.pyrDown(cv2.pyrDown(frame))
        for _ in range(6):
            img_color = cv2.bilateralFilter(img_color, 9, 9, 7)
        img_color = cv2.pyrUp(cv2.pyrUp(img_color))

        # prepare edges
        img_edges = cv2.cvtColor(frame, cv2.COLOR_RGB2GRAY)
        img_edges = cv2.adaptiveThreshold(
            cv2.medianBlur(img_edges, 7),
            255,
            cv2.ADAPTIVE_THRESH_MEAN_C,
            cv2.THRESH_BINARY,
            9,
            2,
        )
        img_edges = cv2.cvtColor(img_edges, cv2.COLOR_GRAY2RGB)
        # combine color and edges
        img = cv2.bitwise_and(img_color, img_edges)
        return img
    elif transform == "edges":
        # perform edge detection
        img = cv2.cvtColor(cv2.Canny(frame, 100, 200), cv2.COLOR_GRAY2BGR)
        return img
    else:
        return np.flipud(frame)

with gr.Blocks() as demo:
    with gr.Row():
        with gr.Column():
            transform = gr.Dropdown(choices=["cartoon", "edges", "flip"],
                                    value="flip", label="Transformation")
            input_img = gr.Image(sources=["webcam"], type="numpy")
        with gr.Column():
            output_img = gr.Image(streaming=True)
        dep = input_img.stream(transform_cv2, [input_img, transform], [output_img],
                                time_limit=30, stream_every=0.1, concurrency_limit=30)

demo.launch()

您会注意到，如果您更改过滤器值，它将立即在输出流中生效。这是流事件与其他 Gradio 事件相比的一个重要区别。流的输入值可以在流正在处理时更改。

提示： 我们将图像输出组件的“streaming”参数设置为“True”。这样做可以让服务器自动将我们的输出图像转换为 base64 格式，这是一种高效的流式传输格式。

统一图像演示

对于某些图像流式传输演示，例如上面的演示，我们不需要显示单独的输入和输出组件。如果我们只显示修改后的输出流，我们的应用程序看起来会更简洁。

我们可以通过仅将输入图像组件指定为流事件的输出来做到这一点。

import gradio as gr
import numpy as np
import cv2

def transform_cv2(frame, transform):
    if transform == "cartoon":
        # prepare color
        img_color = cv2.pyrDown(cv2.pyrDown(frame))
        for _ in range(6):
            img_color = cv2.bilateralFilter(img_color, 9, 9, 7)
        img_color = cv2.pyrUp(cv2.pyrUp(img_color))

        # prepare edges
        img_edges = cv2.cvtColor(frame, cv2.COLOR_RGB2GRAY)
        img_edges = cv2.adaptiveThreshold(
            cv2.medianBlur(img_edges, 7),
            255,
            cv2.ADAPTIVE_THRESH_MEAN_C,
            cv2.THRESH_BINARY,
            9,
            2,
        )
        img_edges = cv2.cvtColor(img_edges, cv2.COLOR_GRAY2RGB)
        # combine color and edges
        img = cv2.bitwise_and(img_color, img_edges)
        return img
    elif transform == "edges":
        # perform edge detection
        img = cv2.cvtColor(cv2.Canny(frame, 100, 200), cv2.COLOR_GRAY2BGR)
        return img
    else:
        return np.flipud(frame)


css=""".my-group {max-width: 500px !important; max-height: 500px !important;}
            .my-column {display: flex !important; justify-content: center !important; align-items: center !important};"""

with gr.Blocks(css=css) as demo:
    with gr.Column(elem_classes=["my-column"]):
        with gr.Group(elem_classes=["my-group"]):
            transform = gr.Dropdown(choices=["cartoon", "edges", "flip"],
                                    value="flip", label="Transformation")
            input_img = gr.Image(sources=["webcam"], type="numpy", streaming=True)
    input_img.stream(transform_cv2, [input_img, transform], [input_img], time_limit=30, stream_every=0.1)


demo.launch()

跟踪过去的输入或输出

您的流式传输功能应该是无状态的。它应该接受当前输入并返回其对应的输出。但是，在某些情况下，您可能希望跟踪过去的输入或输出。例如，您可能希望保留先前 k 个输入的缓冲区，以提高转录演示的准确性。您可以使用 Gradio 的 gr.State() 组件来做到这一点。

让我们用一个示例演示来展示这一点

def transcribe_handler(current_audio, state, transcript):
    next_text = transcribe(current_audio, history=state)
    state.append(current_audio)
    state = state[-3:]
    return state, transcript + next_text

with gr.Blocks() as demo:
    with gr.Row():
        with gr.Column():
            mic = gr.Audio(sources="microphone")
            state = gr.State(value=[])
        with gr.Column():
            transcript = gr.Textbox(label="Transcript")
    mic.stream(transcribe_handler, [mic, state, transcript], [state, transcript],
               time_limit=10, stream_every=1)


demo.launch()

端到端示例

有关从网络摄像头流式传输的端到端示例，请参阅网络摄像头对象检测指南。