-
Notifications
You must be signed in to change notification settings - Fork 3
/
QtThreading.py
108 lines (85 loc) · 2.85 KB
/
QtThreading.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
# -*- coding: utf-8 -*-
"""
Basic classes for Threading a Qt application
Created on Wed Feb 6 11:10:36 2019
@author: ofn77899
"""
#https://www.geeksforgeeks.org/migrate-pyqt5-app-to-pyside2/
import traceback
import sys
from PySide2 import QtCore
from PySide2.QtCore import Slot
class Worker(QtCore.QRunnable):
"""
Worker thread
Inherits from QRunnable to handle worker thread setup, signals and wrapup.
:param (function) callback:
The function callback to run on this worker thread. Supplied
args/kwargs will be pass to the runner.
:param args:
Arguments to pass to the callback function
:param kwargs:
Keyword arguments to pass to the callback function
"""
def __init__(self, fn, *args, **kwargs):
super(Worker, self).__init__()
self.fn = fn
self.args = args
self.kwargs = kwargs
self.signals = WorkerSignals()
# Add progress callback to kwargs
self.kwargs['progress_callback'] = self.signals.progress
self.kwargs['message_callback'] = self.signals.message
self.kwargs['status_callback'] = self.signals.status
@Slot()
def run(self):
"""
Run the worker. Emits signals based on run state.
Signals:
- Error: Emitted when an exception is thrown in the workers function.
- Result: Emitted if function completes successfully. Contains the return value of the function.
- Finished: Emitted on completion of the worker thread.
"""
try:
result = self.fn(*self.args, **self.kwargs)
except:
traceback.print_exc()
exctype, value = sys.exc_info()[:2]
self.signals.error.emit((exctype, value, traceback.format_exc()))
else:
self.signals.result.emit(result)
finally:
self.signals.finished.emit()
class WorkerSignals(QtCore.QObject):
"""
Defines signals available when running a worker thread
Supported Signals:
finished
No Data
error
`tuple` (exctype, value, traceback.format_exc() )
result
`object` data returned from processing, anything
progress
`int` indicating % progress
"""
finished = QtCore.Signal()
error = QtCore.Signal(tuple)
result = QtCore.Signal(object)
progress = QtCore.Signal(int)
message = QtCore.Signal(str)
status = QtCore.Signal(tuple)
class ErrorObserver:
def __init__(self):
self.__ErrorOccurred = False
self.__ErrorMessage = None
self.CallDataType = 'string0'
def __call__(self, obj, event, message):
self.__ErrorOccurred = True
self.__ErrorMessage = message
def ErrorOccurred(self):
occ = self.__ErrorOccurred
self.__ErrorOccurred = False
return occ
def ErrorMessage(self):
return self.__ErrorMessage