当另一个对象在别处实例化时，我如何接收来自另一个对象的 PyQt 信号？
How do I receive a PyQt signal from another object when it is instantiated elsewhere?
我正在构建一个系统，通过蓝牙 LE 将基于 Arduino 的传感器连接到 RPi，并在 GUI 上显示信息（温度和电池寿命）。我的程序中有两个主要 类，一个管理 GUI，一个管理 BLE 连接 (class HubSensor)。 HubSensor 对象采用每个传感器的 MAC 地址，并应该发出一个带有附加元组的信号，该元组包含感测到的温度、电池寿命和一个索引整数，让主程序知道哪个传感器这是。 HubSensor 每秒获取一次信息，应该每次都发出信号。 （已经建立了输入验证，但它与我的问题无关。）到目前为止，大部分工作正常。
我的问题是我不知道如何创建一个插槽来接收信号，以便它可以更新显示（并稍后在 CSV 文件中保存日志）。我正在使用 BluePy 库来管理 BLE 连接，这对我来说有其自身的额外挑战。
所以，这就是我的程序的工作方式（我认为）。每个线程（因为我有多个传感器）创建一个 HubSensor 对象。当对象建立 BLE 连接时，它会创建一个 MyDelegate 对象（从 BluePy 的 DefaultDelegate 继承而来）。在 MyDelegate 对象内部，会发出 Qt 信号。我需要在外部访问该信号所有这些 类 并且因为我不知道创建的 MyDelegate 对象的名称，所以我不知道如何访问它。
我试过让上面提到的每个 类 继承彼此的特征，但我不确定我做对了。
来自trailerTempDisplay.py
import sys
from PyQt5.QtCore import *
from HubSensor import *
from PyQt5 import QtWidgets, uic
from bluepy.btle import *
from datetime import datetime
# mac addresses for the sensors. later, this will need a function to allow new devices to connect
bt_addrs = ['c1:49:02:59:ae:50', 'f3:ad:ed:46:ea:16']
app = QtWidgets.QApplication(sys.argv)
class Worker(QRunnable):
def __init__(self, macAddress, ind):
super(Worker, self).__init__()
self.macAddress = macAddress
self.ind = ind
@pyqtSlot()
#this is where each sensor exists. each object is created and runs here
def run(self):
self.sensor = HubSensor(self.macAddress, self.ind)
self.sensor.notified.connect(self.updateValues())
#close button
def buttonClicked():
app.closeAllWindows()
window = uic.loadUi("mainwindow.ui")
window.pushButton.clicked.connect(buttonClicked)
def updateValues(self):
print("value updated") # debugging
window.show()
window.threadpool = QThreadPool()
index = 0
for addr in bt_addrs:
worker = Worker(addr, index)
index += 1
window.threadpool.start(worker)
app.exec()
来自HubSensor.py
from bluepy.btle import *
from PyQt5.QtCore import QObject, pyqtSignal
class MyDelegate(DefaultDelegate, QObject):
def __init__(self, index):
DefaultDelegate.__init__(self)
QObject.__init__(self)
self.index = index
# class variable for the notified signal
notified = pyqtSignal(tuple)
def handleNotification(self, cHandle, data):
# exception handling prevents bad data from being passed. cHandle is not used but might be useful later
try:
# defining the sensorData tuple. 1, 2, and 3 are dummy values
self.sensorData = (1, 2, 3)
self.notified.emit(self.sensorData) # this should emit a signal every time the function is called and send a tuple with temp, battery, and sensor index(id)
except (ValueError, IndexError):
pass
class HubSensor(MyDelegate):
# constructor. connects to device defined by mac address and position.
# uuid is static and should not change
def __init__(self, mac, index):
self.index = index # keeps track of sensor position
self.mac = mac
self.p = Peripheral(self.mac, 'random') # adafruit feathers must be 'random' for some reason
self.p.setDelegate(MyDelegate(self.index))
self.p.writeCharacteristic(35, b'\x01\x00') # writing these bits to handle '35' enables notifications
while True:
if self.p.waitForNotifications(1):
# when a bluetooth notification is received, handleNotification is called
continue
当我运行程序时，"value updated"不显示在控制台上。它应该每秒弹出两次，然后重复。稍后，我会在将传递的值转换为GUI显示的部分添加。
让我提前道歉，因为我还是一个初学者。我想我已经包含了代码的所有相关部分，但我不确定。另外，我很确定我在某些地方的术语是不正确的，所以我希望你们都能理解我的意思。预先感谢您能给我的任何帮助！
您的代码令人困惑，例如 HubSensor 是一个属性为 Peripheral 的委托，而 Peripheral 有另一个委托，等等。
因此，我没有依赖您的代码，而是创建了 PeripheralManager class，它将通过信号通知指定外围设备接收到的信息。在无限循环的情况下，它将在使用 threading.Thread.
的线程中处理
import threading
from PyQt5 import QtCore, QtWidgets, uic
from bluepy.btle import DefaultDelegate, Peripheral, BTLEDisconnectError
class PeripheralManager(QtCore.QObject, DefaultDelegate):
notified = QtCore.pyqtSignal(bytes)
def __init__(self, peripheral, parent=None):
super().__init__(parent)
self._peripheral = peripheral
self.peripheral.setDelegate(self)
self.peripheral.writeCharacteristic(35, b"\x01\x00")
threading.Thread(target=self._manage_notifications, daemon=True).start()
@property
def peripheral(self):
return self._peripheral
def handleNotification(self, cHandle, data):
self.notified.emit(self.data)
def _manage_notifications(self):
while self.peripheral.waitForNotifications(1):
continue
def buttonClicked():
QtWidgets.QApplication.closeAllWindows()
def updateValues(values):
print("value updated", values)
def main(args):
app = QtWidgets.QApplication(args)
bt_addrs = ["c1:49:02:59:ae:50", "f3:ad:ed:46:ea:16"]
managers = []
for addr in bt_addrs:
try:
p = Peripheral(addr, "random")
except BTLEDisconnectError as e:
print(e)
else:
manager = PeripheralManager(p)
manager.notified.connect(updateValues)
managers.append(manager)
window = uic.loadUi("mainwindow.ui")
window.pushButton.clicked.connect(buttonClicked)
window.show()
ret = app.exec_()
return ret
if __name__ == "__main__":
import sys
sys.exit(main(sys.argv))
我正在构建一个系统，通过蓝牙 LE 将基于 Arduino 的传感器连接到 RPi，并在 GUI 上显示信息（温度和电池寿命）。我的程序中有两个主要 类，一个管理 GUI，一个管理 BLE 连接 (class HubSensor)。 HubSensor 对象采用每个传感器的 MAC 地址，并应该发出一个带有附加元组的信号，该元组包含感测到的温度、电池寿命和一个索引整数，让主程序知道哪个传感器这是。 HubSensor 每秒获取一次信息，应该每次都发出信号。 （已经建立了输入验证，但它与我的问题无关。）到目前为止，大部分工作正常。
我的问题是我不知道如何创建一个插槽来接收信号，以便它可以更新显示（并稍后在 CSV 文件中保存日志）。我正在使用 BluePy 库来管理 BLE 连接，这对我来说有其自身的额外挑战。
所以，这就是我的程序的工作方式（我认为）。每个线程（因为我有多个传感器）创建一个 HubSensor 对象。当对象建立 BLE 连接时，它会创建一个 MyDelegate 对象（从 BluePy 的 DefaultDelegate 继承而来）。在 MyDelegate 对象内部，会发出 Qt 信号。我需要在外部访问该信号所有这些 类 并且因为我不知道创建的 MyDelegate 对象的名称，所以我不知道如何访问它。
我试过让上面提到的每个 类 继承彼此的特征，但我不确定我做对了。
来自trailerTempDisplay.py
import sys
from PyQt5.QtCore import *
from HubSensor import *
from PyQt5 import QtWidgets, uic
from bluepy.btle import *
from datetime import datetime
# mac addresses for the sensors. later, this will need a function to allow new devices to connect
bt_addrs = ['c1:49:02:59:ae:50', 'f3:ad:ed:46:ea:16']
app = QtWidgets.QApplication(sys.argv)
class Worker(QRunnable):
def __init__(self, macAddress, ind):
super(Worker, self).__init__()
self.macAddress = macAddress
self.ind = ind
@pyqtSlot()
#this is where each sensor exists. each object is created and runs here
def run(self):
self.sensor = HubSensor(self.macAddress, self.ind)
self.sensor.notified.connect(self.updateValues())
#close button
def buttonClicked():
app.closeAllWindows()
window = uic.loadUi("mainwindow.ui")
window.pushButton.clicked.connect(buttonClicked)
def updateValues(self):
print("value updated") # debugging
window.show()
window.threadpool = QThreadPool()
index = 0
for addr in bt_addrs:
worker = Worker(addr, index)
index += 1
window.threadpool.start(worker)
app.exec()
来自HubSensor.py
from bluepy.btle import *
from PyQt5.QtCore import QObject, pyqtSignal
class MyDelegate(DefaultDelegate, QObject):
def __init__(self, index):
DefaultDelegate.__init__(self)
QObject.__init__(self)
self.index = index
# class variable for the notified signal
notified = pyqtSignal(tuple)
def handleNotification(self, cHandle, data):
# exception handling prevents bad data from being passed. cHandle is not used but might be useful later
try:
# defining the sensorData tuple. 1, 2, and 3 are dummy values
self.sensorData = (1, 2, 3)
self.notified.emit(self.sensorData) # this should emit a signal every time the function is called and send a tuple with temp, battery, and sensor index(id)
except (ValueError, IndexError):
pass
class HubSensor(MyDelegate):
# constructor. connects to device defined by mac address and position.
# uuid is static and should not change
def __init__(self, mac, index):
self.index = index # keeps track of sensor position
self.mac = mac
self.p = Peripheral(self.mac, 'random') # adafruit feathers must be 'random' for some reason
self.p.setDelegate(MyDelegate(self.index))
self.p.writeCharacteristic(35, b'\x01\x00') # writing these bits to handle '35' enables notifications
while True:
if self.p.waitForNotifications(1):
# when a bluetooth notification is received, handleNotification is called
continue
当我运行程序时，"value updated"不显示在控制台上。它应该每秒弹出两次，然后重复。稍后，我会在将传递的值转换为GUI显示的部分添加。
让我提前道歉，因为我还是一个初学者。我想我已经包含了代码的所有相关部分，但我不确定。另外，我很确定我在某些地方的术语是不正确的，所以我希望你们都能理解我的意思。预先感谢您能给我的任何帮助！
您的代码令人困惑，例如 HubSensor 是一个属性为 Peripheral 的委托，而 Peripheral 有另一个委托，等等。
因此，我没有依赖您的代码，而是创建了 PeripheralManager class，它将通过信号通知指定外围设备接收到的信息。在无限循环的情况下，它将在使用 threading.Thread.
的线程中处理import threading
from PyQt5 import QtCore, QtWidgets, uic
from bluepy.btle import DefaultDelegate, Peripheral, BTLEDisconnectError
class PeripheralManager(QtCore.QObject, DefaultDelegate):
notified = QtCore.pyqtSignal(bytes)
def __init__(self, peripheral, parent=None):
super().__init__(parent)
self._peripheral = peripheral
self.peripheral.setDelegate(self)
self.peripheral.writeCharacteristic(35, b"\x01\x00")
threading.Thread(target=self._manage_notifications, daemon=True).start()
@property
def peripheral(self):
return self._peripheral
def handleNotification(self, cHandle, data):
self.notified.emit(self.data)
def _manage_notifications(self):
while self.peripheral.waitForNotifications(1):
continue
def buttonClicked():
QtWidgets.QApplication.closeAllWindows()
def updateValues(values):
print("value updated", values)
def main(args):
app = QtWidgets.QApplication(args)
bt_addrs = ["c1:49:02:59:ae:50", "f3:ad:ed:46:ea:16"]
managers = []
for addr in bt_addrs:
try:
p = Peripheral(addr, "random")
except BTLEDisconnectError as e:
print(e)
else:
manager = PeripheralManager(p)
manager.notified.connect(updateValues)
managers.append(manager)
window = uic.loadUi("mainwindow.ui")
window.pushButton.clicked.connect(buttonClicked)
window.show()
ret = app.exec_()
return ret
if __name__ == "__main__":
import sys
sys.exit(main(sys.argv))