RFM69 micropython for pi pico

[hamishpyle]

Hey there everyone!
I am a bit of a newbie in regard to micropython and would love some help!
Essentially I am running into a bunch of troubles with the rfm69hcw, I have attached my code below but it hangs while waiting for this to pass: while (self._readReg(REG_IRQFLAGS2) & RF_IRQFLAGS2_PACKETSENT) == 0x00:
It may have to do with the spi.locks using _thread but I am not sure at all, any help is appreciated thank you!
Note, some methods such as the board allocations for are yet to work for pin assigning so I have just done this through hardcoding.

rfm69.py
import time
import _thread
import machine
from machine import Pin, SPI
from registers import *
from packet import Packet
from config import get_config

from registers import RF69_315MHZ as FREQ_315MHZ
from registers import RF69_433MHZ as FREQ_433MHZ
from registers import RF69_868MHZ as FREQ_868MHZ
from registers import RF69_915MHZ as FREQ_915MHZ
from registers import RF69_MAX_DATA_LEN

class Radio:
"""RFM69 Radio interface for MicroPython.

An RFM69 module is expected to be connected to the SPI interface
of the microcontroller. The class can be used with the 'with' keyword.

Args:
    freqBand: Frequency band of radio - 315MHz, 868MHz, 433MHz, or 915MHz.
    nodeID (int): The node ID of this device.
    networkID (int): The network ID.

Keyword Args:
    auto_acknowledge (bool): Automatically send acknowledgements.
    isHighPower (bool): Is this a high power radio model.
    power (int): Power level - a percentage in range 10 to 100.
    use_board_pin_numbers (bool): Use BOARD (not BCM) pin numbers. Defaults to True.
    interruptPin (int): Pin number of interrupt pin. This is a pin index not a GPIO number.
    resetPin (int): Pin number of reset pin. This is a pin index not a GPIO number.
    spiBus (int): SPI bus number.
    spiDevice (int): SPI device number.
    promiscuousMode (bool): Listen to all messages not just those addressed to this node ID.
    enableATC (bool): Enable ATC mode. Defaults to False.
    encryptionKey (str): 16 character encryption key.
    verbose (bool): Verbose mode - Activates logging to console.
    sckPin (int): SPI clock pin number.
    mosiPin (int): SPI MOSI pin number.
    misoPin (int): SPI MISO pin number.
"""

def __init__(self, freqBand, nodeID, networkID=100, **kwargs):
    self.logger = None
    if kwargs.get('verbose', False):
        self.logger = self._init_log()

    self.auto_acknowledge = kwargs.get('autoAcknowledge', True)
    self.isRFM69HW = kwargs.get('isHighPower', True)
    self.intPin = kwargs.get('interruptPin', 15)
    self.rstPin = kwargs.get('resetPin', 3)
    #self.spiBus = kwargs.get('spiBus', 0)  # SPI bus number in MicroPython
    self.spiDevice = kwargs.get('spiDevice', 0)  # SPI device number (not typically used in MicroPython)
    self.promiscuousMode = kwargs.get('promiscuousMode', 0)
    self.enableATC = kwargs.get('enableATC', False)
    self.csPin = Pin( 5, Pin.OUT, value=True )

    # SPI pin numbers from kwargs
    sckPin = kwargs.get('sckPin', 6)  # Default pin number for SCK
    mosiPin = kwargs.get('mosiPin', 7)  # Default pin number for MOSI
    misoPin = kwargs.get('misoPin', 4)  # Default pin number for MISO
    self.powerLevel = 31
    self.enableRSSIack = kwargs.get('rssiACK', False)
    self.lastRSSI = 0

    # Initialize synchronization primitives
    self.spiLock = _thread.allocate_lock()
    self.sendLock = _thread.allocate_lock()  # Use a lock for similar behavior to Condition
    self.intLock = _thread.allocate_lock()
    self.ackLock = _thread.allocate_lock()   # Use a lock for similar behavior to Condition
    self.modeLock = _thread.allocate_lock()  # No direct RLock; use a regular lock

    self.mode = ""
    self.mode_name = ""

    self.address = None
    self._networkID = None

    # ListenMode members
    self._isHighSpeed = True
    self._encryptKey = None
    self.listen_mode_set_durations(DEFAULT_LISTEN_RX_US, DEFAULT_LISTEN_IDLE_US)

    self._packets = []
    self._packetLock = _thread.allocate_lock() 
    self.acks = {}

    self._init_spi(sckPin, mosiPin, misoPin)
    self._init_gpio()
    self._initialize(freqBand, nodeID, networkID)

    self._encrypt(kwargs.get('encryptionKey', 0))
    self.set_power_level(kwargs.get('power', 70))
    if self.isRFM69HW:
        self.set_HighPower(True)

def _initialize(self, freqBand, nodeID, networkID):
    self._reset_radio()
    self._set_config(get_config(freqBand, networkID))
    self._setHighPower(self.isRFM69HW)
    # Wait for ModeReady
    while (self._readReg(REG_IRQFLAGS1) & RF_IRQFLAGS1_MODEREADY) == 0x00:
        pass

    self._setAddress(nodeID)
    self._freqBand = freqBand
    self._networkID = networkID
    self._init_interrupt()

def _init_gpio(self):
    # Initialize GPIO pins
    self.intPin = Pin(self.intPin, Pin.IN)
    if self.rstPin:
        self.rstPin = Pin(self.rstPin, Pin.OUT)

def _init_spi(self, sckPin, mosiPin, misoPin):
    # Initialize SPI
    self.spiBus=SPI(0, baudrate=50000, polarity=0, phase=0, firstbit=SPI.MSB, 
          sck=machine.Pin(6), mosi=machine.Pin(7), miso=machine.Pin(4))


def _reset_radio(self):
    if self.rstPin:
        # Hard reset the RFM module
        self.rstPin.value(1)
        time.sleep(0.3)
        self.rstPin.value(0)
        time.sleep(0.3)
    # Verify chip is syncing
    start = time.time()
    while self._readReg(REG_SYNCVALUE1) != 0xAA:
        self._writeReg(REG_SYNCVALUE1, 0xAA)
    start = time.time()
    while self._readReg(REG_SYNCVALUE1) != 0x55:
        self._writeReg(REG_SYNCVALUE1, 0x55)

def _set_config(self, config):
    for value in config.values():
        self._writeReg(value[0], value[1])

def _init_interrupt(self):
    # Add interrupt detection for GPIO pin
    self.intPin.irq(trigger=Pin.IRQ_RISING, handler=self._interruptHandler)

# init end

def __enter__(self):
    """When the context begins"""
    self.read_temperature()
    self.calibrate_radio()
    self.begin_receive()
    return self

def __exit__(self, *args):
    """When context exits (including when the script is terminated)"""
    self._shutdown()

def set_frequency(self, FRF):  # pragma: no cover
    """Set the radio frequency"""
    self._writeReg(REG_FRFMSB, FRF >> 16)
    self._writeReg(REG_FRFMID, FRF >> 8)
    self._writeReg(REG_FRFLSB, FRF)

def set_frequency_in_Hz(self, frequency_in_Hz):  # pragma: no cover
    """Set the radio frequency in Hertz

    Args:
        frequency_in_Hz (int): Value between 315000000 to 915000000 Hz.
    """
    step = 61.03515625
    freq = int(round(frequency_in_Hz / step))
    self._writeReg(REG_FRFMSB, freq >> 16)
    self._writeReg(REG_FRFMID, freq >> 8)
    self._writeReg(REG_FRFLSB, freq)

def get_frequency_in_Hz(self):
    """Get the radio frequency in Hertz"""
    step = 61.03515625
    freq = (self._readReg(REG_FRFMSB) << 16) + (self._readReg(REG_FRFMID) << 8) + self._readReg(REG_FRFLSB)
    return int(round(freq * step))

def sleep(self):
    """Put the radio into sleep mode"""
    self._setMode(RF69_MODE_SLEEP)

def set_network(self, network_id):
    """Set the network ID (sync)

    Args:
        network_id (int): Value between 1 and 254.
    """
    assert isinstance(network_id, int)
    assert network_id > 0 and network_id < 255
    self._networkID = network_id
    self._writeReg(REG_SYNCVALUE2, network_id)

def set_power_level(self, percent):
    """Set the transmit power level

    Args:
        percent (int): Value between 0 and 100.
    """
    assert isinstance(percent, int)  # type(percent) == int
    powerLevel_new = int(round(31 * (percent / 100)))
    if self.isRFM69HW:
        if powerLevel_new > 23:
            powerLevel_new = 23

        self.powerLevel = powerLevel_new

        if self.powerLevel < 16:
            powerLevel_new += 16
            PA_SETTING = RF_PALEVEL_PA1_ON  # enable PA1 only
        else:
            if self.powerLevel < 20:
                powerLevel_new += 10
            else:
                powerLevel_new += 8
            PA_SETTING = RF_PALEVEL_PA1_ON | RF_PALEVEL_PA2_ON  # enable PA1+PA2

        self.set_HighPower_Regs(True)  # always call this in case we're crossing power boundaries in TX mode

    else:  # this is a W/CW, register value is the same as _powerLevel
        if powerLevel_new > 31:
            powerLevel_new = 31
        self.powerLevel = powerLevel_new
        PA_SETTING = RF_PALEVEL_PA0_ON  # enable PA0 only

    # write value to REG_PALEVEL
    self._writeReg(REG_PALEVEL, PA_SETTING | powerLevel_new)

def set_HighPower(self, _isRFM69HW_HCW):
    assert isinstance(_isRFM69HW_HCW, bool)

    self.isRFM69HW = _isRFM69HW_HCW
    if self.isRFM69HW:
        self._writeReg(REG_OCP, RF_OCP_OFF)  # disable OverCurrentProtection for HW/HCW
    else:
        self._writeReg(REG_OCP, RF_OCP_ON)

    self.set_power_level(self.powerLevel)

def set_HighPower_Regs(self, enable):
    assert isinstance(enable, bool)

    if (not self.isRFM69HW) or self.powerLevel < 20:
        self._writeReg(REG_TESTPA1, 0x55)
        self._writeReg(REG_TESTPA2, 0x70)
    else:
        self._writeReg(REG_TESTPA1, 0x5D)
        self._writeReg(REG_TESTPA2, 0x7C)


def _send(self, toAddress, buff="", requestACK=False):
    """Send a message with a retry mechanism."""
    self._writeReg(REG_PACKETCONFIG2,
                   (self._readReg(REG_PACKETCONFIG2) & 0xFB) | RF_PACKET2_RXRESTART)
    now = time.ticks_ms()
    while not self._canSend() and time.ticks_diff(time.ticks_ms(), now) < RF69_CSMA_LIMIT_S:
        pass
    self._sendFrame(toAddress, buff, requestACK, False)

def broadcast(self, buff=""):
    """Broadcast a message to the network."""
    self.send(RF69_BROADCAST_ADDR, buff, attempts=1, require_ack=False)

def send(self, toAddress, buff="", **kwargs):
    """Send a message with optional acknowledgements."""
    attempts = kwargs.get('attempts', 3)
    wait_time = kwargs.get('wait', 50)
    require_ack = kwargs.get('require_ack', True)
    if attempts > 1:
        require_ack = True

    for _ in range(attempts):
        self._send(toAddress, buff, attempts > 1)

        if not require_ack:
            return None

        start = time.ticks_ms()
        while time.ticks_diff(time.ticks_ms(), start) < (wait_time * 1000):
            if self._ACKReceived(toAddress):
                return True

    return False

def read_temperature(self, calFactor=0):
    """Read the temperature of the radio's CMOS chip."""
    self._setMode(RF69_MODE_STANDBY)
    self._writeReg(REG_TEMP1, RF_TEMP1_MEAS_START)
    while self._readReg(REG_TEMP1) & RF_TEMP1_MEAS_RUNNING:
        pass
    return (int(~self._readReg(REG_TEMP2)) * -1) + COURSE_TEMP_COEF + calFactor

def calibrate_radio(self):
    """Calibrate the internal RC oscillator."""
    self._writeReg(REG_OSC1, RF_OSC1_RCCAL_START)
    while self._readReg(REG_OSC1) & RF_OSC1_RCCAL_DONE == 0x00:
        pass

def read_registers(self):
    """Get all register values."""
    results = []
    for address in range(1, 0x5D):
        results.append([str(hex(address)), str(bin(self._readReg(address)))])
    return results

def begin_receive(self):
    """Begin listening for packets."""
    if self._readReg(REG_IRQFLAGS2) & RF_IRQFLAGS2_PAYLOADREADY:
        self._writeReg(REG_PACKETCONFIG2, (self._readReg(REG_PACKETCONFIG2) & 0xFB) | RF_PACKET2_RXRESTART)
    self._writeReg(REG_DIOMAPPING1, RF_DIOMAPPING1_DIO0_01)
    self._setMode(RF69_MODE_RX)

def has_received_packet(self):
    """Check if a packet has been received."""
    return len(self._packets) > 0

def get_packets(self):
    """Get newly received packets."""
    packets = list(self._packets)
    self._packets = []
    return packets

def send_ack(self, toAddress, buff=[]):
    """Send an acknowledgement packet."""
    while not self._canSend():
        pass

    if isinstance(buff, str):
        buff = [ord(i) for i in buff]

    if self.enableATC:
        buff = buff + [int(abs(self.lastRSSI))]
    self._sendFrame(toAddress, buff, False, True)

@property
def packets(self):
    """Deprecated property to get packets."""
    return self.get_packets()

def num_packets(self):
    """Returns the number of received packets."""
    return len(self._packets)

def get_packet(self, block=True, timeout=None):
    """Gets a single packet (thread-safe)."""
    if len(self._packets) > 0:
        return self._packets.pop(0)

    if block:
        start = time.ticks_ms()
        while timeout is None or time.ticks_diff(time.ticks_ms(), start) < (timeout * 1000):
            if self.has_received_packet():
                return self._packets.pop(0)

    return None

def _setMode(self, newMode):
    """Set the operating mode of the radio."""
    if newMode == self.mode or newMode not in [RF69_MODE_TX, RF69_MODE_RX, RF69_MODE_SYNTH, RF69_MODE_STANDBY, RF69_MODE_SLEEP]:
        return
    if newMode == RF69_MODE_TX:
        self.mode_name = "TX"
        self._writeReg(REG_OPMODE, (self._readReg(REG_OPMODE) & 0xE3) | RF_OPMODE_TRANSMITTER)
        if self.isRFM69HW:
            self.set_HighPower_Regs(True)
    elif newMode == RF69_MODE_RX:
        self.mode_name = "RX"
        self._writeReg(REG_OPMODE, (self._readReg(REG_OPMODE) & 0xE3) | RF_OPMODE_RECEIVER)
        if self.isRFM69HW:
            self.set_HighPower_Regs(False)
    elif newMode == RF69_MODE_SYNTH:
        self.mode_name = "Synth"
        self._writeReg(REG_OPMODE, (self._readReg(REG_OPMODE) & 0xE3) | RF_OPMODE_SYNTHESIZER)
    elif newMode == RF69_MODE_STANDBY:
        self.mode_name = "Standby"
        self._writeReg(REG_OPMODE, (self._readReg(REG_OPMODE) & 0xE3) | RF_OPMODE_STANDBY)
    elif newMode == RF69_MODE_SLEEP:
        self.mode_name = "Sleep"
        self._writeReg(REG_OPMODE, (self._readReg(REG_OPMODE) & 0xE3) | RF_OPMODE_SLEEP)
    while self.mode == RF69_MODE_SLEEP and self._readReg(REG_IRQFLAGS1) & RF_IRQFLAGS1_MODEREADY == 0x00:
        pass

    self.mode = newMode

def _setAddress(self, addr):
    self.address = addr
    self._writeReg(REG_NODEADRS, self.address)

def _canSend(self):
    if self.mode == RF69_MODE_STANDBY:
        self.begin_receive()
        return True
    elif mode == RF69_MODE_RX and _readRSSI() < CSMA_LIMIT:
        self._setMode(RF69_MODE_STANDBY)
        return True
    return False

def _ACKReceived(self, fromNodeID):
    if fromNodeID in acks:
        acks.pop(fromNodeID, None)
        return True
    return False

def _sendFrame(self, toAddress, buff, requestACK, sendACK):
    # Set to standby mode and wait until the mode is ready
    self._setMode(RF69_MODE_STANDBY)
    print("Setting mode to STANDBY")
    while (self._readReg(REG_IRQFLAGS1) & RF_IRQFLAGS1_MODEREADY) == 0x00:
        print("Waiting for mode to be ready...")
        print("Current IRQFLAGS1 register value:", bin(self._readReg(REG_IRQFLAGS1)))
        time.sleep(0.01)
    print("Mode is ready")

    # Configure DIO mapping
    self._writeReg(REG_DIOMAPPING1, RF_DIOMAPPING1_DIO0_00)
    print("DIO mapping set")
    print("Current DIOMAPPING1 register value:", bin(self._readReg(REG_DIOMAPPING1)))

    # Check buffer length
    if len(buff) > RF69_MAX_DATA_LEN:
        buff = buff[:RF69_MAX_DATA_LEN]
    print("Buffer length adjusted to:", len(buff))

    # Determine ACK settings
    ack = 0
    if sendACK:
        ack = 0x80
        if enableATC:
            ack |= 0x20
        print("Sending ACK with ATC enabled" if enableATC else "Sending ACK")
    elif requestACK:
        ack = 0x40
        print("Requesting ACK")

    # Convert buffer to list of bytes if it's a string
    with self.spiLock:
        if isinstance(buff, str):
            buff = [ord(i) for i in buff]
            print("Converted buffer from string to list of bytes:", buff)

    # Prepare and send data
    # Adjusted packet format example
    packet = bytearray([REG_FIFO | 0x80, len(buff) + 5, toAddress, self.address, ack] + buff)
    print("Sending packet:", list(packet))
    try:
        self.spiBus.write(packet)
        print("Packet written to SPI bus")
    except Exception as e:
        print("SPI write error:", e)

    # Wait for transmission to complete
    with self.sendLock:
        self._setMode(RF69_MODE_TX)
        print("Setting mode to TX")
        timeout = 10  # Increase timeout to 10 seconds
        start_time = time.time()
        while (self._readReg(REG_IRQFLAGS2) & RF_IRQFLAGS2_PACKETSENT) == 0x00:
            current_time = time.time()
            if (current_time - start_time) > timeout:
                print("Timeout waiting for packet sent flag")
                break
            print("Waiting for packet sent flag...")
            print("Current IRQFLAGS2 register value:", bin(self._readReg(REG_IRQFLAGS2)))
            print("Current OPMODE register value:", bin(self._readReg(REG_OPMODE)))
            time.sleep(0.1)
        else:
            print("Packet sent flag detected")

    # Set to receive mode
    self._setMode(RF69_MODE_RX)
    print("Setting mode to RX")





def _readRSSI(self, forceTrigger=False):
    if forceTrigger:
        self._writeReg(REG_RSSICONFIG, RF_RSSI_START)
        while self._readReg(REG_RSSICONFIG) & RF_RSSI_DONE == 0x00:
            pass
    rssi = self._readReg(REG_RSSIVALUE) * -1
    rssi = rssi >> 1
    return rssi

def _encrypt(self, key):
    self._setMode(RF69_MODE_STANDBY)
    if key != 0 and len(key) == 16:
        encryptKey = key
        with spiLock:
            spi.xfer([REG_AESKEY1 | 0x80] + [ord(i) for i in key])
        self._writeReg(REG_PACKETCONFIG2, (self._readReg(REG_PACKETCONFIG2) & 0xFE) | RF_PACKET2_AES_ON)
    else:
        encryptKey = None
        self._writeReg(REG_PACKETCONFIG2, (self._readReg(REG_PACKETCONFIG2) & 0xFE) | RF_PACKET2_AES_OFF)

def _readReg(self, register):
        # Read U8 register from module
    data = bytearray(2)
    data[0] = register & ~0x80
    data[1] = 0
    resp = bytearray(2)
    self.csPin.low()
    self.spiBus.write_readinto(data, resp )
    self.csPin.high()
    return resp[1]

def _writeReg(self, register, value):
    # Write U8 value into a module register
    if (type(value) is bytearray) or (type(value) is bytes):
        data = bytearray(1)
        data[0] = register | 0x80
        self.csPin.low()
        self.spiBus.write(data )
        self.spiBus.write(value)
        self.csPin.high()
    else:
        data = bytearray(2)
        data[0] = register | 0x80
        data[1] = value
        self.csPin.low()
        self.spiBus.write(data )
        self.csPin.high()

def _promiscuous(onOff):
    promiscuousMode = onOff

def _setHighPower(self, onOff):
    if onOff:
        self._writeReg(REG_OCP, RF_OCP_OFF)
        self._writeReg(REG_PALEVEL, (self._readReg(REG_PALEVEL) & 0x1F) | RF_PALEVEL_PA1_ON | RF_PALEVEL_PA2_ON)
    else:
        self._writeReg(REG_OCP, RF_OCP_ON)
        self._writeReg(REG_PALEVEL, RF_PALEVEL_PA0_ON | RF_PALEVEL_PA1_OFF | RF_PALEVEL_PA2_OFF | powerLevel)

def _setHighPowerRegs(self, onOff):
    if onOff:
        self._writeReg(REG_TESTPA1, 0x5D)
        self._writeReg(REG_TESTPA2, 0x7C)
    else:
        self._writeReg(REG_TESTPA1, 0x55)
        self._writeReg(REG_TESTPA2, 0x70)

def _shutdown(self):
    GPIO.remove_event_detect(intPin)
    modeLock.acquire()
    self._setHighPower(False)
    sleep()
    GPIO.cleanup([intPin, rstPin])
    self.intLock.acquire()
    self.spiLock.acquire()
    spi.close()

def _init_log(self):
    import logging
    logging.basicConfig(level=logging.DEBUG)
    logger = logging.getLogger(__name__)
    handler = logging.StreamHandler()
    handler.setLevel(logging.DEBUG)
    formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(thread)d - %(message)s')
    handler.setFormatter(formatter)
    logger.addHandler(handler)
    logger.propagate = False
    return logger

def _debug(self, *args):
    if logger is not None:
        logger.debug(*args)

def _error(self, *args):
    if logger is not None:
        logger.error(*args)

def _interruptHandler(self, pin):
    intLock.acquire()
    with modeLock:
        with sendLock:
            sendLock.notify_all()

        if mode == RF69_MODE_RX and _readReg(REG_IRQFLAGS2) & RF_IRQFLAGS2_PAYLOADREADY:
            _setMode(RF69_MODE_STANDBY)

            with self.spiLock:
                payload_length, target_id, sender_id, CTLbyte = spi.xfer2([REG_FIFO & 0x7f, 0, 0, 0, 0])[1:]

            payload_length = min(payload_length, 66)

            if not (promiscuousMode or target_id == address or target_id == RF69_BROADCAST_ADDR):
                _debug("Ignore Interrupt")
                intLock.release()
                self.begin_receive()
                return

            data_length = payload_length - 3
            ack_received = bool(CTLbyte & 0x80)
            ack_requested = bool(CTLbyte & 0x40) and target_id == address
            with self.spiLock:
                data = spi.xfer2([REG_FIFO & 0x7f] + [0 for _ in range(data_length)])[1:]
            lastRSSI = _readRSSI()

            if ack_received:
                _debug("Incoming ack from {}".format(sender_id))
                with ackLock:
                    acks.setdefault(sender_id, 1)
                    ackLock.notify_all()
            elif ack_requested:
                _debug("replying to ack request")
            else:
                _debug("Other ??")

            if not ack_received:
                _debug("Incoming data packet")
                with packetLock:
                    packets.append(
                        Packet(int(target_id), int(sender_id), int(lastRSSI), list(data))
                    )
                    packetLock.notify_all()

            if ack_requested and auto_acknowledge:
                if enableRSSIack:
                    _debug("Sending an RSSI ack")
                    rssi_back = list(struct.pack('B', abs(lastRSSI)))
                    intLock.release()
                    send_ack(sender_id, rssi_back)
                    _debug("RSSI ack sent")
                    self.begin_receive()
                    return
                else:
                    _debug("Sending a normal ack")
                    intLock.release()
                    send_ack(sender_id)
                    self.begin_receive()
                    return

            intLock.release()
            self.begin_receive()
            return

    intLock.release()

def _reinitRadio(self):
    _initialize(freqBand, address, networkID)
    if encryptKey:
        _encrypt(encryptKey)
    if isHighSpeed:
        _writeReg(REG_LNA, (_readReg(REG_LNA) & ~0x3) | RF_LNA_GAINSELECT_AUTO)

def _getUsForResolution(self, resolution):
    if resolution == RF_LISTEN1_RESOL_RX_64 or resolution == RF_LISTEN1_RESOL_IDLE_64:
        return 64
    elif resolution == RF_LISTEN1_RESOL_RX_4100 or resolution == RF_LISTEN1_RESOL_IDLE_4100:
        return 4100
    elif resolution == RF_LISTEN1_RESOL_RX_262000 or resolution == RF_LISTEN1_RESOL_IDLE_262000:
        return 262000
    return 0

def _getCoefForResolution(self, resolution, duration):
    resolDuration = self._getUsForResolution(resolution)
    result = int(duration / resolDuration)
    if abs(duration - ((result + 1) * resolDuration)) < abs(duration - (result * resolDuration)):
        return result + 1
    return result

def _chooseResolutionAndCoef(self, resolutions, duration):
    for resolution in resolutions:
        coef = self._getCoefForResolution(resolution, duration)
        if coef <= 255:
            return (resolution, coef)
    return (None, None)

def listen_mode_set_durations(self, rxDuration, idleDuration):
    rxResolutions = [RF_LISTEN1_RESOL_RX_64, RF_LISTEN1_RESOL_RX_4100, RF_LISTEN1_RESOL_RX_262000]
    idleResolutions = [RF_LISTEN1_RESOL_IDLE_64, RF_LISTEN1_RESOL_IDLE_4100, RF_LISTEN1_RESOL_IDLE_262000]

    resolOut, coefOut = self._chooseResolutionAndCoef(rxResolutions, rxDuration)
    if resolOut and coefOut:
        rxListenResolution = resolOut
        rxListenCoef = coefOut
    else:
        return (None, None)

    resolOut, coefOut = self._chooseResolutionAndCoef(idleResolutions, idleDuration)
    if resolOut and coefOut:
        idleListenResolution = resolOut
        idleListenCoef = coefOut
    else:
        return (None, None)

    rxDuration = self._getUsForResolution(rxListenResolution) * rxListenCoef
    idleDuration = self._getUsForResolution(idleListenResolution) * idleListenCoef
    listenCycleDurationUs = rxDuration + idleDuration
    return (rxDuration, idleDuration)

def listen_mode_get_durations(self):
    rxDuration = _getUsForResolution(rxListenResolution) * rxListenCoef
    idleDuration = _getUsForResolution(idleListenResolution) * idleListenCoef
    return (rxDuration, idleDuration)

def _listenModeApplyHighSpeedSettings(self):
    if not isHighSpeed:
        return
    _writeReg(REG_BITRATEMSB, RF_BITRATEMSB_200000)
    _writeReg(REG_BITRATELSB, RF_BITRATELSB_200000)
    _writeReg(REG_FDEVMSB, RF_FDEVMSB_100000)
    _writeReg(REG_FDEVLSB, RF_FDEVLSB_100000)
    _writeReg(REG_RXBW, RF_RXBW_DCCFREQ_000 | RF_RXBW_MANT_20 | RF_RXBW_EXP_0)

def listen_mode_send_burst(self, toAddress, buff):
    GPIO.remove_event_detect(intPin)
    _setMode(RF69_MODE_STANDBY)
    _writeReg(REG_PACKETCONFIG1, RF_PACKET1_FORMAT_VARIABLE | RF_PACKET1_DCFREE_WHITENING | RF_PACKET1_CRC_ON | RF_PACKET1_CRCAUTOCLEAR_ON)
    _writeReg(REG_PACKETCONFIG2, RF_PACKET2_RXRESTARTDELAY_NONE | RF_PACKET2_AUTORXRESTART_ON | RF_PACKET2_AES_OFF)
    _writeReg(REG_SYNCVALUE1, 0x5A)
    _writeReg(REG_SYNCVALUE2, 0x5A)
    _listenModeApplyHighSpeedSettings()
    _writeReg(REG_FRFMSB, _readReg(REG_FRFMSB) + 1)
    _writeReg(REG_FRFLSB, _readReg(REG_FRFLSB))

    cycleDurationMs = int(listenCycleDurationUs / 1000)
    timeRemaining = int(cycleDurationMs)
    _setMode(RF69_MODE_TX)
    startTime = int(time.time() * 1000)

    while timeRemaining > 0:
        with self.spiLock:
            if isinstance(buff, str):
                spi.xfer2([REG_FIFO | 0x80, len(buff) + 4, toAddress, address, timeRemaining & 0xFF, (timeRemaining >> 8) & 0xFF] + [ord(i) for i in buff])
            else:
                spi.xfer2([REG_FIFO | 0x80, len(buff) + 4, toAddress, address, timeRemaining & 0xFF, (timeRemaining >> 8) & 0xFF] + buff)

        while (_readReg(REG_IRQFLAGS2) & RF_IRQFLAGS2_FIFONOTEMPTY) != 0x00:
            pass
        timeRemaining = cycleDurationMs - (int(time.time() * 1000) - startTime)

    self._setMode(RF69_MODE_STANDBY)
    self._reinitRadio()
    self.begin_receive()

packet.py:

import json
import time

class Packet:
"""Object to represent received packet. Created internally and
returned by radio when getPackets() is called.

Args:
    receiver (int): Node ID of receiver
    sender (int): Node ID of sender
    RSSI (int): Received Signal Strength Indicator i.e. the power present in a received radio signal
    data (list): Raw transmitted data
"""

# Declare slots to reduce memory
__slots__ = 'received', 'received_epoch', 'receiver', 'sender', 'RSSI', 'data'

def __init__(self, receiver, sender, RSSI, data):
    self.received = time.time()  # Use time.time() for the timestamp
    self.received_epoch = int(self.received * 1000)  # Convert to milliseconds
    self.receiver = receiver
    self.sender = sender
    self.RSSI = RSSI
    self.data = data

def to_dict(self, dateFormat=None):
    """Returns a dictionary representation of the class data"""
    if dateFormat is None:
        return_date = self.received
    else:
        return_date = time.strftime(dateFormat, time.gmtime(self.received))
    return dict(received=return_date, receiver=self.receiver,
                sender=self.sender, rssi=self.RSSI, data=self.data)

@property
def data_string(self):
    """Returns the data as a string"""
    return "".join([chr(letter) for letter in self.data])

def __str__(self):
    return json.dumps(self.to_dict())

def __repr__(self):
    return "Radio({}, {}, {}, [data])".format(self.receiver, self.sender, self.RSSI)

config.py:

Importing constants from registers.py

from registers import *

Frequency band settings

frfMSB = {
RF69_315MHZ: RF_FRFMSB_315,
RF69_433MHZ: RF_FRFMSB_433,
RF69_868MHZ: RF_FRFMSB_868,
RF69_915MHZ: RF_FRFMSB_915
}

frfMID = {
RF69_315MHZ: RF_FRFMID_315,
RF69_433MHZ: RF_FRFMID_433,
RF69_868MHZ: RF_FRFMID_868,
RF69_915MHZ: RF_FRFMID_915
}

frfLSB = {
RF69_315MHZ: RF_FRFLSB_315,
RF69_433MHZ: RF_FRFLSB_433,
RF69_868MHZ: RF_FRFLSB_868,
RF69_915MHZ: RF_FRFLSB_915
}

def get_config(freqBand, networkID):
return {
0x01: [REG_OPMODE, RF_OPMODE_SEQUENCER_ON | RF_OPMODE_LISTEN_OFF | RF_OPMODE_STANDBY],
0x02: [REG_DATAMODUL, RF_DATAMODUL_DATAMODE_PACKET | RF_DATAMODUL_MODULATIONTYPE_FSK | RF_DATAMODUL_MODULATIONSHAPING_00],
0x03: [REG_BITRATEMSB, RF_BITRATEMSB_55555], # Example bitrate; adjust as needed
0x04: [REG_BITRATELSB, RF_BITRATELSB_55555], # Example bitrate; adjust as needed
0x05: [REG_FDEVMSB, RF_FDEVMSB_50000], # Example frequency deviation; adjust as needed
0x06: [REG_FDEVLSB, RF_FDEVLSB_50000], # Example frequency deviation; adjust as needed
0x07: [REG_FRFMSB, frfMSB[freqBand]],
0x08: [REG_FRFMID, frfMID[freqBand]],
0x09: [REG_FRFLSB, frfLSB[freqBand]],
0x19: [REG_RXBW, RF_RXBW_DCCFREQ_010 | RF_RXBW_MANT_16 | RF_RXBW_EXP_2],
0x25: [REG_DIOMAPPING1, RF_DIOMAPPING1_DIO0_01],
0x29: [REG_RSSITHRESH, 220], # RSSI threshold value
0x2e: [REG_SYNCCONFIG, RF_SYNC_ON | RF_SYNC_FIFOFILL_AUTO | RF_SYNC_SIZE_2 | RF_SYNC_TOL_0],
0x2f: [REG_SYNCVALUE1, 0x2D],
0x30: [REG_SYNCVALUE2, networkID],
0x37: [REG_PACKETCONFIG1, RF_PACKET1_FORMAT_VARIABLE | RF_PACKET1_DCFREE_OFF |
RF_PACKET1_CRC_ON | RF_PACKET1_CRCAUTOCLEAR_ON | RF_PACKET1_ADRSFILTERING_OFF],
0x38: [REG_PAYLOADLENGTH, 66], # Payload length
0x3C: [REG_FIFOTHRESH, RF_FIFOTHRESH_TXSTART_FIFONOTEMPTY | RF_FIFOTHRESH_VALUE],
0x3D: [REG_PACKETCONFIG2, RF_PACKET2_RXRESTARTDELAY_2BITS | RF_PACKET2_AUTORXRESTART_ON | RF_PACKET2_AES_OFF],
0x6F: [REG_TESTDAGC, RF_DAGC_IMPROVED_LOWBETA0], # DAGC settings
0x00: [255, 0] # End of configuration
}
the registers.py is the same as the rpi-rfm69 found here: https://github.com/jgillula/rpi-rfm69/blob/main/RFM69/radio.py and the tester script is: import time
import machine
from machine import Pin, SPI
from rfm69 import Radio, FREQ_315MHZ, FREQ_433MHZ, FREQ_868MHZ, FREQ_915MHZ

Configuration

node_id = 1
network_id = 100
recipient_id = 2

Board configuration

Adjust these as per your specific hardware setup

board = {
'isHighPower': True,
'interruptPin': 15,
'resetPin': 3,
'spiDevice': 0,
'mosiPin': 7,
'misoPin': 4,
'sckPin':6
}

Setup SPI and GPIO for the radio

interrupt_pin = Pin(board['interruptPin'], Pin.IN)
reset_pin = Pin(board['resetPin'], Pin.OUT)

def reset_radio():
reset_pin.value(0)
time.sleep(0.1)
reset_pin.value(1)
time.sleep(0.1)

Create Radio instance

radio = Radio(FREQ_915MHZ, node_id, network_id, **board)

Main loop

print("Starting loop...")
while True:
start_time = time.time()

# Get packets for at most 5 seconds
while time.time() - start_time < 5:
    time_remaining = max(0, start_time + 5 - time.time())

    # This call will block until a packet is received,
    # or timeout in however much time we have left
    packet = radio.get_packet(timeout=time_remaining)

    if packet is not None:
        # Process packet
        print(packet)
    
# After 5 seconds send a message
print("Sending")
if radio.send(recipient_id, "TEST", attempts=3, wait_time=10):
    print("Acknowledgement received")
else:
    print("No Acknowledgement")

radio.close()
spi.deinit()
interrupt_pin.init(Pin.IN)
reset_pin.init(Pin.OUT)

[peterhinch]

Please provide a link to the driver source.

[hamishpyle]

So the driver source I assume would be this: https://cdn.sparkfun.com/datasheets/Wireless/General/RFM69HCW-V1.1.pdf and the library that I was trying to convert was this: https://github.com/jgillula/rpi-rfm69.
Cheers!

[hamishpyle]

Someone was also able to get this receiving but not sending in the following thread: https://forums.raspberrypi.com/viewtopic.php?t=353136 but I am struggling to implement a sending method for the life of me.

[ElTopollillo1990]

@hamishpyle - I have not had any time to work on the sender still (i.e. transmitter side) but I have had the receiver side working now for over a year for Senders coded with LowPower Lab's Arduino senders. It is certainly not elegant, but it works for my needs. I cleaned up the repo a bit and finally published it: https://github.com/ElTopollillo1990/microPython-RFM69

Hope it helps a bit. I will update that post you referenced as well (just to close the loop).

[sophiedegran]

hamish
Don't know this board.. but it's a 'simply' Semtech SX1231H .. (12 years old transceiver!)
Have a look to the Semtech doc..
SX1231H was made before Lora band exist...
Good luck