Нет, нет способа обернуть индикатор выполнения кодом, к которому у вас нет доступа.
Кроме того, вы можете использовать основанный на итерации интерфейс для tqdm только когда вы зацикливаетесь на итерируемом, которого у вас нет. Поэтому вы должны использовать интерфейс update:
with tqdm(total=100) as pbar:
for i in range(10):
pbar.update(10)
Вопрос в том, как заставить PyYAML называть это pbar.update?
В идеале вы хотите найти место для подключения процесса загрузки, где вы можете позвонить pbar.update. Если это невозможно, вам придется сделать что-то некрасивое (например, fork PyYAML и добавить к его API, или сделать то же самое во время выполнения, запустив его с помощью обезьянки) или переключиться на другую библиотеку. Но это должно быть возможно.
Очевидным вариантом является создание собственного подкласса PyYAML.Loader. Документы для PyYAML объясняют API для этого класса, так что вы можете переопределить любой из методов там, чтобы выдать некоторый прогресс, а затем super в базовый класс.
Но, к сожалению, никто из них не выглядит таким многообещающим. Конечно, вы можете вызываться один раз для токена, один раз для события или один раз для узла, но не зная, сколько там токенов, событий или узлов, это не позволит вам показать, насколько далеко вы находитесь в файле. Если вам нужен неопределенный счетчик прогресса, это нормально, но если вы можете получить реальный прогресс с оценкой продолжительности и т. Д., Это всегда лучше.
Одна вещь, которую вы могли бы сделать, это сделать ваш Loader вызов подкласса tell на stream, чтобы выяснить, сколько байт вы прочитали до сих пор.
У меня нет PyYAML на этом компьютере, и документы довольно запутанные, поэтому вам, вероятно, придется немного поэкспериментировать, но это должно быть примерно так:
class ProgressLoader(yaml.CLoader):
def __init__(self, stream, callback):
super().__init__(stream)
# __ because who knows what names the base class is using?
self.__stream = stream
self.__pos = 0
self.__callback = callback
def get_token(self):
result = super().get_token()
pos = self.__stream.tell()
self.__callback(pos - self.__pos)
self.__pos = pos
return result
Но тогда я не уверен, как заставить PyYAML передать ваш обратный вызов в конструктор ProgressLoader, поэтому вам придется сделать что-то вроде этого:
with open(file_name, 'r') as stream:
size = os.stat(stream.fileno()).st_size
with tqdm(total=size) as progress:
factory = lambda stream: ProgressLoader(stream, progress.update)
data = yaml.load(stream, Loader=factory)
Но как только мы все равно перейдем к файлу, вероятно, легче не облажаться с запутанными в замешательстве типами загрузчиков, а вместо этого просто написать оболочку для файла.
Документы для файловых объектов довольно плотные , но, по крайней мере, они ясны - и фактическая работа довольно проста:
class ProgressFileWrapper(io.TextIOBase):
def __init__(self, file, callback):
self.file = file
self.callback = callback
def read(self, size=-1):
buf = self.file.read(size)
if buf:
self.callback(len(buf))
return buf
def readline(self, size=-1):
buf = self.file.readline(size)
if buf:
self.callback(len(buf))
return buf
Сейчас:
with open(file_name, 'r') as stream:
size = os.stat(stream.fileno()).st_size
with tqdm(total=size) as progress:
wrapper = ProgressFileWrapper(stream, progress.update)
data = yaml.load(wrapper, Loader=Loader)
Конечно, это не идеально. Здесь мы предполагаем, что вся работа заключается в чтении файла с диска, а не в его анализе. Это, вероятно, достаточно близко к истине, что нам это сойдет с рук, но если это не так, у вас будет один из тех индикаторов выполнения, который достигает почти 100%, а затем просто бесполезно остается там долгое время. 1
1. Это не только ужасно раздражает, но и настолько прочно связано с Windows и другими продуктами Microsoft, что они могут подать в суд на вас за кражу внешнего вида. :)