in mapillary_tools/upload.py [0:0]
def _setup_tdqm(emitter: uploader.EventEmitter) -> None:
upload_pbar: T.Optional[tqdm] = None
@emitter.on("upload_fetch_offset")
def upload_fetch_offset(payload: uploader.Progress) -> None:
nonlocal upload_pbar
if upload_pbar is not None:
upload_pbar.close()
nth = payload["sequence_idx"] + 1
total = payload["total_sequence_count"]
import_path = payload.get("import_path")
if import_path is None:
_desc = f"Uploading ({nth}/{total})"
else:
_desc = f"Uploading {os.path.basename(import_path)} ({nth}/{total})"
upload_pbar = tqdm(
total=payload["entity_size"],
desc=_desc,
unit="B",
unit_scale=True,
unit_divisor=1024,
initial=payload["offset"],
disable=LOG.getEffectiveLevel() <= logging.DEBUG,
)
@emitter.on("upload_progress")
def upload_progress(payload: uploader.Progress) -> None:
assert upload_pbar is not None, "progress_bar must be initialized"
upload_pbar.update(payload["chunk_size"])
@emitter.on("upload_end")
def upload_end(_: uploader.Progress) -> None:
nonlocal upload_pbar
if upload_pbar:
upload_pbar.close()
upload_pbar = None