我正在尝试弄清楚如何将进度条添加到我正在制作的 GUI 安装程序中。问题实际上是使进度条工作。我已经实现了它，但它将整个程序冻结了一半。
# Import Libraries
import requests, os, sys, zipfile, shutil, subprocess, wx, urllib, time
from threading import *
# Define variables
url = "Enter any dropbox link .zip file here"
r = requests.get(url, stream = True)
# Button definitions
ID_START = wx.NewId()
# Define notification event for thread completion
EVT_RESULT_ID = wx.NewId()
# Checks for old files
def Check():
if os.path.exists("Folder"):
print('nnRemoving old files...')
subprocess.check_call(('attrib -R ' + 'Folder' + '\* /S').split())
shutil.rmtree('Folder')
print('nRemoved old files.')
else:
pass
# Downloads new file
def Download():
print('nnDownloading:')
urllib.request.urlretrieve(url, 'temp.zip')
print('nDownload Complete.')
# Extracts new file
def Extract():
print('nnExtracting...')
zip_ref = zipfile.ZipFile("temp.zip", 'r')
zip_ref.extractall("Folder")
zip_ref.close()
print('nExtraction Complete')
# Deletes the .zip file but leave the folder
def Clean():
print('nnCleaning up...')
os.remove("temp.zip")
print('nDone!')
# Thread class that executes processing
class WorkerThread(Thread):
"""Worker Thread Class."""
def __init__(self, notify_window):
"""Init Worker Thread Class."""
Thread.__init__(self)
self._notify_window = notify_window
# This starts the thread running on creation.
self.start()
# This is what runs on a separate thread when you click the download button
def run(self):
"""Run Worker Thread."""
# This is the code executing in the new thread.
Check()
Download()
Extract()
Clean()
# GUI Frame class that spins off the worker thread
class MainFrame(wx.Frame):
"""Class MainFrame."""
def __init__(self, parent, id):
"""Create the MainFrame."""
wx.Frame.__init__(self, parent, id, 'RFMP GUInstaller',
style=wx.DEFAULT_FRAME_STYLE ^ wx.RESIZE_BORDER)
self.SetSize(400, 350)
wx.Button(self, ID_START, 'Download', size=(300,50), pos=(42,250))
self.Bind(wx.EVT_BUTTON, self.OnStart, id=ID_START)
self.status = wx.StaticText(self, -1, '', pos=(7,200))
self.gauge = wx.Gauge(self, range = 1000, size = (370, 30), pos=(7,217),
style = wx.GA_HORIZONTAL)
# And indicate we don't have a worker thread yet
self.worker = None
def OnStart(self, event):
"""Start Computation."""
self.count = 0
# Trigger the worker thread unless it's already busy
if not self.worker:
self.status.SetLabel('Downloading...')
self.worker = WorkerThread(self)
while self.count <= 10000:
time.sleep(.001);
self.count = self.count + 1
self.gauge.SetValue(self.count)
self.status.SetLabel('Done!')
def OnResult(self, event):
"""Show Result status."""
# The worker is done
self.worker = None
class MainApp(wx.App):
"""Class Main App."""
def OnInit(self):
"""Init Main App."""
self.frame = MainFrame(None, -1)
self.frame.Show(True)
self.SetTopWindow(self.frame)
return True
# Main Loop
if __name__ == '__main__':
app = MainApp(0)
app.MainLoop()
此外，如果您看到任何可以在不降低效率的情况下改进或简化我的代码的方法，请随意。
好的，首先冻结是由于使用了sleep方法。
您正在"下载"按钮的点击事件中生成新线程，这很好。但是，您必须使该线程以某种方式与主线程/帧通信，而不是在主线程中休眠。
这是可以使用 wx 事件的地方。一个很好的教程在这里。在Clean()方法后添加如下内容：
myEVT_PROGRESS = wx.NewEventType() # Custom Event Type
EVT_PROGRESS = wx.PyEventBinder(myEVT_PROGRESS, 1) # bind specific events to event handlers
class ProgressEvent(wx.PyCommandEvent):
"""Event to signal that a status or progress changed"""
def __init__(self, etype, eid, status=None, progress=None):
"""Creates the event object"""
wx.PyCommandEvent.__init__(self, etype, eid)
self._status = status # field to update label
self._progress = progress # field to update progress bar
def GetValue(self):
"""Returns the value from the event.
@return: the tuple of status and progress
"""
return (self._status, self._progress)
现在工作线程变得有点复杂，因为它必须通知主线程任何进度：
# Thread class that executes processing
class WorkerThread(Thread):
"""Worker Thread Class."""
def __init__(self, notify_window):
"""Init Worker Thread Class."""
Thread.__init__(self)
self._notify_window = notify_window
self.sendEvent('started')
# This starts the thread running on creation.
self.start()
# This is what runs on a separate thread when you click the download button
def run(self):
# This is the code executing in the new thread.
self.sendEvent('checking', 0)
# Check() # this method isn't working for me...?
self.sendEvent('Downloading...', 100)
Download()
self.sendEvent('Downloading complete', 400)
# ... same pattern as above for other methods...
Extract()
Clean()
def sendEvent(self, status=None, progress=None):
# Send event to main frame, first param (str) is for label, second (int) for the progress bar
evt = ProgressEvent(myEVT_PROGRESS, -1, status, progress)
wx.PostEvent(self._notify_window, evt)
现在，剩下的就是现在在主线程上接收事件。
class MainFrame(wx.Frame):
"""Class MainFrame."""
def __init__(self, parent, id):
# ...same as before...
self.Bind(EVT_PROGRESS, self.OnResult) # Bind our new custom event to a function
def OnStart(self, event):
# Trigger the worker thread unless it's already busy
if not self.worker:
self.status.SetLabel('')
self.worker = WorkerThread(self)
def OnResult(self, event):
"""Our handler for our custom progress event."""
status, progress = event.GetValue()
self.status.SetLabel(status)
if progress:
self.gauge.SetValue(progress)
希望这是有道理的。
我已经将您的代码、Sree 的出色答案放在一起，并添加了使用urllib.request.urlretrieve()reporthook选项来显示下载完成。
注意：任何功劳都应该归功于Sree这只是我个人的练习。
# Import Libraries
import requests, os, sys, zipfile, shutil, subprocess, wx, urllib, time
from threading import *
# Define variables
url = "Enter any dropbox link .zip file here"
#r = requests.get(url, stream = True)
# Button definitions
ID_START = wx.NewId()
myEVT_PROGRESS = wx.NewEventType() # Custom Event Type
EVT_PROGRESS = wx.PyEventBinder(myEVT_PROGRESS, 1) # bind specific events to event handlers
class ProgressEvent(wx.PyCommandEvent):
"""Event to signal that a status or progress changed"""
def __init__(self, etype, eid, status=None, progress=None):
"""Creates the event object"""
wx.PyCommandEvent.__init__(self, etype, eid)
self._status = status # field to update label
self._progress = progress # field to update progress bar
def GetValue(self):
"""Returns the value from the event.
@return: the tuple of status and progress
"""
return (self._status, self._progress)
# Checks for old files
def Check():
if os.path.exists("Folder"):
print('nnRemoving old files...')
#subprocess.check_call(('attrib -R ' + 'Folder' + '\* /S').split())
#shutil.rmtree('Folder')
print('nRemoved old files.')
else:
pass
# Extracts new file
def Extract():
print('nnExtracting...')
#zip_ref = zipfile.ZipFile("temp.zip", 'r')
#zip_ref.extractall("Folder")
#zip_ref.close()
time.sleep(5)
print('nExtraction Complete')
# Deletes the .zip file but leave the folder
def Clean():
print('nnCleaning up...')
#os.remove("temp.zip")
print('nDone!')
# Thread class that executes processing
class WorkerThread(Thread):
"""Worker Thread Class."""
def __init__(self, notify_window):
"""Init Worker Thread Class."""
Thread.__init__(self)
self._notify_window = notify_window
# This starts the thread running on creation.
self.start()
# This is what runs on a separate thread when you click the download button
def run(self):
"""Run Worker Thread."""
# This is the code executing in the new thread.
self.SendEvent('Checking...', 50)
Check()
self.SendEvent('Connecting to download...', 0)
#Perform download
urllib.request.urlretrieve(url, 'temp.zip', reporthook=self.Download_Progress)
self.SendEvent('Extracting...', 800)
Extract()
self.SendEvent('Cleaning...', 900)
Clean()
self.SendEvent('Finished...', 1000)
def Download_Progress(self, block_num, block_size, total_size):
downloaded = block_num * block_size
progress = int((downloaded/total_size)*1000)
if progress > 1000:
progress = 1000
self.SendEvent("Download active...",progress)
def SendEvent(self, status=None, progress=None):
# Send event to main frame, first param (str) is for label, second (int) for the progress bar
evt = ProgressEvent(myEVT_PROGRESS, -1, status, progress)
wx.PostEvent(self._notify_window, evt)
# GUI Frame class that spins off the worker thread
class MainFrame(wx.Frame):
"""Class MainFrame."""
def __init__(self, parent, id):
"""Create the MainFrame."""
wx.Frame.__init__(self, parent, id, 'RFMP GUInstaller',
style=wx.DEFAULT_FRAME_STYLE ^ wx.RESIZE_BORDER)
self.SetSize(400, 350)
wx.Button(self, ID_START, 'Download', size=(300,50), pos=(42,250))
self.Bind(wx.EVT_BUTTON, self.OnStart, id=ID_START)
self.status = wx.StaticText(self, -1, '', pos=(7,200))
self.gauge = wx.Gauge(self, range = 1000, size = (370, 30), pos=(7,217),
style = wx.GA_HORIZONTAL)
# And indicate we don't have a worker thread yet
self.worker = None
self.Bind(EVT_PROGRESS, self.OnResult) # Bind our new custom event to a function
def OnStart(self, event):
"""Start Computation."""
self.count = 0
# Trigger the worker thread unless it's already busy
if not self.worker:
self.worker = WorkerThread(self)
def OnResult(self, event):
"""Show Result status."""
# The worker is done
self.worker = None
status, progress = event.GetValue()
self.status.SetLabel(status)
if progress:
self.gauge.SetValue(progress)
class MainApp(wx.App):
"""Class Main App."""
def OnInit(self):
"""Init Main App."""
self.frame = MainFrame(None, -1)
self.frame.Show(True)
self.SetTopWindow(self.frame)
return True
# Main Loop
if __name__ == '__main__':
app = MainApp(0)
app.MainLoop()