python - Download file with Kivy app without locking event loop -
here's kivy application has button , progress bar. when button pressed, zip file downloaded web , unzipped. progress bar advances mark download's progress.
the problem is, downloading locks kivy event loop, freezing application during download. how can download , unzip file in background?
from __future__ import division import os import requests import zipfile kivy.app import app kivy.lang import builder kivy.uix.boxlayout import boxlayout zip_url = 'https://www.python.org/ftp/python/3.5.1/python-3.5.1-embed-win32.zip' zip_filename = 'python351.zip' kv_string = """ <rootwidget> boxlayout: orientation: "vertical" button: id: download_button text: "download content" on_press: self.parent.parent.download_content() progressbar: id: download_progress_bar max: 1 value: 0.1 """ builder.load_string(kv_string) class rootwidget(boxlayout): def __init__(self, **kwargs): super(rootwidget, self).__init__(**kwargs) def download_content(self): self.ids["download_button"].disabled = true total_size_kib = 6023182 / 1024 # debug: hardcoded print("downloading file: %s" % zip_url) open(zip_filename, 'wb') handle: response = requests.get(zip_url, stream=true) if not response.ok: print("something went wrong.") return current_size_kib = 0 block in response.iter_content(1024): percent_downloaded = current_size_kib / total_size_kib if not current_size_kib % 100: print("%.2f%% downloaded far" % (percent_downloaded * 100)) self.ids['download_progress_bar'].value = percent_downloaded handle.write(block) current_size_kib += 1 self.unzip_content() def unzip_content(self): print("unzipping file") fh = open(zip_filename, 'rb') z = zipfile.zipfile(fh) zip_extract_folder = zip_filename + '_extracted' if not os.path.exists(zip_extract_folder): os.makedirs(zip_extract_folder) z.extractall(zip_extract_folder) fh.close() os.remove(zip_filename) print("done") class myapp(app): def build(self): return rootwidget() if __name__ == '__main__': myapp().run()
first, @nykakin suggests, make download asynchronous, use kivy.network.urlrequest.urlrequest
:
def download_content(self): self.ids["download_button"].disabled = true req = urlrequest(zip_url, on_progress=self.update_progress, chunk_size=1024, on_success=self.unzip_content, file_path=zip_filename) def update_progress(self, request, current_size, total_size): self.ids['download_progress_bar'].value = current_size / total_size
second, @keyweeusr suggests, change unzipping method doesn't block event loop, using threading
module:
def unzip_content(self, req, result): threading.thread(target=self.unzip_thread).start() def unzip_thread(self): # ... (same unzip_content method in question)
here's full new version:
from __future__ import division import os import zipfile import threading import time kivy.app import app kivy.lang import builder kivy.uix.boxlayout import boxlayout kivy.network.urlrequest import urlrequest zip_url = 'https://www.python.org/ftp/python/3.5.1/python-3.5.1-embed-win32.zip' zip_filename = 'python351.zip' kv_string = """ <rootwidget> boxlayout: orientation: "vertical" button: id: download_button text: "download content" on_press: self.parent.parent.download_content() progressbar: id: download_progress_bar max: 1 value: 0 """ builder.load_string(kv_string) class rootwidget(boxlayout): stop = threading.event() def __init__(self, **kwargs): super(rootwidget, self).__init__(**kwargs) def download_content(self): self.ids["download_button"].disabled = true req = urlrequest(zip_url, on_progress=self.update_progress, chunk_size=1024, on_success=self.unzip_content, file_path=zip_filename) def update_progress(self, request, current_size, total_size): self.ids['download_progress_bar'].value = current_size / total_size def unzip_content(self, req, result): threading.thread(target=self.unzip_thread).start() def unzip_thread(self): print("unzipping file") fh = open(zip_filename, 'rb') z = zipfile.zipfile(fh) zip_extract_folder = zip_filename + '_extracted' if not os.path.exists(zip_extract_folder): os.makedirs(zip_extract_folder) z.extractall(zip_extract_folder) fh.close() os.remove(zip_filename) time.sleep(4) # debug: stretch out unzip method test threading print("done") class myapp(app): def on_stop(self): # kivy event loop stop, set stop signal; # otherwise app window close, python process # keep running until secondary threads exit. self.root.stop.set() def build(self): return rootwidget() if __name__ == '__main__': myapp().run()
Comments
Post a Comment