You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

如何在wxPython应用中实时更新wx.Gauge并展示下载进度与速度

关于wxPython文件下载进度实时更新的问题

我正在开发一个文件下载的wxPython应用,需要实时更新wx.Gauge组件以展示下载进度,同时显示已下载大小/总大小以及实时下载速度。

在我的基础代码示例中,取值范围为0-100的进度变量存储在perc_value中,大小与当前速度文本存储在progress_text中。

我尝试使用tqdm库,但该库的示例仅展示了自身实现的进度条与速度功能,无法找到能获取这些信息的变量或函数。

能否提供帮助?谢谢!

import wx
import requests
from threading import Thread

class DownloadThread(Thread):
    def __init__(self, parent, url):
        Thread.__init__(self)
        self.parent = parent
        self.url = url
        self.start()

    def run(self):
        perc_value = 0
        progress_text = ''
        with open('file', 'wb') as f:
            r = requests.get(self.url, stream=True)
            for chunk in r.iter_content(4096):
                f.write(chunk)
                wx.CallAfter(self.parent.update_gauge, perc_value, progress_text)

class Main(wx.Frame):
    def __init__(self, parent):
        super().__init__(parent)
        sizer = wx.BoxSizer(wx.VERTICAL)
        
        self.gauge = wx.Gauge(self, -1)
        self.progress_t = wx.StaticText(self, -1, "Testing")

        sizer.Add(self.gauge, flag=wx.ALL, border=10)
        sizer.Add(self.progress_t, flag=wx.ALL, border=10)

        self.SetSizerAndFit(sizer)
        self.start_download()

    def start_download(self):
        url = '.../file.zip'
        DownloadThread(self, url)

    def update_gauge(self, perc_value, progress_text):
        self.gauge.SetValue(perc_value)
        self.progress_t.SetLabel(progress_text)

app = wx.App()
Main(None).Show()
app.MainLoop()

编辑:
我已对代码进行了一些改进,发现最佳方案是不使用tqdm。目前仅存的问题是下载速度计算不准确。

编辑2:问题已解决!修正后的代码如下:

import wx
import requests
from threading import Thread

class DownloadThread(Thread):
    def __init__(self, parent, url):
        Thread.__init__(self)
        self.parent = parent
        self.url = url
        self.start()

    def run(self):
        perc_temp = 0
        r = requests.get(self.url, stream=True)
        self.total_length = r.headers.get('content-length')

        with open('file.data', 'wb') as f:
            if self.total_length is None: # no content length header
                f.write(r.content)
            else:
                self.dl = 0
                self.dl_temp = 0
                self.total_length = int(self.total_length)
                
                self.total_size = self.get_downloaded_value(self.total_length)
                self.total_unit = self.get_unit(self.total_length)

                print(self.total_length)
                for data in r.iter_content(chunk_size=4096):
                    self.dl += len(data)
                    self.dl_temp += len(data)
                    f.write(data)
                    value = int(100 * self.dl / self.total_length)
                    if perc_temp != value:
                        perc_temp = value
                        wx.CallAfter(self.parent.update_gauge, value)

    def get_progress_text(self):
        downloaded = self.get_downloaded_value(self.dl)
        unit_downloaded = self.get_unit(self.dl)

        speed = self.get_downloaded_value(self.dl_temp)
        speed_unit = self.get_unit(self.dl_temp)

        text = f"{downloaded:.2f} {unit_downloaded} / {self.total_size:.2f} {self.total_unit}		{speed:.2f} {speed_unit}/s"

        self.dl_temp = 0
        return text

    def get_unit(self, size):
        unit = ''
        if size < 1024:
            unit = 'B'
        elif size < 1048576:
            unit = 'KB'
        else:
            unit = 'MB'

        return unit

    def get_downloaded_value(self, size):
        if size < 1024:
            return size
        elif size < 104876:
            return size / 1024
        else:
            return size / 1048576


class Main(wx.Frame):
    def __init__(self, parent):
        super().__init__(parent)
        sizer = wx.BoxSizer(wx.VERTICAL)
        
        self.timer = wx.Timer(self)
        self.Bind(wx.EVT_TIMER, self.OnTimer)

        self.gauge = wx.Gauge(self, -1)
        self.progress_t = wx.StaticText(self, -1, "Testing", size=(300, 23))

        sizer.Add(self.gauge, flag=wx.ALL | wx.EXPAND, border=10)
        sizer.Add(self.progress_t, flag=wx.ALL | wx.EXPAND, border=10)

        self.SetSizerAndFit(sizer)
        self.start_download()

    def start_download(self):
        url = 'https://portal.inmet.gov.br/uploads/dadoshistoricos/2008.zip'
        self.thread = DownloadThread(self, url)
        self.timer.Start(1000)

    def update_gauge(self, perc_value):
        self.gauge.SetValue(perc_value)

    def OnTimer(self, event):
        self.progress_t.SetLabel(self.thread.get_progress_text())
        

app = wx.App()
Main(None).Show()
app.MainLoop()

内容的提问来源于stack exchange,提问作者NeoFahrenheit

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.08.17 19:01:01