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)