Spaces:
Running on Zero
Running on Zero
| 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) | |