Fix live interfaces for audio/image streaming (#9883)

* Fix live interfaces for audio/image streaming

* add changeset

* code

* code

---------

Co-authored-by: gradio-pr-bot <[email protected]>
Co-authored-by: Abubakar Abid <[email protected]>

.changeset/eight-women-lead.md

@@ -0,0 +1,6 @@
+---
+"@gradio/audio": patch
+"gradio": patch
+---
+
+fix:Fix live interfaces for audio/image streaming

gradio/events.py

@@ -643,6 +643,17 @@ def inner(*args, **kwargs):
         event_trigger.event_name = _event_name  # type: ignore
         event_trigger.has_trigger = _has_trigger  # type: ignore
         event_trigger.callback = _callback  # type: ignore
+        event_trigger.connection = _connection  # type: ignore
+        event_specific_args = (
+            [
+                d["name"]
+                for d in _event_specific_args
+                if d.get("component_prop", "true") != "false"
+            ]
+            if _event_specific_args
+            else None
+        )
+        event_trigger.event_specific_args = event_specific_args  # type: ignore
         return event_trigger
 
 
@@ -675,6 +686,8 @@ def on(
     concurrency_limit: int | None | Literal["default"] = "default",
     concurrency_id: str | None = None,
     show_api: bool = True,
+    time_limit: int | None = None,
+    stream_every: float = 0.5,
 ) -> Dependency:
     """
     Sets up an event listener that triggers a function when the specified event(s) occur. This is especially
@@ -700,6 +713,8 @@ def on(
         concurrency_limit: If set, this is the maximum number of this event that can be running simultaneously. Can be set to None to mean no concurrency_limit (any number of this event can be running simultaneously). Set to "default" to use the default concurrency limit (defined by the `default_concurrency_limit` parameter in `Blocks.queue()`, which itself is 1 by default).
         concurrency_id: If set, this is the id of the concurrency group. Events with the same concurrency_id will be limited by the lowest set concurrency_limit.
         show_api: whether to show this event in the "view API" page of the Gradio app, or in the ".view_api()" method of the Gradio clients. Unlike setting api_name to False, setting show_api to False will still allow downstream apps as well as the Clients to use this event. If fn is None, show_api will automatically be set to False.
+        time_limit: The time limit for the function to run. Parameter only used for the `.stream()` event.
+        stream_every: The latency (in seconds) at which stream chunks are sent to the backend. Defaults to 0.5 seconds. Parameter only used for the `.stream()` event.
     Example:
         import gradio as gr
         with gr.Blocks() as demo:
@@ -746,6 +761,8 @@ def wrapper(func):
                 concurrency_id=concurrency_id,
                 show_api=show_api,
                 trigger_mode=trigger_mode,
+                time_limit=time_limit,
+                stream_every=stream_every,
             )
 
             @wraps(func)
@@ -793,6 +810,16 @@ def inner(*args, **kwargs):
         max_batch_size=max_batch_size,
         show_api=show_api,
         trigger_mode=trigger_mode,
+        connection="stream"
+        if any(t.connection == "stream" for t in (triggers_typed or []))
+        else "sse",
+        event_specific_args=[
+            a
+            for t in (triggers_typed or [])
+            for a in cast(list[str], t.event_specific_args or [])
+        ],
+        time_limit=time_limit,
+        stream_every=stream_every,
     )
     set_cancel_events(methods, cancels)
     return Dependency(None, dep.get_config(), dep_index, fn)

gradio/interface.py

@@ -142,6 +142,8 @@ def __init__(
         | Literal["auto"]
         | Literal["manual"]
         | None = None,
+        time_limit: int | None = 30,
+        stream_every: float = 0.5,
         **kwargs,
     ):
         """
@@ -182,6 +184,8 @@ def __init__(
             show_progress: how to show the progress animation while event is running: "full" shows a spinner which covers the output component area as well as a runtime display in the upper right corner, "minimal" only shows the runtime display, "hidden" shows no progress animation at all
             example_labels: a list of labels for each example. If provided, the length of this list should be the same as the number of examples, and these labels will be used in the UI instead of rendering the example values.
             fill_width: whether to horizontally expand to fill container fully. If False, centers and constrains app to a maximum width.
+            time_limit: The time limit for the stream to run. Default is 30 seconds. Parameter only used for streaming images or audio if the interface is live and the input components are set to "streaming=True".
+            stream_every: The latency (in seconds) at which stream chunks are sent to the backend. Defaults to 0.5 seconds. Parameter only used for streaming images or audio if the interface is live and the input components are set to "streaming=True".
         """
         super().__init__(
             analytics_enabled=analytics_enabled,
@@ -197,6 +201,8 @@ def __init__(
             fill_width=fill_width,
             **kwargs,
         )
+        self.time_limit = time_limit
+        self.stream_every = stream_every
         self.api_name: str | Literal[False] | None = api_name
         self.interface_type = InterfaceTypes.STANDARD
         if (inputs is None or inputs == []) and (outputs is None or outputs == []):
@@ -702,7 +708,9 @@ def attach_submit_events(
                     preprocess=not (self.api_mode),
                     postprocess=not (self.api_mode),
                     show_progress="hidden" if streaming_event else self.show_progress,
-                    trigger_mode="always_last",
+                    trigger_mode="always_last" if not streaming_event else "multiple",
+                    time_limit=self.time_limit,
+                    stream_every=self.stream_every,
                 )
         else:
             if _submit_btn is None:
@@ -716,6 +724,12 @@ def attach_submit_events(
                 if component.has_event(Events.submit)
             ]
 
+            for component in self.input_components:
+                if getattr(component, "streaming", None):
+                    warnings.warn(
+                        "Streaming components are only supported in live interfaces."
+                    )
+
             if _stop_btn:
                 extra_output = [_submit_btn, _stop_btn]
 

gradio/queueing.py

@@ -666,7 +666,9 @@ async def process_events(
                                 else ServerMessage.process_streaming,
                                 output=old_response,
                                 success=old_response is not None,
-                                time_limit=cast(int, fn.time_limit) - first_iteration
+                                time_limit=None
+                                if not fn.time_limit
+                                else cast(int, fn.time_limit) - first_iteration
                                 if event.streaming
                                 else None,
                             ),
@@ -679,7 +681,11 @@ async def process_events(
                         if awake_events[0].streaming:
                             awake_events, closed_events = await Queue.wait_for_batch(
                                 awake_events,
-                                [cast(float, fn.time_limit) - first_iteration]
+                                # We need to wait for all of the events to have the latest input data
+                                # the max time is the time limit of the function or 30 seconds (arbitrary) but should
+                                # never really take that long to make a request from the client to the server unless
+                                # the client disconnected.
+                                [cast(float, fn.time_limit or 30) - first_iteration]
                                 * len(awake_events),
                             )
                             for closed_event in closed_events:

guides/07_streaming/05_real-time-speech-recognition.md

@@ -57,6 +57,8 @@ To make this a *streaming* demo, we need to make these changes:
 2. Set `live=True` in the `Interface`
 3. Add a `state` to the interface to store the recorded audio of a user
 
+Tip: You can also set `time_limit` and `stream_every` parameters in the interface. The `time_limit` caps the amount of time each user's stream can take. The default is 30 seconds so users won't be able to stream audio for more than 30 seconds. The `stream_every` parameter controls how frequently data is sent to your function. By default it is 0.5 seconds.
+
 Take a look below.
 
 $code_stream_asr

js/audio/interactive/InteractiveAudio.svelte

@@ -201,7 +201,7 @@
 		dispatch("start_recording");
 		if (!inited) await prepare_audio();
 		header = undefined;
-		if (streaming) {
+		if (streaming && recorder.state != "recording") {
 			recorder.start(stream_every * 1000);
 		}
 	}