92 lines
3.6 KiB
Python
92 lines
3.6 KiB
Python
from uPyLora import config_lora
|
|
from uPyLora.sx127x import SX127x
|
|
from uPyLora.controller_esp32 import ESP32Controller
|
|
|
|
class LoRaTransceiver:
|
|
def __init__(self, spreadingfactor=7, bandwidth=125E3, frequency=868E6, syncword=0x12, display=None):
|
|
""" set up radio module
|
|
"""
|
|
self.controller = ESP32Controller()
|
|
self.lora = self.controller.add_transceiver(SX127x(name='LoRa'), pin_id_ss=ESP32Controller.PIN_ID_FOR_LORA_SS, pin_id_RxDone=ESP32Controller.PIN_ID_FOR_LORA_DIO0)
|
|
|
|
self.setSpreadingFactor(spreadingfactor)
|
|
self.setSignalBandwidth(bandwidth)
|
|
self.setFrequency(frequency)
|
|
self.setSyncWord(syncword)
|
|
|
|
self.display = display
|
|
|
|
def setSpreadingFactor(self, spreadingfactor):
|
|
""" set spreading factor
|
|
possible values are 7-12
|
|
"""
|
|
self.spreadingfactor = spreadingfactor
|
|
self.lora.setSpreadingFactor(spreadingfactor)
|
|
|
|
def setFrequency(self, frequency):
|
|
""" set frequency (= channel used)
|
|
for now only 868 Mhz and 868.1 MHz are supported
|
|
"""
|
|
self.frequency = frequency
|
|
self.lora.setFrequency(frequency)
|
|
|
|
def setSignalBandwidth(self, bandwidth):
|
|
""" set the bandwidth
|
|
possible values should be 125 kHz or 250 kHz according to EU 868 frequency plan
|
|
"""
|
|
self.bandwidth = bandwidth
|
|
self.lora.setSignalBandwidth(bandwidth)
|
|
|
|
def setSyncWord(self, syncword):
|
|
""" set the sync word
|
|
should be 0x12 for plain LoRa and 0x34 for LoRaWAN
|
|
"""
|
|
self.syncword = syncword
|
|
self.lora.setSyncWord(syncword)
|
|
|
|
def send(self, data):
|
|
""" send raw data
|
|
"""
|
|
data_hex = " ".join(list(map(lambda x: "{0:02x}".format(x), data))).upper()
|
|
print("Sending data {}".format(data_hex))
|
|
if self.display is not None:
|
|
self.display.show_text_wrap("Sending data {}".format(data_hex))
|
|
self.lora.println_raw(data)
|
|
|
|
|
|
def send_string(self, message):
|
|
""" send a message
|
|
message should be a string
|
|
"""
|
|
print("Sending string {}".format(message))
|
|
if self.display is not None:
|
|
self.display.show_text_wrap("Sending string {}".format(message))
|
|
self.lora.println(message)
|
|
|
|
|
|
def recv(self):
|
|
""" start listening for messages and output them on the console """
|
|
print("Receiving at frequency {} with spreading factor {} ...".format(self.frequency, self.spreadingfactor))
|
|
if self.display is not None:
|
|
self.display.show_text_wrap("Receiving at frequency {} with spreading factor {} ...".format(self.frequency, self.spreadingfactor))
|
|
|
|
while True:
|
|
# check for new message
|
|
if self.lora.receivedPacket():
|
|
self.lora.blink_led()
|
|
|
|
print(" got packet: ")
|
|
try:
|
|
payload_bytes = bytes(self.lora.read_payload_raw())
|
|
payload_hex = " ".join(list(map(lambda x: "{0:02x}".format(x), payload_bytes))).upper()
|
|
payload_string = self.lora.read_payload_raw()
|
|
print(" length : {0}".format(len(payload_bytes)))
|
|
print(" hex : {0}".format(payload_hex))
|
|
print(" bytes : {0}".format(payload_bytes))
|
|
print(" RSSI : {0}\n".format(self.lora.packetRssi()))
|
|
if self.display is not None:
|
|
self.display.show_text_wrap("Received: {0} RSSI: {1}".format(payload_bytes, self.lora.packetRssi()))
|
|
|
|
except Exception as e:
|
|
print(e)
|