前往小程序，Get更优阅读体验！
立即前往
文档建议反馈控制台
首页
学习
活动
专区
工具
TVP
最新优惠活动
文章/答案/技术大牛
发布
首页
学习
活动
专区
工具
TVP最新优惠活动
返回腾讯云官网
社区首页 >专栏 >PySide6 GUI 编程（45）：QRunnable小练习之进度条控制

PySide6 GUI 编程（45）：QRunnable小练习之进度条控制

原创
作者头像
bowenerchen
修改于 2024-09-17 08:40:09
2140
修改于 2024-09-17 08:40:09
举报
文章被收录于专栏：编码视界

简单的进度条功能

示例代码

代码语言：python
代码运行次数：0
复制
from __future__ import annotations

import sys
import threading
import time
from datetime import datetime
from random import randint

from PySide6.QtCore import QObject, QRunnable, QThreadPool, Signal, Slot
from PySide6.QtWidgets import QApplication, QLabel, QMainWindow, QProgressBar, QPushButton, QVBoxLayout, QWidget


def get_time_str() -> str:
    return datetime.now().isoformat(sep = ' ')


class WorkerSignals(QObject):
    begin = Signal(str, str)
    finished = Signal(str, str)
    progress_value = Signal(int)  # 进度值


class MyWorker(QRunnable):
    def __init__(self, *args, **kwargs):
        # 传递参数给 Worker
        super().__init__()
        print('args=', args, 'kwargs=', kwargs)
        # 初始化信号和槽
        self.signals = WorkerSignals()
        self.args = args
        self.kwargs = kwargs

    @Slot()
    def run(self) -> None:
        thread_id = str(threading.get_ident())
        # 开始工作
        self.signals.begin.emit(thread_id, get_time_str())
        total_progress = 1000
        for i in range(total_progress):
            time.sleep(float(randint(100, 500)) / 1000)  # 模拟耗时操作
            self.signals.progress_value.emit(i)  # 发送信号,更新进度值
        # 结束工作
        self.signals.finished.emit(thread_id, get_time_str())


class MyMainWindow(QMainWindow):
    def __init__(self):
        super().__init__()
        self.setWindowTitle('Hello, PySide6!')
        self.setToolTip('A PySide6 GUI Application Demo')
        self.progress_bar = QProgressBar()
        self.progress_bar.setRange(0, 1000)
        self.label = QLabel('0.0%')
        self.button = QPushButton('Start')
        self.button.clicked.connect(self.start_threads)

        v_layout = QVBoxLayout()
        v_layout.addWidget(self.progress_bar)
        v_layout.addWidget(self.label)
        v_layout.addWidget(self.button)

        container = QWidget()
        container.setLayout(v_layout)

        self.setCentralWidget(container)

        self.threads = QThreadPool()
        print('maximum threads: {}'.format(self.threads.maxThreadCount()))

    def update_progress_value(self, value: int) -> None:
        percentage = (value / 1000) * 100  # 计算百分比
        self.label.setText(f'{percentage:.1f}%')
        self.progress_bar.setValue(value)

    def handle_worker_begin(self, thread_id: str, time_str: str) -> None:
        print('BEGIN thread_id=', thread_id, ', time_str =', time_str)

    def handle_worker_finished(self, thread_id: str, time_str: str) -> None:
        print('FINISHED thread_id=', thread_id, ', time_str =', time_str)

    def start_threads(self) -> None:
        worker = MyWorker()
        worker.signals.begin.connect(self.handle_worker_begin)
        worker.signals.finished.connect(self.handle_worker_finished)
        worker.signals.progress_value.connect(self.update_progress_value)
        self.threads.start(worker)


if __name__ == "__main__":
    app = QApplication(sys.argv)
    window = MyMainWindow()
    window.show()
    app.exec()

运行效果

期望的效果

期望的正常运行效果
期望的正常运行效果

随之而来的问题

覆盖写问题
多次点击 start 导致的覆盖写问题
多次点击 start 导致的覆盖写问题
窗口关闭问题
窗口关闭带来的 panic 问题
窗口关闭带来的 panic 问题
异常 panic
异常 panic

优化一：只允许开启一个进度条实例

示例代码

代码语言：python
代码运行次数：0
复制
from __future__ import annotations

import sys
import threading
import time
from datetime import datetime
from random import randint

from PySide6.QtCore import QObject, QRunnable, QThreadPool, Signal, Slot
from PySide6.QtWidgets import QApplication, QLabel, QMainWindow, QProgressBar, QPushButton, QVBoxLayout, QWidget


def get_time_str() -> str:
    return datetime.now().isoformat(sep = ' ')


class WorkerSignals(QObject):
    begin = Signal(str, str)
    finished = Signal(str, str)
    progress_value = Signal(int)  # 进度值


class MyWorker(QRunnable):
    def __init__(self, *args, **kwargs):
        # 传递参数给 Worker
        super().__init__()
        print('args=', args, 'kwargs=', kwargs)
        # 初始化信号和槽
        self.signals = WorkerSignals()
        self.args = args
        self.kwargs = kwargs

    @Slot()
    def run(self) -> None:
        thread_id = str(threading.get_ident())
        # 开始工作
        self.signals.begin.emit(thread_id, get_time_str())
        total_progress = 1000
        for i in range(total_progress):
            time.sleep(float(randint(100, 500)) / 1000)  # 模拟耗时操作
            self.signals.progress_value.emit(i)  # 发送信号,更新进度值
        # 结束工作
        self.signals.finished.emit(thread_id, get_time_str())


class MyMainWindow(QMainWindow):
    def __init__(self):
        super().__init__()
        self.setWindowTitle('Hello, PySide6!')
        self.setToolTip('A PySide6 GUI Application Demo')
        self.progress_bar = QProgressBar()
        self.progress_bar.setRange(0, 1000)
        self.label = QLabel('0.0%')
        self.button = QPushButton('Start')
        self.button.clicked.connect(self.start_threads)

        v_layout = QVBoxLayout()
        v_layout.addWidget(self.progress_bar)
        v_layout.addWidget(self.label)
        v_layout.addWidget(self.button)

        container = QWidget()
        container.setLayout(v_layout)

        self.setCentralWidget(container)

        self.threads = QThreadPool()
        print('maximum threads: {}'.format(self.threads.maxThreadCount()))

    def update_progress_value(self, value: int) -> None:
        percentage = (value / 1000) * 100  # 计算百分比
        self.label.setText(f'{percentage:.1f}%')
        self.progress_bar.setValue(value)

    def handle_worker_begin(self, thread_id: str, time_str: str) -> None:
        self.button.setEnabled(False) # 线程开始后将按钮设置为禁用状态
        print('BEGIN thread_id=', thread_id, ', time_str =', time_str)

    def handle_worker_finished(self, thread_id: str, time_str: str) -> None:
        self.button.setEnabled(True) # 线程完成后将按钮设置为可用状态
        print('FINISHED thread_id=', thread_id, ', time_str =', time_str)

    def start_threads(self) -> None:
        worker = MyWorker()
        worker.signals.begin.connect(self.handle_worker_begin)
        worker.signals.finished.connect(self.handle_worker_finished)
        worker.signals.progress_value.connect(self.update_progress_value)
        self.threads.start(worker)


if __name__ == "__main__":
    app = QApplication(sys.argv)
    window = MyMainWindow()
    window.show()
    app.exec()

示例效果

只允许开启一个进度条实例
只允许开启一个进度条实例

优化二：主窗口关闭检测

示例代码

代码语言：python
代码运行次数：0
复制
from __future__ import annotations

import sys
import threading
import time
from datetime import datetime
from random import randint

from PySide6.QtCore import QEvent, QObject, QRunnable, QThreadPool, Signal, Slot
from PySide6.QtWidgets import QApplication, QLabel, QMainWindow, QMessageBox, QProgressBar, QPushButton, QVBoxLayout, QWidget


def get_time_str() -> str:
    return datetime.now().isoformat(sep = ' ')


class WorkerSignals(QObject):
    begin = Signal(str, str)
    finished = Signal(str, str)
    progress_value = Signal(int)  # 进度值


class MyWorker(QRunnable):
    def __init__(self, *args, **kwargs):
        # 传递参数给 Worker
        super().__init__()
        print('args=', args, 'kwargs=', kwargs)
        # 初始化信号和槽
        self.signals = WorkerSignals()
        self.args = args
        self.kwargs = kwargs

    @Slot()
    def run(self) -> None:
        thread_id = str(threading.get_ident())
        # 开始工作
        self.signals.begin.emit(thread_id, get_time_str())
        total_progress = 1000
        for i in range(total_progress):
            time.sleep(float(randint(100, 500)) / 1000)  # 模拟耗时操作
            self.signals.progress_value.emit(i)  # 发送信号,更新进度值
        # 结束工作
        self.signals.finished.emit(thread_id, get_time_str())


class MyMainWindow(QMainWindow):
    def __init__(self):
        super().__init__()
        self.active_workers = 0
        self.setWindowTitle('Hello, PySide6!')
        self.setToolTip('A PySide6 GUI Application Demo')
        self.progress_bar = QProgressBar()
        self.progress_bar.setRange(0, 1000)
        self.label = QLabel('0.0%')
        self.button = QPushButton('Start')
        self.button.clicked.connect(self.start_threads)

        v_layout = QVBoxLayout()
        v_layout.addWidget(self.progress_bar)
        v_layout.addWidget(self.label)
        v_layout.addWidget(self.button)

        container = QWidget()
        container.setLayout(v_layout)

        self.setCentralWidget(container)

        self.threads = QThreadPool()
        print('maximum threads: {}'.format(self.threads.maxThreadCount()))

    def closeEvent(self, event: QEvent) -> None:
        if self.threads.activeThreadCount() > 0:
            QMessageBox.critical(self, '任务正在执行', '还有进程在运行，不允许关闭',
                                 QMessageBox.StandardButton.Ignore)
        event.ignore()  # 忽略关闭事件

    def update_progress_value(self, value: int) -> None:
        percentage = (value / 1000) * 100  # 计算百分比
        self.label.setText(f'{percentage:.1f}%')
        self.progress_bar.setValue(value)

    def handle_worker_begin(self, thread_id: str, time_str: str) -> None:
        self.button.setEnabled(False)
        self.active_workers += 1
        print('BEGIN thread_id=', thread_id, ', time_str =', time_str)

    def handle_worker_finished(self, thread_id: str, time_str: str) -> None:
        self.button.setEnabled(True)
        self.active_workers -= 1
        print('FINISHED thread_id=', thread_id, ', time_str =', time_str)

    def start_threads(self) -> None:
        worker = MyWorker()
        worker.signals.begin.connect(self.handle_worker_begin)
        worker.signals.finished.connect(self.handle_worker_finished)
        worker.signals.progress_value.connect(self.update_progress_value)
        self.threads.start(worker)


if __name__ == "__main__":
    app = QApplication(sys.argv)
    window = MyMainWindow()
    window.show()
    app.exec()

示例效果

主窗口关闭检测与拦截
主窗口关闭检测与拦截

此时只能强制退出程序，且退出时仍然会有报错：

强制退出程序导致的报错
强制退出程序导致的报错

优化三：优雅退出程序

示例代码

代码语言：python
代码运行次数：0
复制
from __future__ import annotations

import sys
import threading
import time
from datetime import datetime
from random import randint

from PySide6.QtCore import QEvent, QObject, QRunnable, QThreadPool, Signal, Slot
from PySide6.QtWidgets import QApplication, QLabel, QMainWindow, QProgressBar, QPushButton, QVBoxLayout, QWidget


def get_time_str() -> str:
    return datetime.now().isoformat(sep = ' ')


class WorkerSignals(QObject):
    begin = Signal(str, str)
    finished = Signal(str, str)
    progress_value = Signal(int)  # 进度值


class MyWorker(QRunnable):
    def __init__(self, *args, **kwargs):
        # 传递参数给 Worker
        super().__init__()
        print('args=', args, 'kwargs=', kwargs)
        # 初始化信号和槽
        self.signals = WorkerSignals()
        self.args = args
        self.kwargs = kwargs

        self.killed = False

    @Slot()
    def run(self) -> None:
        thread_id = str(threading.get_ident())
        # 开始工作
        self.signals.begin.emit(thread_id, get_time_str())
        total_progress = 1000
        for i in range(total_progress):
            if self.killed:
                print('worker quit....')
                return
            time.sleep(float(randint(100, 1000)) / 1000)  # 模拟耗时操作
            self.signals.progress_value.emit(i)  # 发送信号,更新进度值
        # 结束工作
        self.signals.finished.emit(thread_id, get_time_str())

    def kill(self) -> None:
        print('set worker to killed')
        self.killed = True


class MyMainWindow(QMainWindow):
    def __init__(self):
        super().__init__()
        self.active_workers = 0
        self.setWindowTitle('Hello, PySide6!')
        self.setToolTip('A PySide6 GUI Application Demo')
        self.progress_bar = QProgressBar()
        self.progress_bar.setRange(0, 1000)
        self.label = QLabel('0.0%')
        self.button = QPushButton('Start')
        self.button.clicked.connect(self.start_threads)

        self.worker = MyWorker()  # 创建一个 Worker 对象

        v_layout = QVBoxLayout()
        v_layout.addWidget(self.progress_bar)
        v_layout.addWidget(self.label)
        v_layout.addWidget(self.button)

        container = QWidget()
        container.setLayout(v_layout)

        self.setCentralWidget(container)

        self.threads = QThreadPool()
        print('maximum threads: {}'.format(self.threads.maxThreadCount()))

    def closeEvent(self, event: QEvent) -> None:
        if self.active_workers > 0:
            self.worker.kill()
            self.threads.waitForDone()
            self.active_workers -= 1
            self.worker = None
        event.accept()

    def update_progress_value(self, value: int) -> None:
        percentage = (value / 1000) * 100  # 计算百分比
        self.label.setText(f'{percentage:.1f}%')
        self.progress_bar.setValue(value)

    def handle_worker_begin(self, thread_id: str, time_str: str) -> None:
        self.button.setEnabled(False)
        self.active_workers += 1
        print('BEGIN thread_id=', thread_id, ', time_str =', time_str)

    def handle_worker_finished(self, thread_id: str, time_str: str) -> None:
        self.button.setEnabled(True)
        self.active_workers -= 1
        print('FINISHED thread_id=', thread_id, ', time_str =', time_str)

    def start_threads(self) -> None:
        if self.worker is not None:
            self.worker.signals.begin.connect(self.handle_worker_begin)
            self.worker.signals.finished.connect(self.handle_worker_finished)
            self.worker.signals.progress_value.connect(self.update_progress_value)
            self.threads.start(self.worker)


if __name__ == "__main__":
    app = QApplication(sys.argv)
    window = MyMainWindow()
    window.show()
    app.exec()
退出触发逻辑
退出触发逻辑
退出的事件捕获与清理
退出的事件捕获与清理

示例效果

优雅退出主程序
优雅退出主程序

优化四：进度条暂停

示例代码

代码语言：python
代码运行次数：0
复制
from __future__ import annotations

import sys
import threading
import time
from datetime import datetime
from random import randint

from PySide6.QtCore import QEvent, QObject, QRunnable, QThreadPool, Signal, Slot
from PySide6.QtWidgets import QApplication, QLabel, QMainWindow, QProgressBar, QPushButton, QVBoxLayout, QWidget


def get_time_str() -> str:
    return datetime.now().isoformat(sep = ' ')


class WorkerSignals(QObject):
    begin = Signal(str, str)
    finished = Signal(str, str)
    progress_value = Signal(int)  # 进度值


class MyWorker(QRunnable):
    def __init__(self, *args, **kwargs):
        # 传递参数给 Worker
        super().__init__()
        print('args=', args, 'kwargs=', kwargs)
        # 初始化信号和槽
        self.signals = WorkerSignals()
        self.args = args
        self.kwargs = kwargs

        self.killed = False
        self.paused = False

    @Slot()
    def run(self) -> None:
        thread_id = str(threading.get_ident())
        # 开始工作
        self.signals.begin.emit(thread_id, get_time_str())
        total_progress = 1000
        for i in range(total_progress + 1):
            while self.paused:
                time.sleep(0)
            if self.killed:
                print('worker quit....')
                return
            time.sleep(float(randint(1, 10)) / 1000)  # 模拟耗时操作
            self.signals.progress_value.emit(i)  # 发送信号,更新进度值
        # 结束工作
        self.signals.finished.emit(thread_id, get_time_str())

    def kill(self) -> None:
        print('set worker to killed')
        self.killed = True

    def pause(self) -> None:
        self.paused = True

    def pause_resume(self) -> None:
        self.paused = False

    def is_paused(self) -> bool:
        return self.paused


class MyMainWindow(QMainWindow):
    def __init__(self):
        super().__init__()
        self.active_workers = 0
        self.setWindowTitle('Hello, PySide6!')
        self.setToolTip('A PySide6 GUI Application Demo')
        self.progress_bar = QProgressBar()
        self.progress_bar.setRange(0, 1000)
        self.label = QLabel('0.0%')
        self.button = QPushButton('Start')
        self.button.clicked.connect(self.start_threads)

        self.pause_button = QPushButton('Pause')
        self.pause_button.clicked.connect(self.pause_resume)
        self.pause_button.setEnabled(False)

        self.worker = MyWorker()  # 创建一个 Worker 对象

        v_layout = QVBoxLayout()
        v_layout.addWidget(self.progress_bar)
        v_layout.addWidget(self.label)
        v_layout.addWidget(self.button)
        v_layout.addWidget(self.pause_button)

        container = QWidget()
        container.setLayout(v_layout)

        self.setCentralWidget(container)

        self.threads = QThreadPool()
        print('maximum threads: {}'.format(self.threads.maxThreadCount()))

    def closeEvent(self, event: QEvent) -> None:
        if self.active_workers > 0:
            self.worker.kill()
            self.worker.pause_resume()
            self.threads.waitForDone()
            self.active_workers -= 1
            self.worker = None
        event.accept()

    def update_progress_value(self, value: int) -> None:
        if self.worker is not None and not self.worker.is_paused():
            percentage = (value / 1000) * 100  # 计算百分比
            self.label.setText(f'{percentage:.1f}%')
            self.progress_bar.setValue(value)

    def handle_worker_begin(self, thread_id: str, time_str: str) -> None:
        self.button.setEnabled(False)
        self.active_workers += 1
        print('BEGIN thread_id=', thread_id, ', time_str =', time_str)

    def handle_worker_finished(self, thread_id: str, time_str: str) -> None:
        self.button.setEnabled(True)
        self.active_workers -= 1
        print('FINISHED thread_id=', thread_id, ', time_str =', time_str)
        self.worker = MyWorker()  # 创建一个 Worker 对象
        self.pause_button.setEnabled(False)

    def start_threads(self) -> None:
        if self.worker is not None:
            self.worker.signals.begin.connect(self.handle_worker_begin)
            self.worker.signals.finished.connect(self.handle_worker_finished)
            self.worker.signals.progress_value.connect(self.update_progress_value)
            self.threads.start(self.worker)
            self.pause_button.setEnabled(True)

    def pause_resume(self) -> None:
        if self.worker is not None:
            if not self.worker.is_paused():
                self.worker.pause()
                self.pause_button.setText('Resume')
            else:
                self.worker.pause_resume()
                self.pause_button.setText('Pause')


if __name__ == "__main__":
    app = QApplication(sys.argv)
    window = MyMainWindow()
    window.show()
    app.exec()
暂停逻辑
暂停逻辑
退出逻辑
退出逻辑

示例效果

进度条暂停
进度条暂停
进度条完成效果
进度条完成效果

原创声明：本文系作者授权腾讯云开发者社区发表，未经许可，不得转载。

如有侵权，请联系 cloudcommunity@tencent.com 删除。

python

原创声明：本文系作者授权腾讯云开发者社区发表，未经许可，不得转载。

如有侵权，请联系 cloudcommunity@tencent.com 删除。

python
#PySide6
#GUI 编程
#Python3
评论
登录后参与评论
0 条评论
热度
最新
登录 后参与评论
推荐阅读
LV.
文章
0
获赞
0
目录
  • 简单的进度条功能
    • 示例代码
      • 运行效果
        • 期望的效果
        • 随之而来的问题
    • 优化一：只允许开启一个进度条实例
      • 示例代码
        • 示例效果
        • 优化二：主窗口关闭检测
          • 示例代码
            • 示例效果
            • 优化三：优雅退出程序
              • 示例代码
                • 示例效果
                • 优化四：进度条暂停
                  • 示例代码
                    • 示例效果
                    领券

                    腾讯云开发者

                    扫码关注腾讯云开发者

                    扫码关注腾讯云开发者

                    领取腾讯云代金券

                    热门产品

                    热门推荐

                    更多推荐

                    Copyright © 2013 - 2024 Tencent Cloud. All Rights Reserved. 腾讯云 版权所有 

                    深圳市腾讯计算机系统有限公司 ICP备案/许可证号：粤B2-20090059 深公网安备号 44030502008569

                    腾讯云计算（北京）有限责任公司 京ICP证150476号 |  京ICP备11018762号 | 京公网安备号11010802020287

                    问题归档专栏文章快讯文章归档关键词归档开发者手册归档开发者手册 Section 归档

                    Copyright © 2013 - 2024 Tencent Cloud.

                    All Rights Reserved. 腾讯云 版权所有

                    登录 后参与评论