Я когда-то давно занимался подобной задачей потоковой загрузки данных с отображением процесса и даже реализовал целый класс для выполнения данной задачи.
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()
human_byte_size
в 19 строке