Gradio logo
  1. 附加功能
  2. 流式输入
←

流式输出

警报

→

流式输入

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

端到端示例

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

←

流式输出

警报

→