流式输入

在之前的指南中，我们介绍了如何从事件处理器流式传输一系列输出。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()

端到端示例

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