openmv/scripts/libraries/lora.py
2024-09-30 12:56:20 +02:00

354 lines
10 KiB
Python

# This file is part of the OpenMV project.
#
# Copyright (c) 2013-2021 Ibrahim Abdelkader <iabdalkader@openmv.io>
# Copyright (c) 2013-2021 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 time 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=2, 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() is 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 is not 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