根据mqtt响应运行python代码

Question

我有一个python代码，可以使用ANT+协议设置自行车的功率。我已使用mqtt将此python代码连接到mqtt服务器。如果从mqtt server接收到的功率与以前的功率不同，我将发送自行车的功率（int），以设置新的功率（来自python代码）

这是我的密码。我不确定是否正确使用on_message功能。实际上，在代码中我应该给出参数，但我不知道应该给出哪个参数

if在process函数中，我设置自行车功率的process函数中。因此，如果从MQTT接收的消息与前面的消息不同（power，因此前面的消息不同），我希望运行process函数。因此，每当我更改MQTT server内的电源时，python代码都会向自行车发送一条消息来设置电源

import sys
import time
from ant.core import driver, node, event, message, log
from ant.core.constants import CHANNEL_TYPE_TWOWAY_TRANSMIT, TIMEOUT_NEVER
from ant.core.constants import CHANNEL_TYPE_TWOWAY_RECEIVE, TIMEOUT_NEVER

import paho.mqtt.client as mqtt #import the client1
import paho.mqtt.subscribe as subscribe

broker_address="127.0.0.1"
topic = 'NameMeasurement'
client = mqtt.Client("P1") #create new instance
client.connect(broker_address) #connect to broker

power = 0x01

def on_message(self, client, userdata, message):
    #print("message received " ,str(message.payload.decode("utf-8")))
    return int(str(message.payload.decode("utf-8")))


class MONARK(event.EventCallback):

    def __init__(self, serial, netkey):
        print('init')
        self.serial = serial
        self.netkey = netkey
        self.eventCount = 0
        self.eventTime = 0
        self.cumulativePower = 0
        self.instantaneousPower = 0
        self.antnode = None
        self.channel = None

    def start(self):
        print('start')
        #print("starting node")
        self._start_antnode()
        self._setup_channel()
        self.channel.registerCallback(self)
        #print("start listening for hr events")

    def stop(self):
        if self.channel:
            self.channel.close()
            self.channel.unassign()
        if self.antnode:
            self.antnode.stop()

    def __enter__(self):
        return self

    def __exit__(self, type_, value, traceback):
        self.stop()

    def _start_antnode(self):
        print('start antnode')
        stick = driver.USB2Driver(self.serial)
        self.antnode = node.Node(stick)
        self.antnode.start()

    def _setup_channel(self):
        print('setup channel')
        key = node.NetworkKey('N:ANT+', self.netkey)
        self.antnode.setNetworkKey(0, key)
        self.channel = self.antnode.getFreeChannel()
        self.channel.name = 'C:HRM'
        
        self.channel.assign('N:ANT+', CHANNEL_TYPE_TWOWAY_RECEIVE)
            
        self.channel.setID(17, 0, 0)
        self.channel.setSearchTimeout(TIMEOUT_NEVER)
        self.channel.setPeriod(8182)
        self.channel.setFrequency(57)
        self.channel.open()
        self.process()
        print('open')

    def process(self):

        #if isinstance(msg, message.ChannelBroadcastDataMessage):
        #print('test')

        client.on_message=on_message #attach function to callback
        client.loop_start() #start the loop
        print("Subscribing to topic")
        client.subscribe(topic)
        if (on_message() != power):
            power = on_message()

        time.sleep(4) # wait
        client.loop_stop() #stop the loop

        self.eventCount = self.eventCount + 1
        print(self.eventCount)

        payload = chr(0x31)  # standard power-only message
        payload += chr(0xFF)
        payload += chr(0xFF)  # Pedal power not used
        payload += chr(0xFF)  # Cadence not used
        payload += chr(0xFF)
        payload += chr(0xFF)
        payload += chr(0xFF)
        payload += chr(power)
    
        ant_msg = message.ChannelBroadcastDataMessage(self.channel.number, data=payload)
        sys.stdout.write('+')
        sys.stdout.flush()
        print 'Write message to ANT stick on channel ' + repr(self.channel.number)
        self.antnode.driver.write(ant_msg.encode())
        self.process()

        msg = subscribe.simple("bike", hostname=broker_address)
        print("%s %s" % (msg.topic, msg.payload))

            
SERIAL = '/dev/ttyUSB0'
NETKEY = 'B9A521FBBD72C345'.decode('hex')


with MONARK(serial=SERIAL, netkey=NETKEY) as monark:
    monark.start()
    while True:
        try:
            time.sleep(1)
            input("Press enter to Stop\n")
        except KeyboardInterrupt:
            sys.exit(0) 

Answer 1

我认为你的模型在这里是错误的

首先，on_message()函数不返回任何内容。return语句不会达到您在这里期望的效果

其次，您不能从自己的代码中调用on_message()函数，它是在传递消息时由MQTT客户机在客户机的网络线程上调用的

你有两个选择如何使这项工作

1. 连接到代理，启动网络循环并订阅__init__()函数中的主题。然后让on_message()回调函数在MONARK类上设置一个变量。然后，您可以在每次调用process()时访问此变量

2. 设置MQTT客户机与上一个选项中的设置相同，但不是在类上设置变量，on_message()函数可以在每次新消息到达时调用process()，并将新值作为参数传递给process()