mirror of
https://github.com/openmv/openmv.git
synced 2025-11-04 14:49:50 +08:00
Add LoRa library for Arduino Portenta.
This commit is contained in:
parent
d694a95c54
commit
2d913dc456
45
scripts/examples/Arduino/Portenta H7/39-LoRa/lora-example.py
Normal file
45
scripts/examples/Arduino/Portenta H7/39-LoRa/lora-example.py
Normal file
@ -0,0 +1,45 @@
|
||||
from lora import *
|
||||
|
||||
lora = Lora(band=BAND_EU868, poll_ms=60000, debug=False)
|
||||
|
||||
print("Firmware:", lora.get_fw_version())
|
||||
print("Device EUI:", lora.get_device_eui())
|
||||
print("Data Rate:", lora.get_datarate())
|
||||
print("Join Status:", lora.get_join_status())
|
||||
|
||||
# Example keys for connecting to the backend
|
||||
appEui = "1234567890123456"
|
||||
appKey = "12345678901234567890123456789012"
|
||||
|
||||
try:
|
||||
lora.join_OTAA(appEui, appKey, timeout=20000)
|
||||
# Or ABP:
|
||||
#lora.join_ABP(devAddr, nwkSKey, appSKey, timeout=5000)
|
||||
# You can catch individual errors like timeout, rx etc...
|
||||
except LoraErrorTimeout as e:
|
||||
print("Something went wrong; are you indoor? Move near a window and retry")
|
||||
print("ErrorTimeout:", e)
|
||||
except LoraErrorParam as e:
|
||||
print("ErrorParam:", e)
|
||||
|
||||
print("Connected.")
|
||||
lora.set_port(3)
|
||||
|
||||
try:
|
||||
if lora.send_data("HeLoRA world!", True):
|
||||
print("Message confirmed.")
|
||||
else:
|
||||
print("Message wasn't confirmed")
|
||||
|
||||
except LoraErrorTimeout as e:
|
||||
print("ErrorTimeout:", e)
|
||||
|
||||
# Read downlink messages
|
||||
while (True):
|
||||
if (lora.available()):
|
||||
data = lora.receive_data()
|
||||
if data:
|
||||
print("Port: " + data["port"])
|
||||
print("Data: " + data["data"])
|
||||
lora.poll()
|
||||
sleep_ms(1000)
|
||||
326
scripts/libraries/lora.py
Normal file
326
scripts/libraries/lora.py
Normal file
@ -0,0 +1,326 @@
|
||||
# This file is part of the OpenMV project.
|
||||
#
|
||||
# Copyright (c) 2013-2019 Ibrahim Abdelkader <iabdalkader@openmv.io>
|
||||
# Copyright (c) 2013-2019 Kwabena W. Agyeman <kwagyeman@openmv.io>
|
||||
# Copyright (c) 2021 Arduino SA
|
||||
#
|
||||
# Authors:
|
||||
# Ibrahim Abdelkader <iabdalkader@openmv.io>
|
||||
# Sebastian Romero <s.romero@arduino.cc>
|
||||
#
|
||||
# This work is licensed under the MIT license, see the file LICENSE for details.
|
||||
#
|
||||
# LoRa library for Arduino Portenta.
|
||||
|
||||
from utime import sleep_ms, ticks_ms
|
||||
from pyb import UART, Pin
|
||||
|
||||
MODE_ABP = 0
|
||||
MODE_OTAA = 1
|
||||
|
||||
RF_MODE_RFO = 0
|
||||
RF_MODE_PABOOST = 1
|
||||
|
||||
BAND_AS923 = 0
|
||||
BAND_AU915 = 1
|
||||
BAND_EU868 = 5
|
||||
BAND_KR920 = 6
|
||||
BAND_IN865 = 7
|
||||
BAND_US915 = 8
|
||||
BAND_US915_HYBRID = 9
|
||||
|
||||
CLASS_A = 'A'
|
||||
CLASS_B = 'B'
|
||||
CLASS_C = 'C'
|
||||
|
||||
class LoraError(Exception): pass
|
||||
class LoraErrorTimeout(LoraError): pass
|
||||
class LoraErrorParam(LoraError): pass
|
||||
class LoraErrorBusy(LoraError): pass
|
||||
class LoraErrorOverflow(LoraError): pass
|
||||
class LoraErrorNoNetwork(LoraError): pass
|
||||
class LoraErrorRX(LoraError): pass
|
||||
class LoraErrorUnknown(LoraError): pass
|
||||
|
||||
class Lora():
|
||||
LoraErrors = {
|
||||
"": LoraErrorTimeout, # empty buffer
|
||||
"+ERR": LoraError,
|
||||
"+ERR_PARAM": LoraErrorParam,
|
||||
"+ERR_BUSY": LoraErrorBusy,
|
||||
"+ERR_PARAM_OVERFLOW": LoraErrorOverflow,
|
||||
"+ERR_NO_NETWORK": LoraErrorNoNetwork,
|
||||
"+ERR_RX": LoraErrorRX,
|
||||
"+ERR_UNKNOWN": LoraErrorUnknown
|
||||
}
|
||||
|
||||
def __init__(self, uart=None, rst_pin=None, boot_pin=None, band=BAND_EU868, poll_ms=300000, debug=False):
|
||||
self.debug = debug
|
||||
self.uart = uart
|
||||
self.rst_pin = rst_pin
|
||||
self.boot_pin = boot_pin
|
||||
self.band = band
|
||||
self.poll_ms = poll_ms
|
||||
self.last_poll_ms = ticks_ms()
|
||||
|
||||
self.init_modem()
|
||||
|
||||
# Reset module
|
||||
self.boot_pin.value(0)
|
||||
self.rst_pin.value(1)
|
||||
sleep_ms(200)
|
||||
self.rst_pin.value(0)
|
||||
sleep_ms(200)
|
||||
self.rst_pin.value(1)
|
||||
|
||||
# Restart module
|
||||
self.restart()
|
||||
|
||||
def init_modem(self):
|
||||
# Portenta vision shield configuration
|
||||
if not self.rst_pin:
|
||||
self.rst_pin = Pin('PC6', Pin.OUT_PP, Pin.PULL_UP, value=1)
|
||||
if not self.boot_pin:
|
||||
self.boot_pin = Pin('PG7', Pin.OUT_PP, Pin.PULL_DOWN, value=0)
|
||||
if not self.uart:
|
||||
self.uart = UART(8, 19200)
|
||||
#self.uart = UART(1, 19200) # Use external module
|
||||
self.uart.init(19200, bits=8, parity=None, stop=1, timeout=250, timeout_char=100)
|
||||
|
||||
|
||||
def debug_print(self, data):
|
||||
if self.debug:
|
||||
print(data)
|
||||
|
||||
def is_arduino_firmware(self):
|
||||
return 'ARD-078' in self.fw_version
|
||||
|
||||
def configure_class(self, _class):
|
||||
self.send_command("+CLASS=", _class)
|
||||
|
||||
def configure_band(self, band):
|
||||
self.send_command("+BAND=", band)
|
||||
if (band == BAND_EU868 and self.is_arduino_firmware()):
|
||||
self.send_command("+DUTYCYCLE=", 1)
|
||||
return True
|
||||
|
||||
def set_baudrate(self, baudrate):
|
||||
self.send_command("+UART=", baudrate)
|
||||
|
||||
def set_autobaud(self, timeout=10000):
|
||||
start = ticks_ms()
|
||||
while ((ticks_ms() - start) < timeout):
|
||||
if (self.send_command('', timeout=200, raise_error=False) == '+OK'):
|
||||
sleep_ms(200)
|
||||
while (self.uart.any()):
|
||||
self.uart.readchar()
|
||||
return True
|
||||
return False
|
||||
|
||||
def get_fw_version(self):
|
||||
dev = self.send_command("+DEV?")
|
||||
fw_ver = self.send_command("+VER?")
|
||||
return dev + " " + fw_ver
|
||||
|
||||
def get_device_eui(self):
|
||||
return self.send_command("+DEVEUI?")
|
||||
|
||||
def factory_default(self):
|
||||
self.send_command("+FACNEW")
|
||||
|
||||
def restart(self):
|
||||
if (self.set_autobaud() == False):
|
||||
raise(LoraError("Failed to set autobaud"))
|
||||
|
||||
# Different delimiter as REBOOT response EVENT doesn't end with '\r'.
|
||||
if (self.send_command("+REBOOT", delimiter="+EVENT=0,0", timeout=10000) != "+EVENT=0,0"):
|
||||
raise(LoraError("Failed to reboot module"))
|
||||
sleep_ms(1000)
|
||||
self.fw_version = self.get_fw_version()
|
||||
self.configure_band(self.band)
|
||||
|
||||
def set_rf_power(self, mode, power):
|
||||
self.send_command("+RFPOWER=", mode, ",", power)
|
||||
|
||||
def set_port(self, port):
|
||||
self.send_command("+PORT=", port)
|
||||
|
||||
def set_public_network(self, enable):
|
||||
self.send_command("+NWK=", int(enable))
|
||||
|
||||
def sleep(self, enable):
|
||||
self.send_command("+SLEEP=", int(enable))
|
||||
|
||||
def format(self, hexMode):
|
||||
self.send_command("+DFORMAT=", int(hexMode))
|
||||
|
||||
def set_datarate(self, dr):
|
||||
self.send_command("+DR=", dr)
|
||||
|
||||
def get_datarate(self):
|
||||
return int(self.send_command("+DR?"))
|
||||
|
||||
def set_adr(self, adr):
|
||||
self.send_command("+ADR=", int(adr))
|
||||
|
||||
def get_adr(self):
|
||||
return int(self.send_command("+ADR?"))
|
||||
|
||||
def get_devaddr(self):
|
||||
return self.send_command("+DEVADDR?")
|
||||
|
||||
def get_nwk_skey(self):
|
||||
return self.send_command("+NWKSKEY?")
|
||||
|
||||
def get_appskey(self):
|
||||
return self.send_command("+APPSKEY?")
|
||||
|
||||
def get_rx2dr(self):
|
||||
return int(self.send_command("+RX2DR?"))
|
||||
|
||||
def set_rx2dr(self, dr):
|
||||
self.send_command("+RX2DR=", dr)
|
||||
|
||||
def get_ex2freq(self):
|
||||
return int(self.send_command("+RX2FQ?"))
|
||||
|
||||
def set_rx2freq(self, freq):
|
||||
self.send_command("+RX2FQ=", freq)
|
||||
|
||||
def set_fcu(self, fcu):
|
||||
self.send_command("+FCU=", fcu)
|
||||
|
||||
def get_fcu(self):
|
||||
return int(self.send_command("+FCU?"))
|
||||
|
||||
def set_fcd(self, fcd):
|
||||
self.send_command("+FCD=", fcd)
|
||||
|
||||
def get_fcd(self):
|
||||
return int(self.send_command("+FCD?"))
|
||||
|
||||
def change_mode(self, mode):
|
||||
self.send_command("+MODE=", mode)
|
||||
|
||||
def join(self, timeout_ms):
|
||||
if self.send_command("+JOIN", timeout=timeout_ms) != "+ACK":
|
||||
return False
|
||||
response = self.receive('\r', timeout=timeout_ms)
|
||||
return response == "+EVENT=1,1"
|
||||
|
||||
def get_join_status(self):
|
||||
return int(self.send_command("+NJS?")) == 1
|
||||
|
||||
def get_max_size(self):
|
||||
if (self.is_arduino_firmware()):
|
||||
return 64
|
||||
return int(self.send_command("+MSIZE?", timeout=2000))
|
||||
|
||||
def poll(self):
|
||||
if ((ticks_ms() - self.last_poll_ms) > self.poll_ms):
|
||||
self.last_poll_ms = ticks_ms()
|
||||
# Triggers a fake write
|
||||
self.send_data('\0', True)
|
||||
|
||||
def send_data(self, buff, confirmed=True):
|
||||
max_len = self.get_max_size()
|
||||
if (len(buff) > max_len):
|
||||
raise(LoraError("Packet exceeds max length"))
|
||||
if self.send_command("+CTX " if confirmed else "+UTX ", len(buff), data=buff) != "+OK":
|
||||
return False
|
||||
if confirmed:
|
||||
response = self.receive('\r', timeout=10000)
|
||||
return response == "+ACK"
|
||||
return True
|
||||
|
||||
def receive_data(self, timeout=1000):
|
||||
response = self.receive('\r', timeout=timeout)
|
||||
if response.startswith("+RECV"):
|
||||
params = response.split("=")[1].split(",")
|
||||
port = params[0]
|
||||
length = int(params[1])
|
||||
dummy_data_length = 2 # Data starts with \n\n sequence
|
||||
data = self.receive(max_bytes=length+dummy_data_length, timeout=timeout)[dummy_data_length:]
|
||||
return {'port' : port, 'data' : data}
|
||||
|
||||
def receive(self, delimiter=None, max_bytes=None, timeout=1000):
|
||||
buf = []
|
||||
start = ticks_ms()
|
||||
while ((ticks_ms() - start) < timeout):
|
||||
while (self.uart.any()):
|
||||
buf += chr(self.uart.readchar())
|
||||
|
||||
if max_bytes and len(buf) == max_bytes:
|
||||
data = ''.join(buf)
|
||||
self.debug_print(data)
|
||||
return data
|
||||
if (len(buf) and delimiter != None):
|
||||
data = ''.join(buf)
|
||||
trimmed = data[0:-1] if data[-1] == '\r' else data
|
||||
|
||||
if (isinstance(delimiter, str) and len(delimiter) == 1 and buf[-1] == delimiter):
|
||||
self.debug_print(trimmed)
|
||||
return trimmed
|
||||
if (isinstance(delimiter, str) and trimmed == delimiter):
|
||||
self.debug_print(trimmed)
|
||||
return trimmed
|
||||
if (isinstance(delimiter, list) and trimmed in delimiter):
|
||||
self.debug_print(trimmed)
|
||||
return trimmed
|
||||
|
||||
data = ''.join(buf)
|
||||
self.debug_print(data)
|
||||
return data[0:-1] if len(data) != 0 and data[-1] == '\r' else data
|
||||
|
||||
def available(self):
|
||||
return self.uart.any()
|
||||
|
||||
def join_OTAA(self, appEui, appKey, devEui=None, timeout=60000):
|
||||
self.change_mode(MODE_OTAA)
|
||||
self.send_command("+APPEUI=", appEui)
|
||||
self.send_command("+APPKEY=", appKey)
|
||||
if (devEui):
|
||||
self.send_command("+DEVEUI=", devEui)
|
||||
network_joined = self.join(timeout)
|
||||
# This delay was in MKRWAN.h
|
||||
#delay(1000);
|
||||
return network_joined
|
||||
|
||||
def join_ABP(self, nwkId, devAddr, nwkSKey, appSKey, timeout=60000):
|
||||
self.change_mode(MODE_ABP)
|
||||
# Commented in MKRWAN.h
|
||||
#self.send_command("+IDNWK=", nwkId)
|
||||
self.send_command("+DEVADDR=", devAddr)
|
||||
self.send_command("+NWKSKEY=", nwkSKey)
|
||||
self.send_command("+APPSKEY=", appSKey)
|
||||
self.join(timeout)
|
||||
return self.get_join_status()
|
||||
|
||||
def handle_error(self, command, data):
|
||||
if not data.startswith("+ERR") and data != "":
|
||||
return
|
||||
if (data in self.LoraErrors):
|
||||
raise(self.LoraErrors[data]('Command "%s" has failed!'%command))
|
||||
raise(LoraError('Command: "%s" failed with unknown status: "%s"'%(command, data)))
|
||||
|
||||
def send_command(self, cmd, *args, delimiter='\r', data=None, timeout=1000, raise_error=True):
|
||||
# Write command and args
|
||||
uart_cmd = 'AT'+cmd+''.join([str(x) for x in args])+'\r'
|
||||
self.debug_print(uart_cmd)
|
||||
self.uart.write(uart_cmd)
|
||||
|
||||
# Attach raw data
|
||||
if data:
|
||||
self.debug_print(data)
|
||||
self.uart.write(data)
|
||||
|
||||
# Read status and value (if any)
|
||||
response = self.receive(delimiter, timeout=timeout)
|
||||
|
||||
# Error handling
|
||||
if raise_error:
|
||||
self.handle_error(cmd, response)
|
||||
|
||||
if cmd.endswith('?'):
|
||||
return response.split('=')[1]
|
||||
return response
|
||||
@ -9,3 +9,4 @@ freeze ("$(MPY_LIB_DIR)/", "rtsp.py")
|
||||
freeze ("$(MPY_LIB_DIR)/", "ssd1306.py")
|
||||
freeze ("$(MPY_LIB_DIR)/", "tb6612.py")
|
||||
freeze ("$(MPY_LIB_DIR)/", "vl53l1x.py")
|
||||
freeze ("$(MPY_LIB_DIR)/", "lora.py")
|
||||
|
||||
Loading…
Reference in New Issue
Block a user