File size: 2,717 Bytes
8fdc036 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 |
from __future__ import annotations
from typing import Callable, Literal, Sequence
from gradio.components import Component
from gradio.context import Context, LocalContext
from gradio.events import EventListener, EventListenerMethod
from gradio.layouts import Column
class Renderable:
def __init__(
self,
fn: Callable,
inputs: list[Component] | Component | None = None,
triggers: EventListener | Sequence[EventListener] | None = None,
concurrency_limit: int | None | Literal["default"] = "default",
concurrency_id: str | None = None,
):
if Context.root_block is None:
raise ValueError("Reactive render must be inside a Blocks context.")
self._id = len(Context.root_block.renderables)
Context.root_block.renderables.append(self)
self.column = Column()
self.column_id = Column()._id
self.fn = fn
self.inputs = [inputs] if isinstance(inputs, Component) else inputs
self.triggers: list[EventListenerMethod] = []
if isinstance(triggers, EventListener):
triggers = [triggers]
if triggers:
self.triggers = [
EventListenerMethod(
getattr(t, "__self__", None) if t.has_trigger else None,
t.event_name,
)
for t in triggers
]
Context.root_block.set_event_trigger(
self.triggers,
self.apply,
self.inputs,
None,
show_api=False,
concurrency_limit=concurrency_limit,
concurrency_id=concurrency_id,
renderable=self,
)
def apply(self, *args, **kwargs):
column_copy = Column(render=False)
column_copy._id = self.column_id
LocalContext.is_render.set(True)
LocalContext.render_block.set(column_copy)
try:
self.fn(*args, **kwargs)
blocks_config = LocalContext.blocks_config.get()
if blocks_config is None:
raise ValueError("Reactive render must be inside a LocalContext.")
blocks_config.blocks[self.column_id] = column_copy
finally:
LocalContext.is_render.set(False)
LocalContext.render_block.set(None)
def render(
inputs: list[Component] | None = None,
triggers: list[EventListener] | None = None,
concurrency_limit: int | None | Literal["default"] = None,
concurrency_id: str | None = None,
):
def wrapper_function(fn):
Renderable(fn, inputs, triggers, concurrency_limit, concurrency_id)
return fn
return wrapper_function
|