Spaces:
Running
Running
File size: 4,220 Bytes
58f00c9 | 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 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 | """Throttled progress messages (speed, ETA, cancel button)."""
from __future__ import annotations
import asyncio
import logging
import time
from typing import Any, Optional
from .utils import clip, esc, human_duration, human_size, progress_bar
log = logging.getLogger(__name__)
class ProgressReporter:
"""Callable ``(current, total)`` that edits a status message at most every ``interval`` seconds."""
def __init__(self, client: Any, message: Any, job_id: int, name: str, interval: float = 4.0):
self.client = client
self.message = message
self.job_id = job_id
self.name = name
self.interval = interval
self.stage = ""
self._last_edit = 0.0
self._muted_until = 0.0
self._inflight: Optional[asyncio.Future] = None
self._speed = 0.0
self._last_t = 0.0
self._last_bytes = 0
# ---- helpers -------------------------------------------------------------
def _buttons(self):
from telethon.tl.custom import Button
return [[Button.inline("✖ Cancel", f"job:x:{self.job_id}".encode())]]
def render(self, current: int, total: int) -> str:
head = f"<b>{self.stage}</b>\n<code>{esc(clip(self.name, 120))}</code>\n"
if total <= 0:
return head + "\nWorking…"
frac = current / total
eta = (total - current) / self._speed if self._speed > 0 else 0
speed = f"{human_size(self._speed)}/s" if self._speed > 0 else "…"
eta_text = human_duration(eta) if self._speed > 0 else "…"
return (
f"{head}\n{progress_bar(frac)} {frac * 100:.1f}%\n"
f"{human_size(current)} / {human_size(total)} · {speed} · ETA {eta_text}"
)
async def _edit(self, text: str, buttons: Any, after: Optional[asyncio.Future] = None) -> None:
from telethon import errors
if after is not None: # keep edits in order: wait for the one still in flight
try:
await after
except Exception: # noqa: BLE001
pass
try:
await self.client.edit_message(self.message, text, buttons=buttons, parse_mode="html")
except errors.MessageNotModifiedError:
pass
except errors.FloodWaitError as exc:
self._muted_until = time.monotonic() + exc.seconds + 1
except Exception as exc: # noqa: BLE001 - progress is best effort
log.debug("progress edit failed: %s", exc)
# ---- public API -----------------------------------------------------------
def set_stage(self, stage: str, total: int = 0) -> None:
"""Switch stage and show it right away."""
self.stage = stage
self._speed = 0.0
self._last_t = time.monotonic()
self._last_bytes = 0
self._last_edit = time.monotonic()
self._schedule(self.render(0, total), force=True)
def _schedule(self, text: str, force: bool = False) -> None:
"""Queue an edit; ordinary progress ticks are dropped while one is in flight, a stage change is not."""
busy = self._inflight is not None and not self._inflight.done()
if (busy and not force) or time.monotonic() < self._muted_until:
return
self._inflight = asyncio.ensure_future(self._edit(text, self._buttons(), after=self._inflight if busy else None))
def __call__(self, current: int, total: int) -> None:
now = time.monotonic()
dt = now - self._last_t
if dt >= 1.0:
inst = max(0, current - self._last_bytes) / dt
self._speed = inst if self._speed == 0 else self._speed * 0.6 + inst * 0.4
self._last_t, self._last_bytes = now, current
if now - self._last_edit < self.interval and current < total:
return
self._last_edit = now
self._schedule(self.render(current, total))
async def finish(self, text: str) -> None:
"""Final text, buttons removed."""
if self._inflight is not None:
try:
await self._inflight
except Exception: # noqa: BLE001
pass
self._muted_until = 0.0
await self._edit(text, None)
|