Server-py/3rdparty/xclient/xmqtt.py

"""
cd /home/server/repositories/repositories/gitee.com/casperz.py-project/module-py/xclient

pip3 install paho-mqtt==1.6.1

python3 xmqtt-a1.py
python3 xmqtt-a2.py
python3 xmqtt.py


pip3 install paho-mqtt==1.6.1 -i hhttps://mirror.baidu.com/pypi/simple


"""
import paho.mqtt.client as mqtt
import time
import json


class Client(mqtt.Client):

    def __init__(self, host='127.0.0.1', port=41883):
        super().__init__()
        self.connect(host=host, port=port, keepalive=60)
        # if subscribe_topic:
        #     def on_message(client, userdata, message):
        #         # print(f"#userdata: {userdata}")
        #         # print(f"#message.payload: {str(message.payload.decode())}")
        #         print(f"#message.payload: {json.loads(message.payload)}")
        #         print(f"#message.topic: {message.topic}")
        #
        #     self.on_message = on_message
        #     self.subscribe(subscribe_topic)
        #     self.loop_forever()

    def start_subscribe_loop(self, decorate_method, subscribe_topic):
        """启动订阅循环"""
        self.on_message = decorate_method
        self.subscribe(subscribe_topic)
        self.loop_forever()

    def publish_message(self, publish_topic, message):
        """发布消息"""
        try:
            results = self.publish(publish_topic, message)
            result_code, message_id = results
            """
            result_code: 错误码
            result_code.0: 成功
            result_code.1: 失败
            result_code.4: 无法连接到 MQTT 代理
            result_code.7: QoS错误
            """
            print(f"#result_code: {result_code}, #message_id: {message_id}")
        except Exception as e:
            print(f'#Exception: {e.__class__.__name__}')


if __name__ == '__main__':
    # --- init ---
    # c1 = Client(host='127.0.0.1', port=41883)
    c1 = Client(host='192.168.131.23', port=41883)

    # --- test subscribe ---
    # def m1(_, __, p3):
    #     print(f"#message.payload: {json.loads(p3.payload)}")
    # c1.start_subscribe_loop(decorate_method=m1, subscribe_topic='test/topic')

    # --- test publish ---
    # while True:
    #     data = {
    #         'code': 1001,
    #         '方向': 6000,
    #         '刹车': 6000,
    #         '油门': 8000,
    #     }
    #     # c1.publish_message('bg/log', json.dumps(data))
    #     c1.publish_message('hs/vehicle/state', json.dumps(data))
    #     time.sleep(3)

    # --- test publish ---
    while True:
        data = {
            'address': "192.168.131.180",
            'state': 2,
            'direction': 22,
        }
        c1.publish_message('hs/vehicle/state', json.dumps(data))
        time.sleep(3)