Как продвинуть WXPython прогресс-бар GUI?

Question

Я пытаюсь выяснить, как добавить индикатор выполнения в программу установки 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('\n\nRemoving 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('\n\nDownloading:')
    urllib.request.urlretrieve(url, 'temp.zip')
    print('\nDownload Complete.')

# Extracts new file
def Extract():
    print('\n\nExtracting...')
    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('\n\nCleaning 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()

Кроме того, если вы видите способ улучшить или упростить мой код без минимизации эффективности, не стесняйтесь.

Answer 1

Я собрал ваш код, отличный ответ от Сри и добавил использование опции 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('\n\nRemoving 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('\n\nExtracting...')
    #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('\n\nCleaning 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()

Answer 2

Хорошо, во-первых, замораживание происходит из-за использования метода 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)

Надеюсь, что это имеет смысл.