如何在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
相关产品推荐
相关产品推荐

