Lora32 et Python – Premiers pas

from machine import Pin
import _thread
import time

print("Hello World")

def testThread():
    pin_led = Pin(2, Pin.OUT)
    while True:
        pin_led.value(not pin_led.value())
        time.sleep(0.5)

_thread.start_new_thread(testThread, ())

from uPySensors.ssd1306_i2c import Display
import _thread
import time

print("Hello World")

display = Display()
display.show_text_wrap("Hi Moon")

def testThread():
    n = 0
    while True:
        display.show_text_wrap("Hi Moon " + str(n))
        n = n + 1
        time.sleep(0.5)

_thread.start_new_thread(testThread, ())

import config_lora
import _thread
from sx127x import SX127x
from time import sleep
from uPySensors.ssd1306_i2c import Display
from controller_esp32 import ESP32Controller

print("Hello World")

# Variables Lora
controller = ESP32Controller()
lora = controller.add_transceiver(SX127x(name = 'LoRa'),
                                  pin_id_ss = ESP32Controller.PIN_ID_FOR_LORA_SS,
                                  pin_id_RxDone = ESP32Controller.PIN_ID_FOR_LORA_DIO0)

# Fonction Lora
def sendThread():
    counter = 0
    print("LoRa Sender")
    display = Display()

    while True:
        payload = 'Hello ({0})'.format(counter)
        print("Sending packet: \n{}\n".format(payload))
        display.show_text_wrap("{0} RSSI: {1}".format(payload, lora.packetRssi()))
        lora.println(payload)

        counter += 1
        sleep(5)

# Lancement Lora
_thread.start_new_thread(sendThread, ())

import config_lora
import _thread
from sx127x import SX127x
from uPySensors.ssd1306_i2c import Display
from controller_esp32 import ESP32Controller

print("Hello World")

# Variables Lora
controller = ESP32Controller()
lora = controller.add_transceiver(SX127x(name = 'LoRa'),
                                  pin_id_ss = ESP32Controller.PIN_ID_FOR_LORA_SS,
                                  pin_id_RxDone = ESP32Controller.PIN_ID_FOR_LORA_DIO0)

# Fonction Lora
def receiveThread():
    display = Display()
    print("LoRa Receiver")
    display.show_text_wrap("LoRa Receiver")

    while True:
        if lora.receivedPacket():
            lora.blink_led()

            try:
                payload = lora.read_payload()
                display.show_text_wrap("{0} RSSI: {1}".format(payload.decode(), lora.packetRssi()))
                print("*** Received message ***\n{}".format(payload.decode()))

            except Exception as e:
                print(e)
                display.show_text("RSSI: {}\n".format(lora.packetRssi()), 10, 10)
                print("with RSSI: {}\n".format(lora.packetRssi))

# Lancement Lora
_thread.start_new_thread(receiveThread, ())

import network

print("Hello World")

wlan = network.WLAN(network.STA_IF)
wlan.active(True)
print(wlan.scan())

# Modifier SSID et mdp
wlan.connect(b'ssid', b'mdp') 

from umqttsimple import MQTTClient
import machine
import ubinascii
import network
import _thread
import time

print("Hello World")

# Connexion WiFi
print("Connexion WiFi")

wlan = network.WLAN(network.STA_IF)
wlan.active(True)
print(wlan.scan())
wlan.connect(b'ssid', b'mdp')

# Attente connexion WiFi
print("Attente 10s")
time.sleep(10)

# Parametres connexion MQTT
mqtt_server = b'ip_serveur'
topic_sub = b'notification'
topic_pub = b'hello'

# Fonctions connexion MQTT
client_id = ubinascii.hexlify(machine.unique_id())

def sub_cb(topic, msg):
    print((topic, msg))

def connect_and_subscribe():
    client = MQTTClient(client_id, mqtt_server)
    client.set_callback(sub_cb)
    client.connect()
    client.subscribe(topic_sub)
    print('Connected to %s MQTT broker, subscribed to %s topic' % (mqtt_server, topic_sub))
    return client

# Connexion MQTT
print("Connexion MQTT")
client = connect_and_subscribe()

def mqttThread():
    message_interval = 5
    last_message = 0
    counter = 0
    while True:
        client.check_msg()
        if (time.time() - last_message) > message_interval:
            msg = b'Hello #%d' % counter
            client.publish(topic_pub, msg)
            last_message = time.time()
            counter += 1

_thread.start_new_thread(mqttThread, ())