طراحی نوارهای پیشروی در پایتون با Tqdm — راهنمای کاربردی

1import time
2import sys
3from tqdm import trange
4
5
6def do_something():
7    time.sleep(1)
8
9def do_another_something():
10    time.sleep(1)
11
12
13for i in trange(10, file=sys.stdout, desc='outer loop'):
14    do_something()
15
16    for j in trange(100,file=sys.stdout, leave=False, unit_scale=True, desc='inner loop'):
17        do_another_something()

1import time
2import sys
3from tqdm import tqdm
4
5
6def do_something():
7    time.sleep(1)
8
9
10with tqdm(total=100, file=sys.stdout) as pbar:
11    for i in range(10):
12        do_something()
13        # Manually update the progress bar, useful for streams such as reading files.
14        pbar.update(10)
15        # Updates in increments of 10 stops at 100

1$ pip install requests validators
2#  Copyright 2019 tiptapcode Authors. All Rights Reserved.
3#
4#  Licensed under the Apache License, Version 2.0 (the "License");
5#  you may not use this file except in compliance with the License.
6#  You may obtain a copy of the License at
7#
8#       http://www.apache.org/licenses/LICENSE-2.0
9#
10#  Unless required by applicable law or agreed to in writing, software
11#  distributed under the License is distributed on an "AS IS" BASIS,
12#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13#  See the License for the specific language governing permissions and
14#  limitations under the License.
15
16# -*- coding: utf-8 -*-
17import os
18import sys
19import tqdm
20import requests
21import validators
22
23
24class FileDownloader(object):
25
26    def get_url_filename(self, url):
27        """
28        Discover file name from HTTP URL, If none is discovered derive name from http redirect HTTP content header Location
29        :param url: Url link to file to download
30        :type url: str
31        :return: Base filename
32        :rtype: str
33        """
34        try:
35            if not validators.url(url):
36                raise ValueError('Invalid url')
37            filename = os.path.basename(url)
38            basename, ext = os.path.splitext(filename)
39            if ext:
40                return filename
41            header = requests.head(url, allow_redirects=False).headers
42            return os.path.basename(header.get('Location')) if 'Location' in header else filename
43        except requests.exceptions.HTTPError as errh:
44            print("Http Error:", errh)
45            raise errh
46        except requests.exceptions.ConnectionError as errc:
47            print("Error Connecting:", errc)
48            raise errc
49        except requests.exceptions.Timeout as errt:
50            print("Timeout Error:", errt)
51            raise errt
52        except requests.exceptions.RequestException as err:
53            print("OOps: Something Else", err)
54            raise err
55
56    def download_file(self, url, filename=None, target_dir=None):
57        """
58        Stream downloads files via HTTP
59        :param url: Url link to file to download
60        :type url: str
61        :param filename: filename overrides filename defined in Url param
62        :type filename: str
63        :param target_dir: target destination directory to download file to
64        :type target_dir: str
65        :return: Absolute path to target destination where file has been downloaded to
66        :rtype: str
67        """
68        if target_dir and not os.path.isdir(target_dir):
69            raise ValueError('Invalid target_dir={} specified'.format(target_dir))
70        local_filename = self.get_url_filename(url) if not filename else filename
71
72        req = requests.get(url, stream=True)
73        file_size = int(req.headers['Content-Length'])
74        chunk_size = 1024  # 1 MB
75        num_bars = int(file_size / chunk_size)
76
77        base_path = os.path.abspath(os.path.dirname(__file__))
78        target_dest_dir = os.path.join(base_path, local_filename) if not target_dir else os.path.join(target_dir, local_filename)
79        with open(target_dest_dir, 'wb') as fp:
80            for chunk in tqdm.tqdm(req.iter_content(chunk_size=chunk_size), total=num_bars, unit='KB', desc=local_filename, leave=True, file=sys.stdout):
81                fp.write(chunk)
82
83        return target_dest_dir
84
85
86if __name__== "__main__":
87
88    links = ['https://nodejs.org/dist/v12.13.1/node-v12.13.1.pkg', 'https://aka.ms/windev_VM_virtualbox']
89
90    downloader = FileDownloader()
91
92    for url in links:
93        downloader.download_file(url)

1import time
2
3from random import randrange
4from multiprocessing.pool import ThreadPool
5
6from tqdm import tqdm
7
8
9def func_call(position, total):
10    text = 'progressbar #{position}'.format(position=position)
11    with  tqdm(total=total, position=position, desc=text) as progress:
12        for _ in range(0, total, 5):
13            progress.update(5)
14            time.sleep(randrange(3))
15
16
17pool = ThreadPool(10)
18tasks = range(5)
19for i, url in enumerate(tasks, 1):
20    pool.apply_async(func_call, args=(i, 100))
21pool.close()
22pool.join()

1pip install requests tqdm pandas
2import time
3import pandas as pd
4import requests
5
6from tqdm import tqdm
7
8
9def percent_off(product_price, discount):
10    try:
11        discount = float(discount)
12        if discount < 0  and discount > 100:
13            raise ValueError('discout amount should be between 1 and 100%')
14        value = (product_price - (product_price * (discount / 100.0)))
15        time.sleep(0.0001)
16        return value
17    except ValueError as e:
18        print('invalid product_price or discount amount', e)
19        raise e
20
21def appy_discount(perentage):
22
23    df = pd.DataFrame(pd.read_json('products.json'))
24
25    df.insert(4, 'discount', 0)
26
27    tqdm.pandas(desc='apply_{}_percent_off'.format(perentage))
28
29    df['discount'] = df['price'].progress_apply(lambda x: percent_off(x, perentage))
30
31    return df
32
33
34# Downlaod sample best buy products json file
35# It sucks right that you do not see a progress bar while downloadng this large file below
36r = requests.get('https://github.com/BestBuyAPIs/open-data-set/raw/master/products.json', allow_redirects=True)
37open('products.json', 'wb').write(r.content)
38
39# How about now imagine performing a large pandas dataframe calculation
40df = appy_discount(5)
41
42df # use this to a nice html output in jupyter notebooks else print to sysout

1from tqdm import trange
2from colorama import Fore
3
4# Cross-platform colored terminal text.
5color_bars = [Fore.BLACK,
6    Fore.RED,
7    Fore.GREEN,
8    Fore.YELLOW,
9    Fore.BLUE,
10    Fore.MAGENTA,
11    Fore.CYAN,
12    Fore.WHITE]
13
14for color in color_bars:
15    for i in trange(int(7e7),
16                    bar_format="{l_bar}%s{bar}%s{r_bar}" % (color, Fore.RESET)):
17        pass

1#  Copyright 2019 tiptapcode Authors. All Rights Reserved.
2#
3#  Licensed under the Apache License, Version 2.0 (the "License");
4#  you may not use this file except in compliance with the License.
5#  You may obtain a copy of the License at
6#
7#       http://www.apache.org/licenses/LICENSE-2.0
8#
9#  Unless required by applicable law or agreed to in writing, software
10#  distributed under the License is distributed on an "AS IS" BASIS,
11#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12#  See the License for the specific language governing permissions and
13#  limitations under the License.
14
15import io
16import os
17import sys
18import logging
19import validators
20
21from urllib import request
22from tqdm import tqdm
23
24
25class ProgressBar(tqdm):
26
27    def update_progress(self, block_num=1, block_size=1, total_size=None):
28        if total_size is not None:
29            self.total = total_size
30        self.update(block_num * block_size - self.n)  # will also set self.n = b * bsize
31
32
33class DownloadFileHandler(object):
34
35    @staticmethod
36    def download_file_by_url(url, download_dir=None):
37        if not validators.url(url):
38           raise ValueError('Invalid url := {}'.format(url))
39        if download_dir is not None and not os.path.isdir(download_dir):
40            raise FileNotFoundError('Directory specified := {} does not exist'.format(download_dir))
41        else:
42            download_dir = os.path.abspath(os.path.dirname(__file__))
43        filename = os.path.basename(url)
44        download_destination = os.path.join(download_dir, filename)
45
46        #The magic happens here in order to log to python logger we need to create
47        # A custom logger that channels the output stream to the log stream
48        with ProgressBar(
49                file=TqdmSystemLogger(logger, suppress_new_line=False),
50                unit='B',
51                unit_scale=True,
52                miniters=1,
53                desc=filename
54        ) as progressBar:
55            # request.urlretrieve has an internal callback function that get invoked reporthook
56            # The reporthook argument should be
57            #     a callable that accepts a block number, a read size, and the
58            #     total file size of the URL target. The data argument should be
59            #     valid URL encoded data.
60            #     tqdm uses this data to derive a progress bar as we know the total file size we can estimate ETA
61            request.urlretrieve(url, filename=download_destination, reporthook=progressBar.update_progress, data=None)
62
63        return download_destination
64
65
66class SystemLogger(object):
67
68    def __init__(self):
69        pass
70
71    @staticmethod
72    def get_logger(name, level=None):
73
74        root_logger = logging.getLogger(name)
75        root_logger.setLevel(level if level else logging.INFO)
76
77        # An attempt to replace logger output as to print on same line may not work on some terminals
78        # only applicable to logging to sys.stdout
79        # formatter = logging.Formatter('\x1b[80D\x1b[1A\x1b[K%(message)s')
80
81        formatter = logging.Formatter(fmt='%(levelname)s:%(name)s: %(message)s (%(asctime)s; %(filename)s:%(lineno)d)', datefmt="%d-%m-%YT%H:%M:%S%z")
82
83        handler_stdout = logging.StreamHandler(sys.stdout)
84        handler_stdout.setFormatter(formatter)
85        handler_stdout.setLevel(logging.WARNING)
86        handler_stdout.addFilter(type('', (logging.Filter,), {'filter': staticmethod(lambda r: r.levelno <= logging.INFO)}))
87
88        handler_stdout.flush = sys.stdout.flush
89
90        root_logger.addHandler(handler_stdout)
91
92        handler_stderr = logging.StreamHandler(sys.stderr)
93        handler_stderr.setFormatter(formatter)
94        handler_stderr.setLevel(logging.WARNING)
95
96        handler_stderr.flush = sys.stderr.flush
97
98        root_logger.addHandler(handler_stderr)
99
100        return root_logger
101
102
103class TqdmSystemLogger(io.StringIO):
104
105    def __init__(self, logger, suppress_new_line=True):
106        super(TqdmSystemLogger, self).__init__()
107        self.logger = logger
108        self.buf = ''
109        # only tested and works inside pycharm terminal logging to sys.stdout
110        # by replacing default terminator newline we force logger to override the output on screen
111        # thus giving us a progress depiction in a single line instead of multiple lines
112        if suppress_new_line:
113            for handler in self.logger.handlers:
114                if isinstance(handler, logging.StreamHandler):
115                    handler.terminator = ""
116
117    def write(self, buf):
118        self.buf = buf.strip('\r\n\t ')
119
120    def flush(self):
121        self.logger.log(self.logger.level, '\r' + self.buf)
122
123
124try:
125    logger = SystemLogger.get_logger('DownloadFileHandler', level=logging.WARNING)
126    # Download a file to this scripts relative directory and log output to python logger sysout
127    DownloadFileHandler.download_file_by_url('https://nodejs.org/dist/v12.13.1/node-v12.13.1-darwin-x64.tar.gz')
128except Exception as e:
129    print(str(e))

1import sys
2import subprocess
3
4from tqdm import tqdm
5
6
7def create_test_bash_script():
8    """
9    Create a bash script that generates numbers 1 to 1000000
10    This is just for illustration purpose to simulate a long running bash command
11    """
12    with open('hello', 'w') as bash_file:
13        bash_file.write('''\
14    #!/bin/bash
15    # Tested using bash version 4.1.5
16    for ((i=1;i<=1000000;i++));
17    do
18        # your-unix-command-here
19        echo $i
20    done
21    ''')
22
23
24def run_task(cmd):
25
26    try:
27        # create a default tqdm progress bar object, unit='B' definnes a String that will be used to define the unit of each iteration in our case bytes
28        with tqdm(unit='B', unit_scale=True, miniters=1, desc="run_task={}".format(cmd)) as t:
29            # subprocess.PIPE gets the output of the child process
30            process = subprocess.Popen(cmd, shell=True, bufsize=1, universal_newlines=True, stdout=subprocess.PIPE,
31                                       stderr=subprocess.PIPE)
32
33            # print subprocess output line-by-line as soon as its stdout buffer is flushed in Python 3:
34            for line in process.stdout:
35                # Update the progress, since we do not have a predefined iterator
36                # tqdm doesnt know before hand when to end and cant generate a progress bar
37                # hence elapsed time will be shown, this is good enough as we know
38                # something is in progress
39                t.update()
40                # forces stdout to "flush" the buffer
41                sys.stdout.flush()
42
43            # We explicitly close stdout
44            process.stdout.close()
45
46            # wait for the return code
47            return_code = process.wait()
48
49            # if return code is not 0 this means our script errored out
50            if return_code != 0:
51                raise subprocess.CalledProcessError(return_code, cmd)
52
53    except subprocess.CalledProcessError as e:
54        sys.stderr.write(
55            "common::run_command() : [ERROR]: output = {}, error code = {}\n".format(e.output, e.returncode))
56
57
58create_test_bash_script()
59
60# run your terminal command using below
61run_task('chmod 755 hello && ./hello')
62
63run_task('xx*3238') # this will fail not a valid command