Compare commits

..

1 commit

Author SHA1 Message Date
Dan Halbert
fbd6d1dfe2
Fix default baudrate doc; correct spelling of "MHz" 2024-11-13 10:07:23 -05:00
7 changed files with 21 additions and 254 deletions

View file

@ -8,9 +8,6 @@
# Required
version: 2
sphinx:
configuration: docs/conf.py
build:
os: ubuntu-lts-latest
tools:

View file

@ -639,11 +639,10 @@ class RFM69(RFMSPI):
# Write payload to transmit fifo
self.write_from(_RF69_REG_00_FIFO, complete_payload)
def read_fifo(self) -> Optional[bytearray]:
def read_fifo(self) -> bytearray:
"""Read the packet from the FIFO."""
# Read the length of the FIFO.
fifo_length = self.read_u8(_RF69_REG_00_FIFO)
packet = None # return None if FIFO empty
if fifo_length > 0: # read and clear the FIFO if anything in it
packet = bytearray(fifo_length)
# read the packet

View file

@ -23,7 +23,7 @@ try:
from circuitpython_typing import ReadableBuffer
try:
from typing import Literal, Optional
from typing import Literal
except ImportError:
from typing_extensions import Literal
@ -119,20 +119,20 @@ RX_MODE = 0b101
class RFM9x(RFMSPI):
"""Interface to a RFM95/6/7/8 LoRa radio module. Allows sending and
receiving bytes of data in long range LoRa mode at a support board frequency
(433/915mhz).
(433/915 MHz).
You must specify the following parameters:
- spi: The SPI bus connected to the radio.
- cs: The CS pin DigitalInOut connected to the radio.
- reset: The reset/RST pin DigialInOut connected to the radio.
- frequency: The frequency (in mhz) of the radio module (433/915mhz typically).
- frequency: The frequency (in MHz) of the radio module (433/915 MHz typically).
You can optionally specify:
- preamble_length: The length in bytes of the packet preamble (default 8).
- high_power: Boolean to indicate a high power board (RFM95, etc.). Default
is True for high power.
- baudrate: Baud rate of the SPI connection, default is 5mhz but you might
choose to lower to 1mhz if using long wires or a breadboard.
- baudrate: Baud rate of the SPI connection, default is 10 MHz but you might
choose to lower to MHz if using long wires or a breadboard.
- agc: Boolean to Enable/Disable Automatic Gain Control - Default=False (AGC off)
- crc: Boolean to Enable/Disable Cyclic Redundancy Check - Default=True (CRC Enabled)
Remember this library makes a best effort at receiving packets with pure
@ -169,8 +169,6 @@ class RFM9x(RFMSPI):
auto_agc = RFMSPI.RegisterBits(_RF95_REG_26_MODEM_CONFIG3, offset=2, bits=1)
header_mode = RFMSPI.RegisterBits(_RF95_REG_1D_MODEM_CONFIG1, offset=0, bits=1)
low_datarate_optimize = RFMSPI.RegisterBits(_RF95_REG_26_MODEM_CONFIG3, offset=3, bits=1)
lna_boost_hf = RFMSPI.RegisterBits(_RF95_REG_0C_LNA, offset=0, bits=2)
@ -198,7 +196,7 @@ class RFM9x(RFMSPI):
self.module = "RFM9X"
self.max_packet_length = 252
self.high_power = high_power
# Device support SPI mode 0 (polarity & phase = 0) up to a max of 10mhz.
# Device support SPI mode 0 (polarity & phase = 0) up to a max of 10 MHz.
# Set Default Baudrate to 5 MHz to avoid problems
# self._device = spidev.SPIDevice(spi, cs, baudrate=baudrate, polarity=0, phase=0)
# Setup reset as a digital output - initially High
@ -426,11 +424,6 @@ class RFM9x(RFMSPI):
else:
self.write_u8(0x2F, 0x44)
self.write_u8(0x30, 0)
# set low_datarate_optimize for signal duration > 16 ms
if 1000 / (self.signal_bandwidth / (1 << self.spreading_factor)) > 16:
self.low_datarate_optimize = 1
else:
self.low_datarate_optimize = 0
@property
def coding_rate(self) -> Literal[5, 6, 7, 8]:
@ -468,21 +461,14 @@ class RFM9x(RFMSPI):
if val == 6:
self.detection_optimize = 0x5
self.header_mode = 1 # enable implicit header mode
else:
self.detection_optimize = 0x3
self.header_mode = 0 # enable exlicit header mode
self.write_u8(_RF95_DETECTION_THRESHOLD, 0x0C if val == 6 else 0x0A)
self.write_u8(
_RF95_REG_1E_MODEM_CONFIG2,
((self.read_u8(_RF95_REG_1E_MODEM_CONFIG2) & 0x0F) | ((val << 4) & 0xF0)),
)
# set low_datarate_optimize for signal duration > 16 ms
if 1000 / (self.signal_bandwidth / (1 << self.spreading_factor)) > 16:
self.low_datarate_optimize = 1
else:
self.low_datarate_optimize = 0
@property
def enable_crc(self) -> bool:
@ -505,16 +491,6 @@ class RFM9x(RFMSPI):
self.read_u8(_RF95_REG_1E_MODEM_CONFIG2) & 0xFB,
)
@property
def payload_length(self) -> int:
"""Must be set when using Implicit Header Mode - required for SF = 6"""
return self.read_u8(_RF95_REG_22_PAYLOAD_LENGTH)
@payload_length.setter
def payload_length(self, val: int) -> None:
# Set payload length
self.write_u8(_RF95_REG_22_PAYLOAD_LENGTH, val)
@property
def crc_error(self) -> bool:
"""crc status"""
@ -541,11 +517,10 @@ class RFM9x(RFMSPI):
# Write payload and header length.
self.write_u8(_RF95_REG_22_PAYLOAD_LENGTH, len(payload))
def read_fifo(self) -> Optional[bytearray]:
def read_fifo(self) -> bytearray:
"""Read the data from the FIFO."""
# Read the length of the FIFO.
fifo_length = self.read_u8(_RF95_REG_13_RX_NB_BYTES)
packet = None # return None if FIFO empty
if fifo_length > 0: # read and clear the FIFO if anything in it
packet = bytearray(fifo_length)
current_addr = self.read_u8(_RF95_REG_10_FIFO_RX_CURRENT_ADDR)

View file

@ -563,11 +563,10 @@ class RFM9xFSK(RFMSPI):
# Write payload to transmit fifo
self.write_from(_RF95_REG_00_FIFO, complete_payload)
def read_fifo(self) -> Optional[bytearray]:
def read_fifo(self) -> bytearray:
"""Read the data from the FIFO."""
# Read the length of the FIFO.
fifo_length = self.read_u8(_RF95_REG_00_FIFO)
packet = None # return None if FIFO empty
if fifo_length > 0: # read and clear the FIFO if anything in it
packet = bytearray(fifo_length)
# read the packet

View file

@ -178,7 +178,7 @@ class RFMSPI:
"""
self.ack_retries = 5
"""The number of ACK retries before reporting a failure."""
self.ack_delay: float = None
self.ack_delay = None
"""The delay time before attemting to send an ACK.
If ACKs are being missed try setting this to .1 or .2.
"""
@ -286,6 +286,13 @@ class RFMSPI:
Returns: True if success or False if the send timed out.
"""
# Disable pylint warning to not use length as a check for zero.
# This is a puzzling warning as the below code is clearly the most
# efficient and proper way to ensure a precondition that the provided
# buffer be within an expected range of bounds. Disable this check.
# pylint: disable=len-as-condition
assert 0 < len(data) <= self.max_packet_length
# pylint: enable=len-as-condition
self.idle() # Stop receiving to clear FIFO and keep it clear.
# Combine header and data to form payload
if self.radiohead:
@ -311,13 +318,6 @@ class RFMSPI:
payload = destination.to_bytes(1, "big") + data
else:
payload = data
# Disable pylint warning to not use length as a check for zero.
# This is a puzzling warning as the below code is clearly the most
# efficient and proper way to ensure a precondition that the provided
# buffer be within an expected range of bounds. Disable this check.
# pylint: disable=len-as-condition
assert 0 < len(payload) <= self.max_packet_length
# pylint: enable=len-as-condition
self.fill_fifo(payload)
# Turn on transmit mode to send out the packet.
self.transmit()
@ -430,7 +430,7 @@ class RFMSPI:
self.crc_error_count += 1
else:
packet = self.read_fifo()
if (packet is not None) and self.radiohead:
if self.radiohead:
if len(packet) < 5:
# reject the packet if it is too small to contain the RAdioHead Header
packet = None
@ -503,7 +503,7 @@ class RFMSPI:
self.crc_error_count += 1
else:
packet = self.read_fifo()
if (packet is not None) and self.radiohead:
if self.radiohead:
if len(packet) < 5:
# reject the packet if it is too small to contain the RAdioHead Header
packet = None

View file

@ -1,98 +0,0 @@
# SPDX-FileCopyrightText: 2021 ladyada for Adafruit Industries
# SPDX-License-Identifier: MIT
import time
import board
import busio
import digitalio
# Define radio parameters.
RADIO_FREQ_MHZ = 915.0 # Frequency of the radio in Mhz. Must match your
# module! Can be a value like 915.0, 433.0, etc.
# Define pins connected to the chip, use these if wiring up the breakout according to the guide:
CS = digitalio.DigitalInOut(board.CE1)
RESET = digitalio.DigitalInOut(board.D25)
# Initialize SPI bus.
spi = busio.SPI(board.SCK, MOSI=board.MOSI, MISO=board.MISO)
# Initialze RFM radio
# uncommnet the desired import and rfm initialization depending on the radio boards being used
# Use rfm9x for two RFM9x radios using LoRa
from adafruit_rfm import rfm9x
rfm = rfm9x.RFM9x(spi, CS, RESET, RADIO_FREQ_MHZ)
rfm.radiohead = False # don't appent RadioHead heade
# set spreading factor
rfm.spreading_factor = 7
print("spreading factor set to :", rfm.spreading_factor)
print("low_datarate_optimize set to: ", rfm.low_datarate_optimize)
# rfm.signal_bandwidth = 500000
print("signal_bandwidth set to :", rfm.signal_bandwidth)
print("low_datarate_optimize set to: ", rfm.low_datarate_optimize)
if rfm.spreading_factor == 12:
rfm.xmit_timeout = 5
print("xmit_timeout set to: ", rfm.xmit_timeout)
if rfm.spreading_factor == 12:
rfm.receive_timeout = 5
elif rfm.spreading_factor > 7:
rfm.receive_timeout = 2
print("receive_timeout set to: ", rfm.receive_timeout)
rfm.enable_crc = True
# send startup message
message = bytes(f"startup message from base", "UTF-8")
if rfm.spreading_factor == 6:
payload = bytearray(40)
rfm.payload_length = len(payload)
payload[0 : len(message)] = message
rfm.send(
payload,
keep_listening=True,
)
else:
rfm.send(
message,
keep_listening=True,
)
# Wait to receive packets.
print("Waiting for packets...")
# initialize flag and timer
# set a delay before sending the echo packet
# avoide multibples of .5 second to minimize chances of node missing
# the packet between receive attempts
transmit_delay = 0.75
last_transmit_time = 0
packet_received = False
while True:
if rfm.payload_ready():
packet_received = False
packet = rfm.receive(timeout=None)
if packet is not None:
# Received a packet!
# Print out the raw bytes of the packet:
print(f"Received (raw payload): {packet}")
print([hex(x) for x in packet])
print(f"RSSI: {rfm.last_rssi}")
packet_received = True
last_transmit_time = time.monotonic()
if packet_received and ((time.monotonic() - last_transmit_time) > transmit_delay):
# send back the received packet
if rfm.spreading_factor == 6:
payload = bytearray(40)
rfm.payload_length = len(payload)
payload[0 : len(packet)] = packet
rfm.send(
payload,
keep_listening=True,
)
else:
rfm.send(
packet,
keep_listening=True,
)
packet_received = False

View file

@ -1,105 +0,0 @@
# SPDX-FileCopyrightText: 2021 ladyada for Adafruit Industries
# SPDX-License-Identifier: MIT
# Example to send a packet periodically between addressed nodes with ACK
# Author: Jerry Needell
#
import time
import board
import busio
import digitalio
# Define radio parameters.
RADIO_FREQ_MHZ = 915.0 # Frequency of the radio in Mhz. Must match your
# module! Can be a value like 915.0, 433.0, etc.
# Define pins connected to the chip, use these if wiring up the breakout according to the guide:
CS = digitalio.DigitalInOut(board.CE1)
RESET = digitalio.DigitalInOut(board.D25)
# Initialize SPI bus.
spi = busio.SPI(board.SCK, MOSI=board.MOSI, MISO=board.MISO)
# Initialze RFM radio
# uncommnet the desired import and rfm initialization depending on the radio boards being used
# Use rfm9x for two RFM9x radios using LoRa
from adafruit_rfm import rfm9x
rfm = rfm9x.RFM9x(spi, CS, RESET, RADIO_FREQ_MHZ)
rfm.radiohead = False # Do not use RadioHead Header
# set spreading factor
rfm.spreading_factor = 7
print("spreading factor set to :", rfm.spreading_factor)
print("low_datarate_optimize set to: ", rfm.low_datarate_optimize)
# rfm.signal_bandwidth = 500000
print("signal_bandwidth set to :", rfm.signal_bandwidth)
print("low_datarate_optimize set to: ", rfm.low_datarate_optimize)
if rfm.spreading_factor == 12:
rfm.xmit_timeout = 5
print("xmit_timeout set to: ", rfm.xmit_timeout)
if rfm.spreading_factor == 12:
rfm.receive_timeout = 5
elif rfm.spreading_factor > 7:
rfm.receive_timeout = 2
print("receive_timeout set to: ", rfm.receive_timeout)
rfm.enable_crc = True
# set the time interval (seconds) for sending packets
transmit_interval = 10
# initialize counter
counter = 0
# send startup message from my_node
message = bytes(f"startup message from node", "UTF-8")
if rfm.spreading_factor == 6:
payload = bytearray(40)
rfm.payload_length = len(payload)
payload[0 : len(message)] = message
rfm.send(
payload,
keep_listening=True,
)
else:
rfm.send(
message,
keep_listening=True,
)
# Wait to receive packets.
print("Waiting for packets...")
# initialize flag and timer
last_transmit_time = time.monotonic()
while True:
# Look for a new packet: only accept if addresses to my_node
packet = rfm.receive()
# If no packet was received during the timeout then None is returned.
if packet is not None:
# Received a packet!
# Print out the raw bytes of the packet:
print(f"Received (raw payload): {packet}")
print([hex(x) for x in packet])
print(f"RSSI: {rfm.last_rssi}")
# send reading after any packet received
if time.monotonic() - last_transmit_time > transmit_interval:
# reset timeer
last_transmit_time = time.monotonic()
# send a mesage to destination_node from my_node
message = bytes(f"message from node {counter}", "UTF-8")
if rfm.spreading_factor == 6:
payload = bytearray(40)
rfm.payload_length = len(payload)
payload[0 : len(message)] = message
rfm.send(
payload,
keep_listening=True,
)
else:
rfm.send(
message,
keep_listening=True,
)
counter += 1