Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use __download_file method in toolium

Best Python code snippet using toolium_python

downloader.py

Source:downloader.py

1from urllib.request import urlopen, Request2from multiprocessing.pool import ThreadPool3from multiprocessing import Lock4import time5from threading import Thread6class Downloader:7    __block_size = 81928    __threads = 89    __url = None10    __download_path = None11    __content_type = None12    __downloaded_size = 013    __file_size = 014    __file_name = None15    __download_file = None16    __pause_able = False17    __connection = None18    __speed = 019    __percent = 020    __remaining_time = 021    update_meter = 0.522    __headers = None23    __lock = Lock()24    __downloading = False25    __remaining_partitions = list()26    def __init__(self, url, download_path, threads=8, block_size=8196, limited_speed=float('inf')):27        """28            Args:29                url(str): Download file URL30                download_path(str): Download path of file31                threads(int): Threads count32                block_size(int) : Download size in each block33                limited_speed(int|float) : Speed limiter for download speed34        """35        self.__url = url36        self.__download_path = download_path37        self.thread_pool = ThreadPool(self.__threads)38        self.__threads = threads39        self.__block_size = block_size40        self.limited_speed = limited_speed41        self.get_details()42    def resume(self, remaining_partitions):43        self.__remaining_partitions = remaining_partitions44        self.download()45    def __speed_meter(self):46        previous_downloaded_size = self.__downloaded_size47        while self.__downloading:48            self.__speed = (self.__downloaded_size - previous_downloaded_size) / self.update_meter49            self.__percent = float(self.__downloaded_size * 100 / self.__file_size)50            self.__remaining_time = (self.__file_size - self.__downloaded_size) / self.__speed if self.__speed != 0 else float('inf')51            previous_downloaded_size = self.__downloaded_size52            time.sleep(self.update_meter)53    def get_details(self):54        u = urlopen(self.__url)55        meta = u.info()56        self.__file_name = self.__url.split('/')[-1]57        self.__file_size = int(meta.get("Content-Length"))58        self.__content_type = meta.get("Content-Type")59        self.__connection = meta.get("Connection")60        self.__headers = meta61        if meta.get('Accept-Ranges') == 'bytes':62            self.__pause_able = True63    def pause(self):64        self.__downloading = False65    def download(self):66        self.__downloading = True67        self.__download_file = open(self.__download_path, 'w+b')68        self.__download_file.seek(self.__file_size - 1)69        self.__download_file.write(b'\0')70        self.__download_file.seek(0)71        Thread(target=self.__speed_meter).start()72        if self.__pause_able:73            if not self.__remaining_partitions:74                i = 075                partitions = list()76                while True:77                    if self.__block_size * (i + 1) > self.__file_size:78                        if self.__file_size % self.__block_size != 0:79                            partitions.append([i * self.__block_size, self.__file_size])80                        break81                    partitions.append([i * self.__block_size, (i + 1) * self.__block_size])82                    i += 183            else:84                partitions = self.__remaining_partitions85            self.thread_pool.map(self.threaded_download, partitions)86        else:87            self.single_thread_download()88        self.__download_file.close()89        self.__downloading = False90    def threaded_download(self, download_partition):91        beginning_pointer, ending_pointer = download_partition92        if self.__downloading:93            self.__lock.acquire()94            while self.__speed > self.limited_speed:95                time.sleep(0.1)96            self.__lock.release()97            request = Request(self.__url)98            request.add_header("Range", f"bytes={beginning_pointer}-{ending_pointer}")99            with urlopen(request) as response:100                buffer = response.read(self.__block_size)101                self.write_file(buffer, beginning_pointer)102        else:103            self.__lock.acquire()104            self.__remaining_partitions.append(download_partition)105            self.__lock.release()106    def single_thread_download(self):107        request = Request(self.__url)108        response = urlopen(request)109        while buffer := response.read(self.__block_size):110            while self.__speed > self.limited_speed:111                time.sleep(0.1)112            self.__download_file.write(buffer)113            self.__downloaded_size += len(buffer)114    def write_file(self, buffer, pointer):115        self.__download_file.seek(pointer)116        self.__download_file.write(buffer)117        self.__lock.acquire()118        self.__downloaded_size += len(buffer)119        self.__lock.release()120    # region Attributes121    @property122    def pause_able(self):123        return self.__pause_able124    @property125    def speed(self):126        return self.__speed127    @property128    def percent(self):129        return self.__percent130    @property131    def remaining_time(self):132        return self.__remaining_time133    @property134    def remaining_partitions(self):135        return self.__remaining_partitions136    @property137    def content_type(self):138        return self.__content_type139    @property140    def downloaded_size(self):141        return self.__downloaded_size142    @property143    def file_size(self):144        return self.__file_size145    @property146    def file_name(self):147        return self.__file_name148    @property149    def download_file(self):150        return self.__download_file151    @property152    def block_size(self):153        return self.__block_size154    @property155    def threads(self):156        return self.__threads157    @property158    def url(self):159        return self.__url160    @property161    def download_path(self):162        return self.__download_path163    @property164    def downloading(self):165        return self.__downloading...

downloadhandler.py

Source:downloadhandler.py

...3233    # Method to download Driver of Token34    def download_driver(self, drive_name):35        # try download drive36        download = self.__download_file(self.__drive_url, drive_name)37        return download3839    # Method to download CSP of Token40    def download_csp(self, csp_name):41        # try download CSP42        download = self.__download_file(self.__csp_url, csp_name)43        return download4445    # Method to download Mozilla Firefox46    def download_mozilla(self, firefox_name):47        # try download Mozilla48        download = self.__download_file(self.__firefox_url, firefox_name)49        return download5051    # Method to download Java52    def download_java(self, java_name):53        # try download Java54        download = self.__download_file(self.__java_url, java_name)55        return download5657    # Method to download PjeOffice58    def download_pje_office(self, pje_name):59        # try download PjeOffice60        download = self.__download_file(self.__pje_url, pje_name)61        return download6263    # Method for downloads64    def __download_file(self, url, file_name):65        # Path and name to the file66        file_path = self.__local_path + file_name67        # Try download the file68        try:69            # aliases file url70            with urllib.request.urlopen(url) as file_download:71                # size of the file72                file_size = int(file_download.getheader('Content-Length'))73                # Enable color to windows cmd74                system("color")75                # Progressbar76                bar = progressbar.ProgressBar(0, file_size)77                # Blocksize of download78                blocksize = max(4096, int(file_size) // 100)
...

tf_image_net.py

Source:tf_image_net.py

1import classify_image2import subprocess3import urllib.request4import elice_utils5def main():6    __download_file = "./download.jpg"7    image_url = "http://i.imgur.com/A8eQsll.jpg"8    download_image(image_url, __download_file)9    elice_utils.send_image(__download_file)10    classify_image.main(__download_file)11def download_image(image_url, download_filepath):12    urllib.request.urlretrieve(image_url, download_filepath)13if __name__ == "__main__":...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.