

### 現状

Pythonのライブラリeelを使用してデスクトップアプリを作成してます。  
eelはelectronライクでバックエンドをPython、フロントエンドをhtml, css, JavaScriptで作成出来ます。  
フロント側にプログレスバーの機能を実装したく質問させて頂きます。

下記の投稿と(当質問における)コメント欄のアドバイスでPyTubeに実装されているプログレスバーを表示させる事が出来ました。

[Is there a progress bar feature for
pytube?](https://stackoverflow.com/questions/55882563/is-there-a-progress-bar-
feature-for-pytube)

* * *

ここからさらにプログレスバーの値をフロント側に送信出来るようにしたく、[How to combine pytube and tkinter label to
show progress?](https://stackoverflow.com/questions/49477256/how-to-combine-
pytube-and-tkinter-label-to-show-progress) の内容を元にコードを書きました。  
その際、上記で使用したPyTubeのメソッド`on_progress_callback=on_progress`は用いず、`register_on_progress_callback(self.show_progress_bar)`を記述し（show_progress_barは自分で作成した関数）  
おそらくYouTubeの動画をダウンロード中に登録した関数を呼び出せる機能だと思う。  
`self.progress`にファイルのプログレス状態が格納される事になっているのだと思うのですが、どうやって取得すればいいか分らないです。

おそらく非同期処理で一定の間隔で動画のダウンロードが終了するまでアクセスするコードを書く必要があると思うのですが、どうやって組み立てていけば分らずここで躓いています。

### 作成したコード

[code]

    from pytube import YouTube
    #from pytube.cli import on_progress
    import os
    
    class Downloader():
        """docstring for Youtubehq"""
        def __init__(self, url):
            #yt = YouTube(url, on_progress_callback=on_progress)
            yt = YouTube(url)
            self.title = yt.title
            self.alltag = yt.streams
            self.yt = yt
            print(self.title)
    
    ####省略（画質・音質がよいitagを選択self.vieo_hq, self.audio_hqの取得をする。）#####
    
    
    def show_progress_bar(self, stream=None, chunk=None, file_handle=None, bytes_remaining=None):
            self.progress = str(int(100 - (100*(bytes_remaining/self.MaxfileSize))))
    
    
        def download_video_audio(self):
            yt = self.yt
            yt.register_on_progress_callback(self.show_progress_bar)
            itag_v = self.video_hq['itag']
            
            itag_a = self.audio_hq['itag']
            
            if not os.path.exists('video/original'):
                os.makedirs('video/original')
            yt.streams.get_by_itag(itag_v).download('video/original')
    
            if not os.path.exists('audio/webm'):
                os.makedirs('audio/webm')
            yt.streams.get_by_itag(itag_a).download('audio/webm')
    
    
    Downloader = Downloader('https://www.youtube.com/watch?v=3hPI3xjsNKE')
    ###省略（画質・音質選択処理）
    Downloader.download_video_audio()
    
[/code]

### 追記 自分なりに非同期化してみた。

イメージはリストに格納されたurlを非同期にしてさらにダウンロード最中に別のタスクで`self.progress`にアクセスする事でプログレスバーの進捗を取得出来ると考えた。  
プログレスバーは動画が複数の場合は別々で取得して、必要であればそれを足して総合的な進捗を表示出来るようにしたいです。

[code]

    from pytube import YouTube
    import os
    import asyncio
    
    class Downloader():
        """docstring for Youtubehq"""
        def __init__(self):
            self.__gtask = []
            self.working = True
    
    """
            self.makelist()
            self.select_mobile_v()
            self.select_hq_a()
            の処理が入る。
            ここで高画質・音質を選択する。self.video_hq, self.audio_hq
            中身を取得する。
    """
    
    
    def download_video_audio(self):
            yt = self.yt
            yt.register_on_progress_callback(self.show_progress_bar)
            itag_v = self.video_hq['itag']
            
            itag_a = self.audio_hq['itag']
            
            if not os.path.exists('video/original'):
                os.makedirs('video/original')
            yt.streams.get_by_itag(itag_v).download('video/original')
    
            if not os.path.exists('audio/webm'):
                os.makedirs('audio/webm')
            yt.streams.get_by_itag(itag_a).download('audio/webm')
    
    def progress_value(self):
            while self.working:
                print(self.progress)
        
    
    
        def download(self, url):
            
            yt = YouTube(url)
            self.title = yt.title
            self.alltag = yt.streams
            self.yt = yt
            print(self.title)
    
            self.makelist()
            self.select_mobile_v()
            self.select_hq_a()
            self.download_video_audio()
            self.working = False
    
        
        def async_dl(self, urls):
            async def async_wrap(loop, urls):
                sem = asyncio.Semaphore(20)
                async def async_ex(i):
                    async with sem:
                        return await loop.run_in_executor(None, self.download, url[i])
                tasks = [async_ex(i) for i in range(len(urls))]
                tasks =+ [await loop.run_in_executor(None, self.progress_value)]
                self.__gtask = asyncio.gather(*tasks)
                return await self.__gtask
                loop.run_until_complete(self.__gtask)
            
            loop = asyncio.get_event_loop()
            return async_wrap(loop, urls)
    
    
    
                
    Downloader = Downloader()
    Downloader.async_dl(['https://www.youtube.com/watch?v=qZj8lrSZHsE'])
    
[/code]

### エラー

[code]

    yt_download.py:203: RuntimeWarning: coroutine 'Downloader.async_dl.<locals>.async_wrap' was never awaited
      Downloader.async_dl(['https://www.youtube.com/watch?v=qZj8lrSZHsE'])
    RuntimeWarning: Enable tracemalloc to get the object allocation traceback
    
[/code]


CC BY-SA 4.0
2020-09-27T11:34:19.813
70754
2020-09-29T01:51:20.207
2020-09-29T01:51:20.207
22565
22565
question
python python3 非同期
PyTubeのプログレスバーの値を取得してフロントエンド送信出来る状態にしたい