Spaces:
Running on Zero
Running on Zero
File size: 1,402 Bytes
cf8e3da | 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 | import time
from typing import Any, Callable, Optional, Tuple
from gradio.context import LocalContext
POLL_SECONDS = 0.2
TIMEOUT_SECONDS = 130.0
FINISHED_STATUSES = ("complete", "failed")
def current_event_id() -> Optional[str]:
return LocalContext.event_id.get(None)
def event_progress(event_id: Optional[str]) -> Optional[Tuple[float, str]]:
queue = getattr(LocalContext.blocks.get(None), "_queue", None)
if not event_id or queue is None:
return None
for job in list(queue.active_jobs):
for event in list(job or []):
if event._id == event_id and event.progress and event.progress.progress_data:
unit = event.progress.progress_data[-1]
return None if unit.progress is None else (unit.progress, unit.desc)
return None
def mirror_progress(
entry: Any,
progress: Callable[..., Any],
read: Callable[[Optional[str]], Optional[Tuple[float, str]]] = event_progress,
sleep: Callable[[float], Any] = time.sleep,
clock: Callable[[], float] = time.monotonic,
) -> None:
deadline = clock() + TIMEOUT_SECONDS
shown = None
while clock() < deadline and entry.status not in FINISHED_STATUSES and not entry.cancelled:
level = read(entry.progress_event)
if level and level != shown:
shown = level
progress(level[0], desc=level[1])
sleep(POLL_SECONDS)
|