Qthread 读取 json 值并发送 json 与切换按钮问题
Qthread with reading json values and sending json with toggle button issues
我遇到了一些问题，当切换按钮设置为关闭时它会返回到打开，因为 json 读取仍然没有刷新值并启动 slot_method。但是我需要这个 if 语句来在启动应用程序时读取初始值，并且可以从硬件控制器中选择此模式。如何修改在切换按钮上单击 1 json 读取线程可以被排除？
py_toggle.py
from PySide2.QtCore import *
from PySide2.QtGui import *
from PySide2.QtWidgets import *
class PyToggle(QCheckBox):
def __init__(
self,
width=60,
bg_color="#777",
circle_color="#DDD",
active_color="#3A3A66",
animation_curve=QEasingCurve.OutBounce
):
QCheckBox.__init__(self)
# SET DEFAULT PARAMETERS
self.setFixedSize(width, 28)
self.setCursor(Qt.PointingHandCursor)
# COLORS
self._bg_color = bg_color
self._circle_color = circle_color
self._active_color = active_color
# CREATE ANIMATION
self._circle_position = 3
self.animation = QPropertyAnimation(self, b"circle_position", self)
self.animation.setEasingCurve(animation_curve)
self.animation.setDuration(400) # Time in milliseconds
# CONNECT STATE CHANGED
# self.stateChanged.connect(self.debug)
self.stateChanged.connect(self.start_transition)
# self.stateChanged.connect()
# CREATE NEW SET AND GET PROPERTIE
@Property(float) # Decorator Getter
def circle_position(self):
return self._circle_position
@circle_position.setter
def circle_position(self, pos):
self._circle_position = pos
self.update()
# def state(self):
# print(f"Status: {self.isChecked()}")
def start_transition(self, value):
self.animation.stop() # Stop animation if running
if value:
self.animation.setEndValue(self.width() - 26)
else:
self.animation.setEndValue(3)
# START ANIMATION
self.animation.start()
# SET NEW HIT AREA
def hitButton(self, pos: QPoint):
return self.contentsRect().contains(pos)
# DRAW NEW ITEMS
def paintEvent(self, e):
p = QPainter(self)
p.setRenderHint(QPainter.Antialiasing)
# SET AS NO PEN
p.setPen(Qt.NoPen)
# DRAW RECTANGLE
rect = QRect(0, 0, self.width(), self.height())
if not self.isChecked():
# DRAW BG
p.setBrush(QColor(self._bg_color))
p.drawRoundedRect(0, 0, rect.width(), self.height(), self.height() / 2, self.height() / 2)
# DRAW CIRCLE
p.setBrush(QColor(self._circle_color))
p.drawEllipse(self._circle_position, 3, 22, 22)
else:
# DRAW BG
p.setBrush(QColor(self._active_color))
p.drawRoundedRect(0, 0, rect.width(), self.height(), self.height() / 2, self.height() / 2)
# DRAW CIRCLE
p.setBrush(QColor(self._circle_color))
p.drawEllipse(self._circle_position, 3, 22, 22)
# END DRAW
p.end()
test.py
import sys
import requests
from PySide2.QtCore import (QTimer, QThread, Signal)
from PySide2.QtWidgets import *
from py_toggle import PyToggle
class WorkerThread(QThread):
measurements_signals = Signal(str, name = 'm_signals') # declare the signal
def __init__(self, parent=None):
QThread.__init__(self)
self.timer = QTimer()
self.timer.timeout.connect(lambda: WorkerThread.run(self))
self.timer.setInterval(6000) # 6000ms = 6s
self.timer.start()
def run(self):
url = "http://192.168.8.150/json"
try:
res = requests.get(url)
msg = res.json()
print(msg)
try:
if res.status_code == 200:
quiet = msg["heatpump"][18]["Value"]
self.measurements_signals.emit(quiet)
else:
print("Not Working")
except requests.exceptions.InvalidURL or requests.exceptions.ConnectionError as err:
print(err)
except requests.exceptions.InvalidURL or requests.exceptions.ConnectionError as err:
print(err)
def stop(self):
self.terminate()
print("stop")
class Tester(QWidget):
def __init__(self, parent = None):
super(Tester, self).__init__(parent)
# ==> TOGGLE BUTTON1
self.setWindowTitle("Test")
self.resize(200, 150)
layout = QGridLayout()
self.toggle = PyToggle()
layout.addWidget(self.toggle)
self.toggle.stateChanged.connect(self.postCommand)
self.setLayout(layout)
# ==> Worker Thread start
self.wt = WorkerThread() # This is the thread object
self.wt.start()
# Connect the signal from the thread to the slot_method
self.wt.measurements_signals.connect(self.slot_method) ### 3) connect to the slot
app.aboutToQuit.connect(self.wt.stop) # to stop the thread when closing the GUI
def slot_method(self, quiet):
if quiet == "1":
self.toggle.setChecked(True)
def postCommand(self):
if self.toggle.isChecked():
setting = "SetQuietMode=1"
else:
setting = "SetQuietMode=0"
url = f"http://192.168.8.150/command?{setting}"
r = requests.request('GET', url)
if __name__ == "__main__":
app = QApplication(sys.argv)
form = Tester()
form.show()
sys.exit(app.exec_())
你的实现有两个问题。
第一个是 QThread 实际上只是第一次在单独的线程中运行该函数（当调用 start() 时），而 QTimer 将始终在主线程中执行它, 它是在哪里创建的。 run() 应该 永远不会 被显式调用。
那么，这就是您问题的重点，信号应该始终仅在值更改时发出，而不是每次被请求时发出。
最后，由于请求可能会阻塞，因此从 UI 发送的命令也应该被线程化。我们已经有一个线程 运行，因此我们可以将其与 python Queue.
一起使用
还有其他问题：
- 作为 documentation explicitly says，不鼓励使用
terminate()，因为使用它可能会导致问题和稳定性问题；使用简单的标志更简单、更安全；在任何情况下，使用您的原始代码它都不会 nothing ，除非在线程启动后立即调用：如上所述，每次进一步执行都是在错误的线程中完成的，所以 terminate() 不会做任何有用的事情；
- 第二个
try/except 块毫无意义，因为那里不可能出现那种类型的异常；在这一点上唯一有意义的例外是用于字典查找的 KeyError；
stateChanged 信号用于 tri-state 复选框，对于标准双状态，正确的信号是 toggled;
正确的实现必须使线程 运行 处于 while 循环中，并在必要时最终退出。使用队列 timeout 参数我们也得到 6 秒的间隔，所以不需要定时器。
from queue import Queue, Empty
class WorkerThread(QThread):
measurements_signals = Signal(str, name = 'm_signals')
def __init__(self, parent=None):
QThread.__init__(self, parent)
self.queue = Queue()
def run(self):
self.keepRunning = True
url = "http://192.168.8.150/"
quiet = None
while self.keepRunning:
try:
res = requests.get(url + 'json')
msg = res.json()
print(msg)
if res.status_code == 200:
new = msg["heatpump"][18]["Value"]
if new != quiet:
quiet = new
self.measurements_signals.emit(quiet)
else:
print("Not Working")
except requests.exceptions.InvalidURL or requests.exceptions.ConnectionError as err:
print(err)
try:
q = self.queue.get(timeout=6)
if q == -1:
break
cmd, value = q
if cmd == 'SetQuietMode':
quiet = value
requests.request('GET', f'{url}command?{cmd}={value}')
except Empty:
pass
def stop(self):
print("stop")
self.keepRunning = False
self.queue.put(-1)
self.wait()
def setQuiet(self, state):
self.queue.put(('SetQuietMode', int(state)))
class Tester(QWidget):
def __init__(self, parent=None):
# ...
self.wt = WorkerThread()
self.wt.measurements_signals.connect(self.slot_method)
app.aboutToQuit.connect(self.wt.stop)
self.toggle.toggled.connect(self.wt.setQuiet)
# it's usually better to start the thread *after* connecting signals
self.wt.start()
def slot_method(self, quiet):
if quiet == "1":
# temporarily disconnect to avoid calling setQuiet unnecessarily
self.toggle.toggled.disconnect(self.wt.setQuiet)
self.toggle.setChecked(True)
self.toggle.toggled.connect(self.wt.setQuiet)
请注意，QtNetwork 模块提供了 QNetworkAccessManager，它已经可以异步工作了。使用它，您可以避免创建单独的线程 class，并且您将能够更轻松地扩展程序的功能。
我遇到了一些问题，当切换按钮设置为关闭时它会返回到打开，因为 json 读取仍然没有刷新值并启动 slot_method。但是我需要这个 if 语句来在启动应用程序时读取初始值，并且可以从硬件控制器中选择此模式。如何修改在切换按钮上单击 1 json 读取线程可以被排除？
py_toggle.py
from PySide2.QtCore import *
from PySide2.QtGui import *
from PySide2.QtWidgets import *
class PyToggle(QCheckBox):
def __init__(
self,
width=60,
bg_color="#777",
circle_color="#DDD",
active_color="#3A3A66",
animation_curve=QEasingCurve.OutBounce
):
QCheckBox.__init__(self)
# SET DEFAULT PARAMETERS
self.setFixedSize(width, 28)
self.setCursor(Qt.PointingHandCursor)
# COLORS
self._bg_color = bg_color
self._circle_color = circle_color
self._active_color = active_color
# CREATE ANIMATION
self._circle_position = 3
self.animation = QPropertyAnimation(self, b"circle_position", self)
self.animation.setEasingCurve(animation_curve)
self.animation.setDuration(400) # Time in milliseconds
# CONNECT STATE CHANGED
# self.stateChanged.connect(self.debug)
self.stateChanged.connect(self.start_transition)
# self.stateChanged.connect()
# CREATE NEW SET AND GET PROPERTIE
@Property(float) # Decorator Getter
def circle_position(self):
return self._circle_position
@circle_position.setter
def circle_position(self, pos):
self._circle_position = pos
self.update()
# def state(self):
# print(f"Status: {self.isChecked()}")
def start_transition(self, value):
self.animation.stop() # Stop animation if running
if value:
self.animation.setEndValue(self.width() - 26)
else:
self.animation.setEndValue(3)
# START ANIMATION
self.animation.start()
# SET NEW HIT AREA
def hitButton(self, pos: QPoint):
return self.contentsRect().contains(pos)
# DRAW NEW ITEMS
def paintEvent(self, e):
p = QPainter(self)
p.setRenderHint(QPainter.Antialiasing)
# SET AS NO PEN
p.setPen(Qt.NoPen)
# DRAW RECTANGLE
rect = QRect(0, 0, self.width(), self.height())
if not self.isChecked():
# DRAW BG
p.setBrush(QColor(self._bg_color))
p.drawRoundedRect(0, 0, rect.width(), self.height(), self.height() / 2, self.height() / 2)
# DRAW CIRCLE
p.setBrush(QColor(self._circle_color))
p.drawEllipse(self._circle_position, 3, 22, 22)
else:
# DRAW BG
p.setBrush(QColor(self._active_color))
p.drawRoundedRect(0, 0, rect.width(), self.height(), self.height() / 2, self.height() / 2)
# DRAW CIRCLE
p.setBrush(QColor(self._circle_color))
p.drawEllipse(self._circle_position, 3, 22, 22)
# END DRAW
p.end()
test.py
import sys
import requests
from PySide2.QtCore import (QTimer, QThread, Signal)
from PySide2.QtWidgets import *
from py_toggle import PyToggle
class WorkerThread(QThread):
measurements_signals = Signal(str, name = 'm_signals') # declare the signal
def __init__(self, parent=None):
QThread.__init__(self)
self.timer = QTimer()
self.timer.timeout.connect(lambda: WorkerThread.run(self))
self.timer.setInterval(6000) # 6000ms = 6s
self.timer.start()
def run(self):
url = "http://192.168.8.150/json"
try:
res = requests.get(url)
msg = res.json()
print(msg)
try:
if res.status_code == 200:
quiet = msg["heatpump"][18]["Value"]
self.measurements_signals.emit(quiet)
else:
print("Not Working")
except requests.exceptions.InvalidURL or requests.exceptions.ConnectionError as err:
print(err)
except requests.exceptions.InvalidURL or requests.exceptions.ConnectionError as err:
print(err)
def stop(self):
self.terminate()
print("stop")
class Tester(QWidget):
def __init__(self, parent = None):
super(Tester, self).__init__(parent)
# ==> TOGGLE BUTTON1
self.setWindowTitle("Test")
self.resize(200, 150)
layout = QGridLayout()
self.toggle = PyToggle()
layout.addWidget(self.toggle)
self.toggle.stateChanged.connect(self.postCommand)
self.setLayout(layout)
# ==> Worker Thread start
self.wt = WorkerThread() # This is the thread object
self.wt.start()
# Connect the signal from the thread to the slot_method
self.wt.measurements_signals.connect(self.slot_method) ### 3) connect to the slot
app.aboutToQuit.connect(self.wt.stop) # to stop the thread when closing the GUI
def slot_method(self, quiet):
if quiet == "1":
self.toggle.setChecked(True)
def postCommand(self):
if self.toggle.isChecked():
setting = "SetQuietMode=1"
else:
setting = "SetQuietMode=0"
url = f"http://192.168.8.150/command?{setting}"
r = requests.request('GET', url)
if __name__ == "__main__":
app = QApplication(sys.argv)
form = Tester()
form.show()
sys.exit(app.exec_())
你的实现有两个问题。
第一个是 QThread 实际上只是第一次在单独的线程中运行该函数（当调用 start() 时），而 QTimer 将始终在主线程中执行它, 它是在哪里创建的。 run() 应该 永远不会 被显式调用。
那么，这就是您问题的重点，信号应该始终仅在值更改时发出，而不是每次被请求时发出。
最后，由于请求可能会阻塞，因此从 UI 发送的命令也应该被线程化。我们已经有一个线程 运行，因此我们可以将其与 python Queue.
还有其他问题：
- 作为 documentation explicitly says，不鼓励使用
terminate()，因为使用它可能会导致问题和稳定性问题；使用简单的标志更简单、更安全；在任何情况下，使用您的原始代码它都不会 nothing ，除非在线程启动后立即调用：如上所述，每次进一步执行都是在错误的线程中完成的，所以terminate()不会做任何有用的事情； - 第二个
try/except块毫无意义，因为那里不可能出现那种类型的异常；在这一点上唯一有意义的例外是用于字典查找的KeyError； stateChanged信号用于 tri-state 复选框，对于标准双状态，正确的信号是toggled;
正确的实现必须使线程 运行 处于 while 循环中，并在必要时最终退出。使用队列 timeout 参数我们也得到 6 秒的间隔，所以不需要定时器。
from queue import Queue, Empty
class WorkerThread(QThread):
measurements_signals = Signal(str, name = 'm_signals')
def __init__(self, parent=None):
QThread.__init__(self, parent)
self.queue = Queue()
def run(self):
self.keepRunning = True
url = "http://192.168.8.150/"
quiet = None
while self.keepRunning:
try:
res = requests.get(url + 'json')
msg = res.json()
print(msg)
if res.status_code == 200:
new = msg["heatpump"][18]["Value"]
if new != quiet:
quiet = new
self.measurements_signals.emit(quiet)
else:
print("Not Working")
except requests.exceptions.InvalidURL or requests.exceptions.ConnectionError as err:
print(err)
try:
q = self.queue.get(timeout=6)
if q == -1:
break
cmd, value = q
if cmd == 'SetQuietMode':
quiet = value
requests.request('GET', f'{url}command?{cmd}={value}')
except Empty:
pass
def stop(self):
print("stop")
self.keepRunning = False
self.queue.put(-1)
self.wait()
def setQuiet(self, state):
self.queue.put(('SetQuietMode', int(state)))
class Tester(QWidget):
def __init__(self, parent=None):
# ...
self.wt = WorkerThread()
self.wt.measurements_signals.connect(self.slot_method)
app.aboutToQuit.connect(self.wt.stop)
self.toggle.toggled.connect(self.wt.setQuiet)
# it's usually better to start the thread *after* connecting signals
self.wt.start()
def slot_method(self, quiet):
if quiet == "1":
# temporarily disconnect to avoid calling setQuiet unnecessarily
self.toggle.toggled.disconnect(self.wt.setQuiet)
self.toggle.setChecked(True)
self.toggle.toggled.connect(self.wt.setQuiet)
请注意，QtNetwork 模块提供了 QNetworkAccessManager，它已经可以异步工作了。使用它，您可以避免创建单独的线程 class，并且您将能够更轻松地扩展程序的功能。