0

Как заполнять progress bar по мере загрузки файла?
У нас есть файл, допустим он весит 3.67mb - это 100 процентов, как найти сколько 1 процент от этого числа каждый школьник справиться. Но как заполнять progress bar по степени загрузки файла - вот в чем мой вопрос.
Допустим у нас есть картинка, которая храниться на хостинге мы ее скачиваем с помощью
img = requests.get(hosting.com/img.jpg).content и записуем в соответствующий файл:

with open("img.jpg", "wb") as file:
    file.write(img)

Как отобразить прогресс и скорость загрузки прям как в pip? Как должен выглядеть progress bar
Мне неважно с какой библиотекой вы приведете пример, главное в примере - узнать размер файла, скорость загрузки и вывести это в консоль

3

2 ответа 2

2

Используем:

  • requests - для скачивания. Нужно использовать флаг stream=True, чтобы вручную качать порциями через iter_content
  • tqdm - для вывода прогресса

Пример:

import os
import sys

from pathlib import Path

# pip install tqdm
from tqdm import tqdm

# pip install requests
import requests


def sizeof_fmt(num: int | float) -> str:
    for x in ['bytes', 'KB', 'MB', 'GB']:
        if num < 1024.0:
            return "%.1f %s" % (num, x)

        num /= 1024.0

    return "%.1f %s" % (num, 'TB')


url = 'https://github.com/gil9red/NotesManager/raw/master/bin.rar'
# Streaming, so we can iterate over the response.
rs = requests.get(url, stream=True)

# Total size in bytes.
total_size = int(rs.headers.get('content-length', 0))
print('From content-length:', sizeof_fmt(total_size))

chunk_size = 1024
num_bars = int(total_size / chunk_size)

file_name = os.path.basename(url)

with open(file_name, mode='wb') as f:
    for data in tqdm(rs.iter_content(chunk_size), total=num_bars, unit='KB', file=sys.stdout):
        f.write(data)

Проверяем размер файла

file_data = open(file_name, mode='rb').read()
print('File data size:', sizeof_fmt(len(file_data)))

Выглядит так:

введите сюда описание изображения

4
  • Все прекрасно работает. Но не разобрался как сделать вывод скорости в Мб/с
    – KaKAbik07
    18 фев в 11:18
  • @KaKAbik07, в unit='KB' укажите MB
    – gil9red
    18 фев в 12:06
  • Вот тут какая-то ошибочка. Я сижу на самом слабом wifi'е (20мбит/сек) и мне отображается что у меня 1200-330 MB/s
    – KaKAbik07
    18 фев в 14:50
  • @KaKAbik07, подправьте еще тут: chunk_size = 1024 * 1024
    – gil9red
    18 фев в 16:37
1

Я когда-то давно занимался подобной задачей потоковой загрузки данных с отображением процесса и даже реализовал целый класс для выполнения данной задачи.

import datetime

class ProgressBar:
    """
    Класс ProgressBar реализует простой и удобный интерфейс визуализации динамической загрузки данных
    """
    progress_element = ['', '▏', '▎', '▍', '▌', '▋', '▊', '▉', '█']
    
    def __init__(self, length_data, chunk_size=2**10*10, length_bar=30, counts=0):
        """
        :param length_data: общее количество байт которое необходимо загрузить
        :param chunk_size: размер в байтах одного пакета данных
        :param length_bar: Длинна прогрес бара в символах
        :param counts: количество уже загруженых пакетов
        """
        self.start_time = datetime.datetime.now()
        self.length_data = self.floor(int(length_data)/int(chunk_size))
        self.length_bar = int(length_bar)
        self.chunk = Size(chunk_size)
        self.counts = int(counts) // chunk_size
        self.load_progress = []
    
    def update(self):
        """
        Функция обновления визуальной составляющей прогресс бара
        :return: None
        """
        self.counts += 1
        wasted_time = actual_time(datetime.datetime.now() - self.start_time)
        args = (self.__bar(), self.__indicator(), round(self.__percent()), wasted_time, str(self.__speed()), self.__end_time())
        string = "\r|{}| {} [{:>3}%] in {} ({:>5} Mb/s, eta: {})"
        print(string.format(*args), end='')
    
    def __speed(self):
        """
        Функция вычисляет скорость загрузки пакетов за единицу времени
        :return: Количество байт которые программа получит при текущей скорости загрузки за одну секунду
        """
        self.load_progress.append(datetime.datetime.now())
        if len(self.load_progress) > 250:
            del self.load_progress[0]
        elif len(self.load_progress) < 10:
            return 0
        package_time = (self.load_progress[-1] - self.load_progress[0]).total_seconds()
        package_size = int(self.chunk) * len(self.load_progress)
        try:
            speed = Size(package_size / package_time)
        except ZeroDivisionError:
            speed = 0
        return speed
    
    def __bar(self):
        """
        Функция построения самого прогрес бара
        :return: прогрес бар
        """
        advance = int(self.interp(self.counts, [0, self.length_data], [0, self.length_bar * len(self.progress_element)]))
        finished, during = advance // len(self.progress_element), advance % len(self.progress_element)
        string = finished * self.progress_element[-1] + self.progress_element[during]
        bar = string + ' ' * (self.length_bar - len(string))
        return bar
    
    def __indicator(self):
        """
        Функция для отображения отношения количества загруженых пакетов к общему количеству
        :return: строка соотношения пакетов
        """
        sample = "{:<" + str(len(str(self.length_data))) + "}/" + str(self.length_data)
        return sample.format(self.counts)
    
    def __percent(self):
        """
        Определяет сколько процентов данных уже загружено
        :return: проценты float
        """
        return self.interp(self.counts, [0, self.length_data], [0, 100])
    
    def __end_time(self):
        """
        Функция на основе предыдущих данных вычесляет оставшиеся время до конца загрузки
        :return: Возвращает время до окончания загрузки в формате hh:mm:ss
        """
        percent = self.__percent() * 0.01
        if percent == 0:
            return 0
        load_time = datetime.datetime.now() - self.start_time
        return actual_time((load_time / percent) - load_time)
    
    @staticmethod
    def print(*args, **kwargs):
        """
        Модифицированая функция print, что бы не конфликтовать с активным прогрес баром
        """
        args = ('\r' + str(args[0]),) + args[1:]
        print(*args, **kwargs)
    
    @staticmethod
    def interp(x, xp, fp):  # ранее использовалась numpy функция поэтому здесь такие странные аргументы которые почти не используются
        return fp[1] * x / xp[1]
    
    @staticmethod
    def floor(num):
        """
        Округление в большую сторону
        :param num: число которое нужно округлить
        :return: округлённое число
        """
        return num if num % 1 == 0 else int(num + 1)
    
    def __enter__(self):
        return self
    
    def __exit__(self, exc_type, exc_value, traceback):
        print()
    
    def __iter__(self, *args):
        self.iter = 0
        return self
    
    def __next__(self):
        if self.iter < self.length_data:
            self.iter += 1
            return self.iter
        else:
            raise StopIteration
    
    def __del__(self):
        if self.counts != self.length_data:
            raise BufferError("Файл загружен не полностью!")

class Size:
    """
    Своего рода небольшой костыль предназначен для динамической конвертации байт в другие размерности
    """

    def __init__(self, B=None):
        """
        :param B: количество байт
        """
        self.B = None
        self.KB = None
        self.MB = None
        self.GB = None
        self.__byte(B)


    def __byte(self, memory):
        self.B = memory
        self.KB = self.__rounds(memory / (2 ** 10))
        self.MB = self.__rounds(memory / (2 ** 20))
        self.GB = self.__rounds(memory / (2 ** 30))

    @staticmethod
    def __rounds(numb, floor=3):
        if type(numb) == int:
            return numb
        return round(numb, floor)

    def __str__(self):
        return self.__rounds(self.MB, 1).__str__()

    def __int__(self):
        return self.__rounds(self.B, 1).__int__()

    def __truediv__(self, other):
        return round(self.B / other.B)

def actual_time(time) -> str:
    """
    Приводит время к единому стандарту
    :param time: время datetime/timedelta
    :return: возвращает стандартизированую строку типа hh:mm:ss
    """
    return str(time).split('.')[0]

Код выдрал из старого своего пет-проекта поэтому что-то может с первого раза не запуститься хотя я постарался обрезать все зависимости на другие файлы. Ну и небольшой пример применения:

import requests

link = "your_link"
name_file = "your_file_name"
response = requests.get(link, stream=True)

print(f"{name_file} начинает загрузку...")
with open(name_file, "wb") as file:
    with ProgressBar(length_data=response.headers['Content-Length']) as progress:
        for chunk in response.iter_content(chunk_size=(2**10)*10):
            if chunk:
                file.write(chunk)
                progress.update()
4
  • Выглядит как целый модуль, но к сожалению возникли ошибки. "Size" is not defined Ln 19, "Size" is not defined Ln 47, "interp" is not defined Ln 57, "interp" is not defined Ln 76
    – KaKAbik07
    18 фев в 11:09
  • @KaKAbik07, впрочем-то, я модуль тогда и писал, как можно заметить даже с комментариями заморочился)). А Size это небольшой костыль для корректности отображения данных. Ответ поправил теперь всё должно работать
    – kristal
    18 фев в 13:17
  • Да, все работает. Единственное, что очень быстро изменяется показатель скорости, ну это уже больше я придираюсь к мелочам. А есть возможность конвертации скорости например в GB? Или в те же KB?
    – KaKAbik07
    18 фев в 14:55
  • Для этого уже есть готовый инструментарий, однако такой реализации у него нет, вы можете попробовать сами реализовать нужный вам функционал. На счёт прыгающей скорости вы правы. Её измерение происходит посредством хранения в списке времени последних пришедших 30 пакетов и разницы между последним и текущим временем и напрямую зависит от размера текущего chunk_size где я обычно указываю 10240 байт, а не 1000 как вышло сейчас. Я подправил свой ответ ещё раз, теперь скорость загрузки должна отображаться корректней, обратите внимание что я не реализую выбор размерности скорости загрузки.
    – kristal
    18 фев в 16:54

Ваш ответ

Нажимая на кнопку «Отправить ответ», вы соглашаетесь с нашими условиями использования и подтверждаете, что прочитали и поняли наши политику конфиденциальности и нормы поведения.

Всё ещё ищете ответ? Посмотрите другие вопросы с метками или задайте свой вопрос.