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__":...
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.
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Get 100 minutes of automation test minutes FREE!!