I've been thinking about creating a simple downloading script with wxPython for quite a while. Then I saw someone on StackOverflow asking about how to do it and I decided it was time to figure it out. Downloading a file with Python is trivial. I wrote about a couple of different ways to do that in a previous article. The big question was how to do it in such a way that I could update the UI as the file was downloading. It's actually pretty easy and this article will show you how!
The script in this article will require you to have the 3rd party requests package installed. You will also need wxPython, of course. I'll be using wxPython 2.9 in this article.
I ended up taking a few of my previous articles and combining their code bit by bit until I got what I wanted. I always find diving into the code the quickest way to see how to do things, so let's take a look at the source:
import requests
import os
import wx
import wx.lib.scrolledpanel as scrolled
from threading import Thread
from wx.lib.pubsub import pub
########################################################################
class DownloadThread(Thread):
"""Downloading thread"""
#----------------------------------------------------------------------
def __init__(self, gnum, url, fsize):
"""Constructor"""
Thread.__init__(self)
self.fsize = fsize
self.gnum = gnum
self.url = url
self.start()
#----------------------------------------------------------------------
def run(self):
"""
Run the worker thread
"""
local_fname = os.path.basename(self.url)
count = 1
while True:
if os.path.exists(local_fname):
tmp, ext = os.path.splitext(local_fname)
cnt = "(%s)" % count
local_fname = tmp + cnt + ext
count += 1
else:
break
req = requests.get(self.url, stream=True)
total_size = 0
print local_fname
with open(local_fname, "wb") as fh:
for byte in req.iter_content(chunk_size=1024):
if byte:
fh.write(byte)
fh.flush()
total_size += len(byte)
if total_size < self.fsize:
wx.CallAfter(pub.sendMessage,
"update_%s" % self.gnum,
msg=total_size)
print "DONE!"
wx.CallAfter(pub.sendMessage,
"update_%s" % self.gnum,
msg=self.fsize)
########################################################################
class MyGauge(wx.Gauge):
""""""
#----------------------------------------------------------------------
def __init__(self, parent, range, num):
"""Constructor"""
wx.Gauge.__init__(self, parent, range=range)
pub.subscribe(self.updateProgress, "update_%s" % num)
#----------------------------------------------------------------------
def updateProgress(self, msg):
""""""
self.SetValue(msg)
########################################################################
class MyPanel(scrolled.ScrolledPanel):
""""""
#----------------------------------------------------------------------
def __init__(self, parent):
"""Constructor"""
scrolled.ScrolledPanel.__init__(self, parent)
self.data = []
self.download_number = 1
# create the sizers
self.main_sizer = wx.BoxSizer(wx.VERTICAL)
dl_sizer = wx.BoxSizer(wx.HORIZONTAL)
# create the widgets
lbl = wx.StaticText(self, label="Download URL:")
self.dl_txt = wx.TextCtrl(self)
btn = wx.Button(self, label="Download")
btn.Bind(wx.EVT_BUTTON, self.onDownload)
# layout the widgets
dl_sizer.Add(lbl, 0, wx.ALL|wx.CENTER, 5)
dl_sizer.Add(self.dl_txt, 1, wx.EXPAND|wx.ALL, 5)
dl_sizer.Add(btn, 0, wx.ALL, 5)
self.main_sizer.Add(dl_sizer, 0, wx.EXPAND)
self.SetSizer(self.main_sizer)
self.SetAutoLayout(1)
self.SetupScrolling()
#----------------------------------------------------------------------
def onDownload(self, event):
"""
Update display with downloading gauges
"""
url = self.dl_txt.GetValue()
try:
header = requests.head(url)
fsize = int(header.headers["content-length"]) / 1024
sizer = wx.BoxSizer(wx.HORIZONTAL)
fname = os.path.basename(url)
lbl = wx.StaticText(self, label="Downloading %s" % fname)
gauge = MyGauge(self, fsize, self.download_number)
sizer.Add(lbl, 0, wx.ALL|wx.CENTER, 5)
sizer.Add(gauge, 0, wx.ALL|wx.EXPAND, 5)
self.main_sizer.Add(sizer, 0, wx.EXPAND)
self.Layout()
# start thread
DownloadThread(self.download_number, url, fsize)
self.dl_txt.SetValue("")
self.download_number += 1
except Exception, e:
print "Error: ", e
########################################################################
class DownloaderFrame(wx.Frame):
""""""
#----------------------------------------------------------------------
def __init__(self):
"""Constructor"""
wx.Frame.__init__(self, None, title="Downloader", size=(800, 400))
panel = MyPanel(self)
self.Show()
#----------------------------------------------------------------------
if __name__ == "__main__":
app = wx.App(False)
frame = DownloaderFrame()
app.MainLoop()
Let's take this code a part piece and piece to make it easier to explain. First off, we have to subclass the Thread class as follows:
########################################################################
class DownloadThread(Thread):
"""Downloading thread"""
#----------------------------------------------------------------------
def __init__(self, gnum, url, fsize):
"""Constructor"""
Thread.__init__(self)
self.fsize = fsize
self.gnum = gnum
self.url = url
self.start()
#----------------------------------------------------------------------
def run(self):
"""
Run the worker thread
"""
local_fname = os.path.basename(self.url)
count = 1
while True:
if os.path.exists(local_fname):
tmp, ext = os.path.splitext(local_fname)
cnt = "(%s)" % count
local_fname = tmp + cnt + ext
count += 1
else:
break
req = requests.get(self.url, stream=True)
total_size = 0
print local_fname
with open(local_fname, "wb") as fh:
for byte in req.iter_content(chunk_size=1024):
if byte:
fh.write(byte)
fh.flush()
total_size += 1024
if total_size < self.fsize:
wx.CallAfter(pub.sendMessage,
"update_%s" % self.gnum,
msg=total_size)
print "DONE!"
wx.CallAfter(pub.sendMessage,
"update_%s" % self.gnum,
msg=self.fsize)
We pass in the gauge number (gnum), the url to download and the file size. The reason we pass in the gauge number is that each gauge instance needs to be able to update independently of the others gauges, so we need to keep track of which gauge we need to update. In the run method, we check and see if the filename already exists. If it does, we try appending a number to it and check to see if that file exists, etc. Once we find a filename that does not exist, we continue and tell requests to download the file as a stream. Then we pull down a byte at a time, write it to disk and tell the display to update. Once we drop out of the for loop that is pulling the file down, we send one last update to the gauge to tell it that it is done downloading.
Next up, we have to subclass the wx.Gauge:
########################################################################
class MyGauge(wx.Gauge):
""""""
#----------------------------------------------------------------------
def __init__(self, parent, range, num):
"""Constructor"""
wx.Gauge.__init__(self, parent, range=range)
pub.subscribe(self.updateProgress, "update_%s" % num)
#----------------------------------------------------------------------
def updateProgress(self, msg):
""""""
self.SetValue(msg)
This is pretty simple. We just subclass wx.Gauge and set up a pubsub subscriber so it listens for updates just for itself. Then we add a method called updateProgress that we'll call whenever the listener fires to actually update the gauge widget. Now we're ready to look at the panel code:
########################################################################
class MyPanel(scrolled.ScrolledPanel):
""""""
#----------------------------------------------------------------------
def __init__(self, parent):
"""Constructor"""
scrolled.ScrolledPanel.__init__(self, parent)
self.data = []
self.download_number = 1
# create the sizers
self.main_sizer = wx.BoxSizer(wx.VERTICAL)
dl_sizer = wx.BoxSizer(wx.HORIZONTAL)
# create the widgets
lbl = wx.StaticText(self, label="Download URL:")
self.dl_txt = wx.TextCtrl(self)
btn = wx.Button(self, label="Download")
btn.Bind(wx.EVT_BUTTON, self.onDownload)
# layout the widgets
dl_sizer.Add(lbl, 0, wx.ALL|wx.CENTER, 5)
dl_sizer.Add(self.dl_txt, 1, wx.EXPAND|wx.ALL, 5)
dl_sizer.Add(btn, 0, wx.ALL, 5)
self.main_sizer.Add(dl_sizer, 0, wx.EXPAND)
self.SetSizer(self.main_sizer)
self.SetAutoLayout(1)
self.SetupScrolling()
#----------------------------------------------------------------------
def onDownload(self, event):
"""
Update display with downloading gauges
"""
url = self.dl_txt.GetValue()
try:
header = requests.head(url)
fsize = int(header.headers["content-length"]) / 1024
sizer = wx.BoxSizer(wx.HORIZONTAL)
fname = os.path.basename(url)
lbl = wx.StaticText(self, label="Downloading %s" % fname)
gauge = MyGauge(self, fsize, self.download_number)
sizer.Add(lbl, 0, wx.ALL|wx.CENTER, 5)
sizer.Add(gauge, 0, wx.ALL|wx.EXPAND, 5)
self.main_sizer.Add(sizer, 0, wx.EXPAND)
self.Layout()
# start thread
DownloadThread(self.download_number, url, fsize)
self.dl_txt.SetValue("")
self.download_number += 1
except Exception, e:
print "Error: ", e
For the panel, I decided to go with the ScrolledPanel widget because I figured I might want to download a bunch of files and it would be nice if my downloaded added a scrollbar should the gauges run out of space. We set up the widgets and sizers in the __init__ method as usual. The onDownload method is where the action's at. Here we grab the URL from the text control and attempt to get the file size from the server using the requests package. We divide by 1024 to get the Kilobyte size. If that is successful, we create a horizontal box sizer and add a static text and a gauge to it. Note that we tell the gauge which download it is and we also pass that number along to a new thread. We also reset the text control to be empty to make it easier to add a new URL. Finally we increment the download number so we'll be ready to download a new file.
The rest of the code is pretty much boilerplate.
As you can see, this code was pretty simple to put together. I can see a lot to improve though. For example, it would be good if the application had some way to change where the downloads end up. Currently they just get saved in the same folder as the script. In retrospect, it should add the new downloads to the top and push the older ones down instead of the vice-versa. It would be cool to show the actual size of the file and how much is left. We could even add a tab for files that have finished downloading! Anyway, I hope that gives you some fun ideas for what you could do to improve the script.
Copyright © 2024 Mouse Vs Python | Powered by Pythonlibrary