Python client for GEX
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
gex-client-py/demo_nrf24.py

137 lines
3.7 KiB

import time
import gex
import sx_fsk as sx
# this is a demo with two NRF24L01+ modules connected to SPI and some GPIO.
# using the ESB function.
class Nrf:
def __init__(self, ce: gex.DOut, irq: gex.DIn, spi: gex.SPI, num):
self.snum = num
self.ce = ce
self.irq = irq
self.spi = spi
def rd(self, addr, count=1):
# addr 0-31
return self.spi.query(self.snum, [addr&0x1F], rlen=count)
def wr(self, addr, vals):
if type(vals) == int:
vals = [vals]
# addr 0-31
ba = bytearray()
ba.append(addr | 0x20)
ba.extend(vals)
self.spi.write(self.snum, ba)
def rd_payload(self, count):
""" Read a received payload """
return self.spi.query(self.snum, [0x61], rlen=count)
def wr_payload(self, pld):
""" Write a payload """
ba = bytearray()
ba.append(0xA0)
ba.extend(pld)
self.spi.write(self.snum, ba)
def flush_tx(self):
self.spi.write(self.snum, [0xE1])
def flush_rx(self):
self.spi.write(self.snum, [0xE2])
def reuse_tx_pld(self):
self.spi.write(self.snum, [0xE3])
def get_pld_len(self):
""" Read length of the first Rx payload in the FIFO - available only if dyn len enabled """
return self.spi.query(self.snum, [0x60], rlen=1)[0]
def wr_ack_payload(self, pipe, pld):
""" Write a payload to be attached to the next sent ACK """
ba = bytearray()
ba.append(0xA8|(pipe&0x7))
ba.extend(pld)
self.spi.write(self.snum, ba)
def wr_payload_noack(self, pld):
""" Send a payload without ACK """
ba = bytearray()
ba.append(0xB0)
ba.extend(pld)
self.spi.write(self.snum, ba)
def status(self):
""" Send a payload without ACK """
return self.spi.query(self.snum, [0xFF], rlen=1, rskip=0)[0]
with gex.Client(gex.TrxRawUSB()) as client:
spi = gex.SPI(client, 'spi')
ce = gex.DOut(client, 'ce')
irq = gex.DIn(client, 'irq')
a = Nrf(ce, irq, spi, 0)
b = Nrf(ce, irq, spi, 1)
a.flush_tx()
a.flush_rx()
b.flush_tx()
b.flush_rx()
# transmit demo
# a is PTX, b is PRX
ce.clear(0b11)
# a_adr = [0xA1,0xA2,0xA3,0xA4,0x01]
b_pipe0_adr = [0xA1, 0xA2, 0xA3, 0xA4, 0x02]
# --- Configure A for Tx of a 32-long payload ---
# PWR DN - simulate reset
a.wr(0x00, 0)
b.wr(0x00, 0)
time.sleep(0.001)
a.wr(0x00, 0b00001010) # set PWR_ON=1, EN_CRC=1, all irq enabled
a.wr_payload([0,1,2,3,4,5,6,7,8,9,10,11,12,13,14,15,16,17,18,19,20,21,22,23,24,25,26,27,28,29,30,31])
chnl = 5
# Set B's address as target and also pipe 0 Rx (used for ack)
a.wr(0x10, b_pipe0_adr) # target addr
a.wr(0x0A, b_pipe0_adr) # pipe 0 rx addr for ACK
a.wr(0x04, 0b00101111) # configure retransmit
a.wr(0x05, chnl)
# --- Configure B for Rx ---
b.wr(0x00, 0b00001011) # set PWR_ON=1, PRIM_RX=1, EN_CRC=1, all irq enabled
b.wr(0x02, 0b000001) # enable pipe 0
b.wr(0x11, 32) # set pipe 0 len to 11
b.wr(0x0A, b_pipe0_adr) # set pipe 0's address
b.wr(0x05, chnl)
ce.set(0b10) # CE high for B
time.sleep(0.01)
ce.pulse_us(us=10, pins=0b01) # Pulse A's CE
#
# testing B's FIFO by sending another payload...
a.wr_payload([0xFF, 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20, 21, 22, 23, 24, 25, 26, 27, 28, 29, 30])
ce.pulse_us(us=20, pins=0b01) # Pulse A's CE
time.sleep(0.01)
print("A's status after Tx: %02x" % a.status())
print("B's status after Tx: %02x" % b.status())
ce.clear(0b11)
# read the two payloads
print(b.rd_payload(32))
print(b.rd_payload(32))