MQTT用戶端入門 - 三、Python 用戶端程式設計

import paho.mqtt.publish as publish

# publish a message then disconnect.
host = "localhost"
topic = "tw/rocksaying"
payload = "hello mqtt"

# If broker asks user/password.
auth = {'username': "", 'password': ""}

# If broker asks client ID.
client_id = ""

publish.single(topic, payload, qos=1, hostname=host)

#publish.single(topic, payload, qos=1, host=host,
#    auth=auth, client_id=client_id)

# coding: utf-8
import sys, os, time
reload(sys)
sys.setdefaultencoding('utf-8')

import paho.mqtt.client as mqtt

# If broker asks client ID.
client_id = ""

client = mqtt.Client(client_id=client_id)

# If broker asks user/password.
user = ""
password = ""
client.username_pw_set(user, password)

client.connect("localhost")

topic = "tw/rocksaying"
payload = "你好 mqtt"

for i in xrange(10):
    client.publish(topic, "%s - %d" % (payload, i))
    time.sleep(0.01)
    # 當 qos = 0, 若訊息間隔太短，就可能會漏發訊息。這是正常現象。

# coding: utf-8
import sys, os, time, signal
reload(sys)
sys.setdefaultencoding('utf-8')
import paho.mqtt.client as mqtt

client = None
mqtt_looping = False

TOPIC_ROOT = "tw/rocksaying"

def on_connect(mq, userdata, rc, _):
    # subscribe when connected.
    mq.subscribe(TOPIC_ROOT + '/#')

def on_message(mq, userdata, msg):
    print "topic: %s" % msg.topic
    print "payload: %s" % msg.payload
    print "qos: %d" % msg.qos

def mqtt_client_thread():
    global client, mqtt_looping
    client_id = "" # If broker asks client ID.
    client = mqtt.Client(client_id=client_id)

    # If broker asks user/password.
    user = ""
    password = ""
    client.username_pw_set(user, password)

    client.on_connect = on_connect
    client.on_message = on_message

    try:
        client.connect("localhost")
    except:
        print "MQTT Broker is not online. Connect later."

    mqtt_looping = True
    print "Looping..."

    #mqtt_loop.loop_forever()
    cnt = 0
    while mqtt_looping:
        client.loop()

        cnt += 1
        if cnt > 20:
            try:
                client.reconnect() # to avoid 'Broken pipe' error.
            except:
                time.sleep(1)
            cnt = 0

    print "quit mqtt thread"
    client.disconnect()

def stop_all(*args):
    global mqtt_looping
    mqtt_looping = False

if __name__ == '__main__':
    signal.signal(signal.SIGTERM, stop_all)
    signal.signal(signal.SIGQUIT, stop_all)
    signal.signal(signal.SIGINT,  stop_all)  # Ctrl-C

    mqtt_client_thread()

    print "exit program"
    sys.exit(0)

mqttc.loop_start()  # enter a looping thread.

# main thread
while True:
    temperature = sensor.blocking_read()
    mqttc.publish("paho/temperature", temperature)

# coding: utf-8
import sys, os, time, signal
reload(sys)
sys.setdefaultencoding('utf-8')
import threading
import paho.mqtt.client as mqtt

client = None
mqtt_looping = False
reading_thread = None

def status_reading():
    while True:
        #status = gate.read()
        time.sleep(1)   # 假設讀取狀態
        status = "noop"

        if mqtt_looping:
            client.publish(TOPIC_ROOT + '/status', status, qos=1)
        else:
            print "quit status reading thread"
            return

def on_message(mq, userdata, msg):
    print "topic: %s" % msg.topic
    print "payload: %s" % msg.payload
    print "qos: %d" % msg.qos

    chn = msg.topic.rpartition('/')[-1]
    if chn == 'command':
        print "Should invoke gate %s function" % msg.payload

        # 假裝開門狀態
        client.publish(TOPIC_ROOT + '/status', "opening", qos=1)
        time.sleep(0.5)
        client.publish(TOPIC_ROOT + '/status', "opened", qos=1)

def mqtt_client_thread():
    #
    # 省略前後文
    #
    print "Looping..."

    reading_thread = threading.Thread(target=status_reading)
    reading_thread.start()

    #mqtt_loop.loop_forever()
    cnt = 0
    while mqtt_looping:


if __name__ == '__main__':
    #
    # 省略前後文
    #
    mqtt_client_thread()

    reading_thread.join()
    print "exit program"
    sys.exit(0)

#
# 省略前後文
#
import paho.mqtt.client as mqtt
import gobject, dbus, dbus.service, dbus.mainloop.glib

reading_thread = None
dbus_service = None
dbus_loop = None

def status_reading():
    while True:
        #status = gate.read()
        time.sleep(1)
        status = "noop"

        if mqtt_looping:
            client.publish(TOPIC_ROOT + '/status', status, qos=1)
            if dbus_service:
                dbus_service.StatusChanged(status)
        else:
            print "quit status reading thread"
            return

def mqtt_client_thread():
    #
    # 省略前後文
    #
    while mqtt_looping:
        client.loop()

        cnt += 1
        if cnt > 20:
            try:
                client.reconnect() # to avoid 'Broken pipe' error.
            except:
                time.sleep(1)
            cnt = 0

    print "quit mqtt thread"
    dbus_loop.quit()
    client.disconnect()

class DbusService(dbus.service.Object):
    NAME = "tw.rocksaying.Gate"
    PATH = '/'+NAME.replace('.','/')

    def __init__(self, event_loop):
        self.event_loop = event_loop
        self.bus = dbus.SystemBus()
        bus_name = dbus.service.BusName(DbusService.NAME, bus=self.bus)
        dbus.service.Object.__init__(self, bus_name, DbusService.PATH)

    @dbus.service.signal('tw.rocksaying.Gate', signature='s')
    def StatusChanged(self, msg):
        pass

    @dbus.service.method('tw.rocksaying.Gate')
    def Open(self):
        print "Invoke gate open function"

def dbus_service_thread():
    try:
        print "dbus looping"
        dbus_loop.run()
    except:
        pass
    print "quit dbus thread"

if __name__ == '__main__':
    signal.signal(signal.SIGTERM, stop_all)
    signal.signal(signal.SIGQUIT, stop_all)
    signal.signal(signal.SIGINT,  stop_all)  # Ctrl-C

    gobject.threads_init()
    dbus.mainloop.glib.threads_init()
    dbus.mainloop.glib.DBusGMainLoop(set_as_default=True)
    dbus_loop = gobject.MainLoop()
    dbus_service = DbusService(dbus_loop)

    dbus_thread = threading.Thread(target=dbus_service_thread)
    dbus_thread.start()

    mqtt_client_thread()

    dbus_thread.join()
    reading_thread.join()
    print "exit program"
    sys.exit(0)