如何从新线程更新 kivy 进度条值？
How to update kivy progress bar value from new thread?
我有这个图像下载器作为新线程和一个包含进度条的弹出窗口。进度条在下载期间不会更新，但会在下载完成后更新（下载器是用请求编写的，gui 应用程序是用 kivy 制作的）。有什么解决办法吗？
下载者：
它被分开在另一个文件中
class Downloader(threading.Thread):
def __init__(self, url: str, download_monitor):
super(Downloader, self).__init__(daemon=True) # daemon dies when main die
self.url = url
self.download_monitor = download_monitor # url popup
def run(self) -> None:
# Reset
self.download_monitor.reset()
file_name = self.url.split('/')[-1]
# Less RAM usage
with requests.get(self.url, stream=True) as req: # stream=True not to read at once
req.raise_for_status()
with open('temp/'+file_name, 'wb') as file:
chunks = list(enumerate(req.iter_content(chunk_size=8192)))
self.download_monitor.downloading_progress.max = chunks[-1][0] # last element
for progress, chunk in chunks:
self.download_monitor.downloading_progress.value = progress
file.write(chunk)
弹出 .py:
它被分开在另一个文件中
class UrlPopup(Popup):
url_input = ObjectProperty()
downloading_progress = ObjectProperty()
def __init__(self, **kwargs):
super(UrlPopup, self).__init__(**kwargs)
def download(self):
# https://www.nasa.gov/sites/default/files/thumbnails/image/hubble_ngc2903_potw2143a.jpg.jpg
if self.url_input.text.startswith('https://'): # if it is url address
download(self.url_input.text, self)
def on_dismiss(self):
self.reset()
self.url_input.text = ''
def reset(self):
self.downloading_progress.max = 0
self.downloading_progress.value = 0
弹出 .kv:
它被分开在另一个文件中
<UrlPopup>:
url_input: url_input
downloading_progress: downloading_progress
id: downloader
title: 'URL address'
size_hint: .25, None
height: 157
BoxLayout:
orientation: 'vertical'
size_hint_y: None
height: 64
TextInput:
id: url_input
multiline: False
size_hint_y: None
height: 32
font_size: 16
ProgressBar:
id: downloading_progress
size_hint_y: None
height: 32
BoxLayout:
orientation: 'horizontal'
size_hint_y: None
height: 32
Button:
text: 'Download'
on_press: root.download()
Button:
text: 'Close'
on_press: root.dismiss()
EDIT1
ApuCoder 我按照你写的做了但是下载后进度仍然更新。
还有其他想法吗？
PopUP .py:
class UrlPopup(Popup):
url_input = ObjectProperty()
downloading_progress = ObjectProperty()
progress_value = NumericProperty()
def update_progress(self, dt):
self.progress_value += 1
下载器.py:
with requests.get(self.url, stream=True) as req: # stream=True not to read at once
req.raise_for_status()
with open('temp/'+file_name, 'wb') as file:
chunks = list(enumerate(req.iter_content(chunk_size=8192)))
self.download_monitor.downloading_progress.max = chunks[-1][0] # last element
Clock.schedule_interval(self.download_monitor.update_progress, .1)
for progress, chunk in chunks:
#self.download_monitor.downloading_progress.value = progress
file.write(chunk)
弹出 .kv:
ProgressBar:
id: downloading_progress
value: root.progress_value
size_hint_y: None
height: 32
EDIT2
这与 class 下载器位于同一文件中。我在按下按钮时调用此函数
def download(url: str, download_monitor):
"""Other thread"""
downloader = Downloader(url, download_monitor)
downloader.start()
假设您想下载一些内容并在 kivy 中显示正在进行的过程（或当前状态），我更新并修改了您的一些代码以制作一个最小示例。
在这种情况下不需要创建新的Thread class，而是每次创建一个新的线程对象并将target设置为某个方法（这里，start_download) 用于获取和写入磁盘中的二进制数据。这样就可以在该方法内控制进度，无需调度。
from threading import Thread
import requests
from kivy.app import runTouchApp
from kivy.lang import Builder
from kivy.properties import (
BooleanProperty,
NumericProperty,
ObjectProperty,
)
from kivy.uix.popup import Popup
from kivy.uix.screenmanager import Screen
Builder.load_string("""
<DownLoadScreen>:
Button:
text: "Open Downloader"
on_release: root.open_downloader()
<UrlPopup>:
url_input: url_input
title: 'URL address'
size_hint: .75, None
height: "450dp"
BoxLayout:
orientation: "vertical"
TextInput:
id: url_input
text: "https://www.nasa.gov/sites/default/files/thumbnails/image/hubble_ngc2903_potw2143a.jpg.jpg"
multiline: False
size_hint_y: None
height: "64dp"
font_size: "16sp"
ProgressBar:
pos_hint: {"center_x" : 0.5}
value: root.prog_val
max: root.tot_size
Label:
id: lbl
text: "Downloading file...({:.0%})".format(root.prog_val/root.tot_size) if root.has_started else ""
BoxLayout:
size_hint_y: None
height: dp(48)
Button:
text: 'Download'
on_release: root.download()
Button:
text: 'Close'
on_press: root.dismiss()
""")
class UrlPopup(Popup):
url_input = ObjectProperty()
prog_val = NumericProperty(0) # To capture the current progress.
tot_size = NumericProperty(1) # Total size of the file/content. Setting the default value to 1 to avoid ZeroDivisionError, though will not affect anyhow.
has_started = BooleanProperty(False) # Just to manipulate the label text.
def start_download(self):
self.has_started = True
self.url = self.url_input.text
# file_name = self.url.split('/')[-1]
with requests.get(self.url, stream=True) as req:
if req.status_code == 200: # Here, you can create the binary file.
# chunks = list(enumerate(req.iter_content(chunk_size=8192))) # This may take more memory for larger file.
self.tot_size = int(req.headers["Content-Length"])
item_size = 2048 # Reducing the chunk size increases writing time and so needs more time in progress.
for i, chunk in enumerate(req.iter_content(chunk_size = item_size)):
self.prog_val = i*item_size
# file.write(chunk)
self.ids.lbl.text = "Download completed." # A confirmation message.
def download(self):
"""A new thread object will be created each time this method is revoked. But be careful about the threads already created."""
Thread(target = self.start_download).start()
def on_dismiss(self):
self.url_input.text = ""
self.has_started = False
class DownLoadScreen(Screen):
def open_downloader(self):
UrlPopup().open()
runTouchApp(DownLoadScreen())
让我知道它是否符合您的需要。
我有这个图像下载器作为新线程和一个包含进度条的弹出窗口。进度条在下载期间不会更新，但会在下载完成后更新（下载器是用请求编写的，gui 应用程序是用 kivy 制作的）。有什么解决办法吗？
下载者： 它被分开在另一个文件中
class Downloader(threading.Thread):
def __init__(self, url: str, download_monitor):
super(Downloader, self).__init__(daemon=True) # daemon dies when main die
self.url = url
self.download_monitor = download_monitor # url popup
def run(self) -> None:
# Reset
self.download_monitor.reset()
file_name = self.url.split('/')[-1]
# Less RAM usage
with requests.get(self.url, stream=True) as req: # stream=True not to read at once
req.raise_for_status()
with open('temp/'+file_name, 'wb') as file:
chunks = list(enumerate(req.iter_content(chunk_size=8192)))
self.download_monitor.downloading_progress.max = chunks[-1][0] # last element
for progress, chunk in chunks:
self.download_monitor.downloading_progress.value = progress
file.write(chunk)
弹出 .py: 它被分开在另一个文件中
class UrlPopup(Popup):
url_input = ObjectProperty()
downloading_progress = ObjectProperty()
def __init__(self, **kwargs):
super(UrlPopup, self).__init__(**kwargs)
def download(self):
# https://www.nasa.gov/sites/default/files/thumbnails/image/hubble_ngc2903_potw2143a.jpg.jpg
if self.url_input.text.startswith('https://'): # if it is url address
download(self.url_input.text, self)
def on_dismiss(self):
self.reset()
self.url_input.text = ''
def reset(self):
self.downloading_progress.max = 0
self.downloading_progress.value = 0
弹出 .kv: 它被分开在另一个文件中
<UrlPopup>:
url_input: url_input
downloading_progress: downloading_progress
id: downloader
title: 'URL address'
size_hint: .25, None
height: 157
BoxLayout:
orientation: 'vertical'
size_hint_y: None
height: 64
TextInput:
id: url_input
multiline: False
size_hint_y: None
height: 32
font_size: 16
ProgressBar:
id: downloading_progress
size_hint_y: None
height: 32
BoxLayout:
orientation: 'horizontal'
size_hint_y: None
height: 32
Button:
text: 'Download'
on_press: root.download()
Button:
text: 'Close'
on_press: root.dismiss()
EDIT1 ApuCoder 我按照你写的做了但是下载后进度仍然更新。 还有其他想法吗？ PopUP .py:
class UrlPopup(Popup):
url_input = ObjectProperty()
downloading_progress = ObjectProperty()
progress_value = NumericProperty()
def update_progress(self, dt):
self.progress_value += 1
下载器.py:
with requests.get(self.url, stream=True) as req: # stream=True not to read at once
req.raise_for_status()
with open('temp/'+file_name, 'wb') as file:
chunks = list(enumerate(req.iter_content(chunk_size=8192)))
self.download_monitor.downloading_progress.max = chunks[-1][0] # last element
Clock.schedule_interval(self.download_monitor.update_progress, .1)
for progress, chunk in chunks:
#self.download_monitor.downloading_progress.value = progress
file.write(chunk)
弹出 .kv:
ProgressBar:
id: downloading_progress
value: root.progress_value
size_hint_y: None
height: 32
EDIT2 这与 class 下载器位于同一文件中。我在按下按钮时调用此函数
def download(url: str, download_monitor):
"""Other thread"""
downloader = Downloader(url, download_monitor)
downloader.start()
假设您想下载一些内容并在 kivy 中显示正在进行的过程（或当前状态），我更新并修改了您的一些代码以制作一个最小示例。
在这种情况下不需要创建新的Thread class，而是每次创建一个新的线程对象并将target设置为某个方法（这里，start_download) 用于获取和写入磁盘中的二进制数据。这样就可以在该方法内控制进度，无需调度。
from threading import Thread
import requests
from kivy.app import runTouchApp
from kivy.lang import Builder
from kivy.properties import (
BooleanProperty,
NumericProperty,
ObjectProperty,
)
from kivy.uix.popup import Popup
from kivy.uix.screenmanager import Screen
Builder.load_string("""
<DownLoadScreen>:
Button:
text: "Open Downloader"
on_release: root.open_downloader()
<UrlPopup>:
url_input: url_input
title: 'URL address'
size_hint: .75, None
height: "450dp"
BoxLayout:
orientation: "vertical"
TextInput:
id: url_input
text: "https://www.nasa.gov/sites/default/files/thumbnails/image/hubble_ngc2903_potw2143a.jpg.jpg"
multiline: False
size_hint_y: None
height: "64dp"
font_size: "16sp"
ProgressBar:
pos_hint: {"center_x" : 0.5}
value: root.prog_val
max: root.tot_size
Label:
id: lbl
text: "Downloading file...({:.0%})".format(root.prog_val/root.tot_size) if root.has_started else ""
BoxLayout:
size_hint_y: None
height: dp(48)
Button:
text: 'Download'
on_release: root.download()
Button:
text: 'Close'
on_press: root.dismiss()
""")
class UrlPopup(Popup):
url_input = ObjectProperty()
prog_val = NumericProperty(0) # To capture the current progress.
tot_size = NumericProperty(1) # Total size of the file/content. Setting the default value to 1 to avoid ZeroDivisionError, though will not affect anyhow.
has_started = BooleanProperty(False) # Just to manipulate the label text.
def start_download(self):
self.has_started = True
self.url = self.url_input.text
# file_name = self.url.split('/')[-1]
with requests.get(self.url, stream=True) as req:
if req.status_code == 200: # Here, you can create the binary file.
# chunks = list(enumerate(req.iter_content(chunk_size=8192))) # This may take more memory for larger file.
self.tot_size = int(req.headers["Content-Length"])
item_size = 2048 # Reducing the chunk size increases writing time and so needs more time in progress.
for i, chunk in enumerate(req.iter_content(chunk_size = item_size)):
self.prog_val = i*item_size
# file.write(chunk)
self.ids.lbl.text = "Download completed." # A confirmation message.
def download(self):
"""A new thread object will be created each time this method is revoked. But be careful about the threads already created."""
Thread(target = self.start_download).start()
def on_dismiss(self):
self.url_input.text = ""
self.has_started = False
class DownLoadScreen(Screen):
def open_downloader(self):
UrlPopup().open()
runTouchApp(DownLoadScreen())
让我知道它是否符合您的需要。