#!/usr/bin/env python import argparse import sys import time # 27.1.2020 parser = argparse.ArgumentParser() parser.add_argument("--animations", "-a", action="store_true", help="Enable Animations") parser.add_argument('what', nargs='?', choices=["PySide6", "PySide2", "PyQt5"], default='PySide6') result = parser.parse_args() MODULE = result.what if MODULE == 'PySide6': from PySide6.QtCore import QElapsedTimer, QLibraryInfo, QThread, QTimer, qVersion, Qt from PySide6.QtGui import QGuiApplication from PySide6.QtWidgets import (QApplication, QHBoxLayout, QLabel, QWidget, QPushButton, QProgressBar, QComboBox, QStyle, QVBoxLayout) elif MODULE == 'PySide2': from PySide2.QtCore import QElapsedTimer, QLibraryInfo, QThread, QTimer, qVersion, Qt from PySide2.QtGui import QGuiApplication from PySide2.QtWidgets import (QApplication, QHBoxLayout, QLabel, QWidget, QPushButton, QProgressBar, QComboBox, QStyle, QVBoxLayout) else: from PyQt5.QtCore import QElapsedTimer, QLibraryInfo, QThread, QTimer, qVersion, Qt from PyQt5.QtGui import QGuiApplication from PyQt5.QtWidgets import (QApplication, QHBoxLayout, QLabel, QWidget, QPushButton, QProgressBar, QComboBox, QStyle, QVBoxLayout) #sys.setswitchinterval(sys.getswitchinterval() / 100) run_flag = False PROGRESS_RANGE = 500 TIMER_INTERVAL = 20 THREAD_PRIORITIES = [ (QThread.IdlePriority, 'IdlePriority'), (QThread.LowestPriority, 'LowestPriority'), (QThread.LowPriority, 'LowPriority'), (QThread.NormalPriority, 'NormalPriority'), (QThread.HighPriority, 'HighPriority'), (QThread.HighestPriority, 'HighestPriority'), (QThread.TimeCriticalPriority, 'TimeCriticalPriority'), (QThread.InheritPriority, 'InheritPriority') ] elapsed_samples_ms = [] class Task(QThread): def __init__(self, n): super().__init__() self.n = n self._timer = QElapsedTimer() def run(self): global run_flag # some heavy tasks print('task #{} started'.format(self.n)) self._timer.start() k = 0 i = 0 while run_flag and i < 10000: j = 0 while run_flag and j < 50000: k += 1 j += 1 i += 1 elapsed_s = self._timer.elapsed() / 1000 print(f'task #{self.n} finished: {elapsed_s}s') class SleepyTask(QThread): def __init__(self, n): super().__init__() self.n = n self._timer = QElapsedTimer() def run(self): global run_flag # some heavy tasks print('Sleepy task #{} started'.format(self.n)) self._timer.start() k = 0 i = 0 while run_flag and i < 10000: for j in range(50000): k += 1 # this helps PySide2 when there are only few tasks. if k % 100 == 0: time.sleep(0) if k % 1000000 == 0: print(self.n, end=" ") sys.stdout.flush() i += 1 elapsed_s = self._timer.elapsed() / 1000 print(f'Sleepy task #{self.n} finished: {elapsed_s}s') class Gui(QWidget): def __init__(self): super().__init__() self.threads = [] self.start_button = QPushButton('Start Thread') self.start_button.clicked.connect(self.start_task) self.start_sleepy_button = QPushButton('Start Sleepy Thread') self.start_sleepy_button.clicked.connect(self.start_sleepy_task) self.stop_button = QPushButton('Stop 0 threads') self.stop_button.clicked.connect(self.stop_tasks) self.quit_button = QPushButton('Quit') self.quit_button.clicked.connect(self.close) build = 'debug' if QLibraryInfo.isDebugBuild() else 'release' p = QGuiApplication.platformName() s = QApplication.style().objectName() description = f'{MODULE} Qt {qVersion()} {p} {build} {s}' combobox = QComboBox() combobox.addItems(['1', '2', '3', '4', '5']) layout = QVBoxLayout(self) layout.addWidget(QLabel(description)) layout.addWidget(QLabel(f'Python {sys.version}')) hLayout = QHBoxLayout() layout.addLayout(hLayout) hLayout.addWidget(self.start_button) hLayout.addWidget(self.start_sleepy_button) hLayout.addWidget(self.stop_button) hLayout.addWidget(self.quit_button) layout.addWidget(combobox) self.priority_combo = QComboBox() for tp in THREAD_PRIORITIES: self.priority_combo.addItem(tp[1]) layout.addWidget(self.priority_combo) self.priority_combo.setCurrentIndex(7) self.progressBar = QProgressBar() layout.addWidget(self.progressBar) self.progressBar.setRange(0, PROGRESS_RANGE) self.progressValue = 0 self.progressBar.setValue(self.progressValue) progressBarTimer = QTimer(self) progressBarTimer.setTimerType(Qt.PreciseTimer) progressBarTimer.setInterval(TIMER_INTERVAL) progressBarTimer.timeout.connect(self.timeout) progressBarTimer.start() self.elapsedTimer = QElapsedTimer() self.elapsedTimer.start() layout.addStretch() self.enable_buttons() self.setWindowTitle('{} {}'.format(MODULE, qVersion())) def timeout(self): """Main thread time which counts 10s in 500 20ms steps and displays elapsed interval""" self.progressValue += 1 if self.progressValue >= PROGRESS_RANGE: elapsedMs = self.elapsedTimer.elapsed() self.elapsedTimer.restart() self.progressValue = 0 r = TIMER_INTERVAL * PROGRESS_RANGE print(f'timeout: elapsed {elapsedMs}ms, vs timer interval {r}ms') elapsed_samples_ms.append(elapsedMs) self.progressBar.setValue(self.progressValue) def enable_buttons(self): thread_count = len(self.threads) self.stop_button.setEnabled(thread_count > 0) self.stop_button.setText(f'Stop {thread_count} threads') def thread_priority(self): index = self.priority_combo.currentIndex() return THREAD_PRIORITIES[index][0] def start_task(self): self._start_task(Task(len(self.threads))) def start_sleepy_task(self): self._start_task(SleepyTask(len(self.threads))) def _start_task(self, thread): global run_flag run_flag = True self.threads.append(thread) thread.finished.connect(self.thread_finished) thread.start() priority = self.thread_priority() if priority != QThread.InheritPriority: thread.setPriority(priority) print(thread.priority()) self.enable_buttons() def thread_finished(self): new_thread_list = [] for t in self.threads: if t.isRunning(): new_thread_list.append(t) self.threads = new_thread_list self.enable_buttons() def stop_tasks(self): if self.threads: global run_flag print('stopping {}'.format(len(self.threads))) run_flag = False for t in self.threads: if t.isRunning(): t.wait() def closeEvent(self, event): print('closeEvent') self.stop_tasks() event.accept() if __name__ == '__main__': print("Python ", sys.version) if MODULE.startswith('PySide'): print(QLibraryInfo.build()) print("\n") app = QApplication(sys.argv) if not result.animations: print("Disabling UI_AnimateMenu, UI_AnimateCombo") app.setEffectEnabled(Qt.UI_AnimateMenu, False) app.setEffectEnabled(Qt.UI_AnimateCombo, False) window = Gui() window.show() ex = app.exec_() if elapsed_samples_ms: sum = 0 for e in elapsed_samples_ms: sum += e mean = sum / len(elapsed_samples_ms) print(f'Mean elapsed {mean}ms') sys.exit(ex)