Raspberry Pi 使用 Grove LoRa 無線模組負擔 LoRa Gateway 角色

#!/usr/bin/python3
# -*- coding: utf-8 -*-
import sys, os, signal
# the path to pyrfm library.
#sys.path.append(os.path.join(os.path.dirname(__file__), '..'))
import pyrfm

lora_settings = {
    'id': 255, # my (gateway) lora id.
    'll':{
        'type':'rfm95'
    },
    'pl':{
        'type': 'serial_seed',
        'port': '/dev/serial0',  # default UART port in Raspberry Pi 3
    }
}

if len(sys.argv) >= 2:
    lora_settings['pl']['port'] = sys.argv[1]

service_looping = False

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

signal.signal(signal.SIGTERM, stop_all)
signal.signal(signal.SIGINT,  stop_all)  # Ctrl-C


print("LoRa Port: ", lora_settings['pl']['port'])
print("LoRa ID: ", lora_settings['id']) # 非必要

ll = pyrfm.getLL(lora_settings)
print("HW-Version: ", ll.getVersion())

if ll.setOpModeSleep(True, True):
    ll.setFiFo()
    ll.setOpModeIdle()
    ll.setModemConfig('Bw125Cr45Sf128');
    #ll.setModemConfig('Bw31_25Cr48Sf512');  # slow/long, and more than 2 seconds timeout
    ll.setPreambleLength(8)
    ll.setFrequency(434) # 頻率依你使用的 LoRa 模組規格而定
    ll.setTxPower(13) # useRFO default is false.

    service_looping = True
    while service_looping:
        if ll.waitRX(timeout=3):
            data = ll.recv()

            header = data[0:4]
            # <to> <from> <id> <flags>
            msg = array.array('B', data[4:]).tostring()
            print("read %d bytes" % len(data))
            print("header: ", header)
            print("message: ", msg)

            from_id = header[1]

            if header[0] != lora_settings['id']:
                print("recv_id is not matched")
                continue
                
            print("TODO received task")

    sys.exit(0)

#!/usr/bin/python3
.
.
import threading, queue
import paho.mqtt.publish as publish

publishing_messages = queue.Queue(10000)

def append_publishing_message(topic, id, data):
    if publishing_messages.full():
        # drop oldest message.
        publishing_messages.get()
        publishing_messages.task_done()
    publishing_messages.put((topic, id, data))

def publish_worker():
    '''
    daemon thread.
    它會從 publishing_messages 中取出一筆待發佈訊息 (如果沒有則等待)。
    接著會嘗試發佈這一筆訊息，直到成功或 publishing_messages 滿了為止。
    在這一筆訊息發佈成功前，主執行緒仍然可以持續往 publishing_messages 中增加訊息。
    若發佈失敗，且 publishing_messages 已滿，就會放棄這筆訊息，取出下一筆訊息。
    否則等待重試。
    '''
    topic = None
    data = None
    sensor_id = 0

    while True:
        if data == None or publishing_messages.full(): # 佇列已滿時，則放棄尚未發佈的舊訊息
            topic, sensor_id, data = publishing_messages.get()
            publishing_messages.task_done()
            # 待發佈訊息已由 data 保存，故現在就可釋放佇列，讓主執行緒繼續堆放訊息。

        full_topic = "%s/sensor/%s/%d" % (mqtt_settings['topic_base'], topic, sensor_id)
        # mqtt_auth = {
        #     'username': mqtt_settings['username'],
        #     'password': mqtt_settings['password']
        # }
        try:
            publish.single(full_topic, data, qos=1,
                hostname=mqtt_settings['host'], client_id=gethostname(), auth=mqtt_settings)
            print("published")
            topic = None
            data = None
            sensor_id = 0
        except Exception as e:
            logging.error("%s Retry publish later." % e)
            time.sleep(10)


publish_thread = threading.Thread(target=publish_worker, daemon=True)
publish_thread.start()

if ll.setOpModeSleep(True, True):
    .
    .

    while service_looping:
        if ll.waitRX(timeout=3):
            data = ll.recv()
                .
                .

            # print("TODO received task")
            append_publishing_message(topic, id, data)