Args:
parent: Parent Widget
Returns:
self
"""
xpInstance = pyqtSignal(QHostAddress, int)
drefUpdate = pyqtSignal(dict) # will only bind if in a class and is instantiated
def __init__(self, parent=None):
print("inside class pyG5_PX4_UDPListener(QObject): __init__")
"""Object constructor.
Args:
parent: Parent Widget
Returns:
self
"""
QObject.__init__(self, parent)
self.logger = logging.getLogger(self.__class__.__name__)
self.XPAddr = QHostAddress("127.0.0.1") #"239.255.1.1")
self.XPPort = 65432 #49707
# create the socket
self.udpSock = QUdpSocket(self)
self.udpSock.stateChanged.connect(self.stateChangedSlot)
self.udpSock.readyRead.connect(self.udpData)
self.udpSock.connected.connect(self.connectedSlot)
self.udpSock.bind(QHostAddress.Null, self.XPPort, QUdpSocket.ShareAddress) #self.udpSock.bind(QHostAddress.AnyIPv4, self.XPPort, QUdpSocket.ShareAddress)
#if not self.udpSock.joinMulticastGroup(self.XPAddr):
# logging.error("Failed to join multicast group")
@pyqtSlot(QAbstractSocket.SocketState)
def stateChangedSlot(self, state):
"""stateChangedSlot."""
self.logger.debug("Sock new state: {}".format(state))
@pyqtSlot()
def connectedSlot(self):
"""connectedSlot."""
self.logger.debug("udp connected: {}".format(self.udpSock.state()))
@pyqtSlot()
def udpData(self):
"""udpData."""
print("inside class pyG5NetWorkManager(QObject): udpData())")
while self.udpSock.hasPendingDatagrams():
data = self.udpSock.receiveDatagram()
print("inside class pyG5NetWorkManager(QObject): len(data), data= ", len(data.data()), " , ", data.data() )
if (len(data.data()) < 5):
return
data_dict = pickle.loads(data.data())
# if data_dict has entries then emit
print("Received data_dict= ", data_dict )
for k, v in data_dict.items():
print(k, v)
print("Received len(data_dict), len(data_dict.keys(), data_dict.keys()=", len(data_dict), len(data_dict.keys()), data_dict.keys() )
print("data_dict=", data_dict["'roll'"])
print("data_dict=", data_dict["'pitch'"])
print("data_dict=", data_dict["'yaw'"])
retvalues = {}
datarefs = [
# ( dataref, frequency, unit, description, num decimals to display in formatted output )
(
"sim/flightmodel/position/mag_psi",
30,
"°",
"Magnetic heading of the aircraft",
0,
"_magHeading",
),
(
"sim/flightmodel/position/phi",
30,
"°",
"Roll of the aircraft",
0,
"_rollAngle",
),
(
"sim/flightmodel/position/theta",
30,
"°",
"Pitch of the aircraft",
0,
"_pitchAngle",
),
]
numvalues = 1 #test with rollAngle #int(len(values) / lenvalue)
idx = 0
value = 0
for i in range(0, numvalues):
#singledata = data[(5 + lenvalue * i) : (5 + lenvalue * (i + 1))]
#(idx, value) = struct.unpack("<if", singledata)
idx = 1
value = '10.0' #data_dict["'roll'"]
retvalues[idx] = (
value,
datarefs[idx][1],
datarefs[idx][0],
datarefs[idx][5],
)
self.drefUpdate.emit(retvalues)
break
#if "BECN" in str(datagram.data())[2:6]:
# self.xpInstance.emit(
# datagram.senderAddress(),
# int.from_bytes(bytes(datagram.data())[19:21], byteorder="little"),
# )
# self.udpSock.leaveMulticastGroup(self.XPAddr)
#self.udpSock.close()