Raspberry Pico i nRF24L01+ - ktoś to ożenił?

[ethanak]

Pytanko jak w temacie: ktoś to może uruchomił albo zna działającą bibliotekę? Mi się jakoś nie udaje (tzn. moduł działa ślicznie, w rejestrach ma wszystko co trzeba, ogólnie miód i malinki tylko odbierać nie chce)...

[Lukasc]

Na pewno do tego czasu poradziłeś sobie, ale może dla potomnych wrzucę tutaj działające rozwiązanie, które ja używam.

Mała uwaga - nrf24l01+ to bardzo delikatny moduł. Udało mi się już dwie sztuki "spalić". Trzeba obchodzić się z nim jak z jajkiem...

1. Schemat podłączenia na przykład na tej stronie albo tej

(3,3 V!!)

 

2. Treść biblioteki  - plik nrf24l01.py (plik kopiujemy i wrzucamy albo do folderu głównego albo folderu lib)

"""NRF24L01 driver for MicroPython
"""

from micropython import const
import utime

# nRF24L01+ registers
CONFIG = const(0x00)
EN_RXADDR = const(0x02)
SETUP_AW = const(0x03)
SETUP_RETR = const(0x04)
RF_CH = const(0x05)
RF_SETUP = const(0x06)
STATUS = const(0x07)
RX_ADDR_P0 = const(0x0A)
TX_ADDR = const(0x10)
RX_PW_P0 = const(0x11)
FIFO_STATUS = const(0x17)
DYNPD = const(0x1C)

# CONFIG register
EN_CRC = const(0x08)  # enable CRC
CRCO = const(0x04)  # CRC encoding scheme; 0=1 byte, 1=2 bytes
PWR_UP = const(0x02)  # 1=power up, 0=power down
PRIM_RX = const(0x01)  # RX/TX control; 0=PTX, 1=PRX

# RF_SETUP register
POWER_0 = const(0x00)  # -18 dBm
POWER_1 = const(0x02)  # -12 dBm
POWER_2 = const(0x04)  # -6 dBm
POWER_3 = const(0x06)  # 0 dBm
SPEED_1M = const(0x00)
SPEED_2M = const(0x08)
SPEED_250K = const(0x20)

# STATUS register
RX_DR = const(0x40)  # RX data ready; write 1 to clear
TX_DS = const(0x20)  # TX data sent; write 1 to clear
MAX_RT = const(0x10)  # max retransmits reached; write 1 to clear

# FIFO_STATUS register
RX_EMPTY = const(0x01)  # 1 if RX FIFO is empty

# constants for instructions
R_RX_PL_WID = const(0x60)  # read RX payload width
R_RX_PAYLOAD = const(0x61)  # read RX payload
W_TX_PAYLOAD = const(0xA0)  # write TX payload
FLUSH_TX = const(0xE1)  # flush TX FIFO
FLUSH_RX = const(0xE2)  # flush RX FIFO
NOP = const(0xFF)  # use to read STATUS register


class NRF24L01:
    def __init__(self, spi, cs, ce, channel=46, payload_size=16):
        assert payload_size <= 32

        self.buf = bytearray(1)

        # store the pins
        self.spi = spi
        self.cs = cs
        self.ce = ce

        # init the SPI bus and pins
        self.init_spi(4000000)

        # reset everything
        ce.init(ce.OUT, value=0)
        cs.init(cs.OUT, value=1)

        self.payload_size = payload_size
        self.pipe0_read_addr = None
        utime.sleep_ms(5)

        # set address width to 5 bytes and check for device present
        self.reg_write(SETUP_AW, 0b11)
        if self.reg_read(SETUP_AW) != 0b11:
            raise OSError("nRF24L01+ Hardware not responding")

        # disable dynamic payloads
        self.reg_write(DYNPD, 0)

        # auto retransmit delay: 1750us
        # auto retransmit count: 8
        self.reg_write(SETUP_RETR, (6 << 4) | 8)

        # set rf power and speed
        self.set_power_speed(POWER_3, SPEED_250K)  # Best for point to point links

        # init CRC
        self.set_crc(2)

        # clear status flags
        self.reg_write(STATUS, RX_DR | TX_DS | MAX_RT)

        # set channel
        self.set_channel(channel)

        # flush buffers
        self.flush_rx()
        self.flush_tx()

    def init_spi(self, baudrate):
        try:
            master = self.spi.MASTER
        except AttributeError:
            self.spi.init(baudrate=baudrate, polarity=0, phase=0)
        else:
            self.spi.init(master, baudrate=baudrate, polarity=0, phase=0)

    def reg_read(self, reg):
        self.cs(0)
        self.spi.readinto(self.buf, reg)
        self.spi.readinto(self.buf)
        self.cs(1)
        return self.buf[0]

    def reg_write_bytes(self, reg, buf):
        self.cs(0)
        self.spi.readinto(self.buf, 0x20 | reg)
        self.spi.write(buf)
        self.cs(1)
        return self.buf[0]

    def reg_write(self, reg, value):
        self.cs(0)
        self.spi.readinto(self.buf, 0x20 | reg)
        ret = self.buf[0]
        self.spi.readinto(self.buf, value)
        self.cs(1)
        return ret

    def flush_rx(self):
        self.cs(0)
        self.spi.readinto(self.buf, FLUSH_RX)
        self.cs(1)

    def flush_tx(self):
        self.cs(0)
        self.spi.readinto(self.buf, FLUSH_TX)
        self.cs(1)

    # power is one of POWER_x defines; speed is one of SPEED_x defines
    def set_power_speed(self, power, speed):
        setup = self.reg_read(RF_SETUP) & 0b11010001
        self.reg_write(RF_SETUP, setup | power | speed)

    # length in bytes: 0, 1 or 2
    def set_crc(self, length):
        config = self.reg_read(CONFIG) & ~(CRCO | EN_CRC)
        if length == 0:
            pass
        elif length == 1:
            config |= EN_CRC
        else:
            config |= EN_CRC | CRCO
        self.reg_write(CONFIG, config)

    def set_channel(self, channel):
        self.reg_write(RF_CH, min(channel, 125))

    # address should be a bytes object 5 bytes long
    def open_tx_pipe(self, address):
        assert len(address) == 5
        self.reg_write_bytes(RX_ADDR_P0, address)
        self.reg_write_bytes(TX_ADDR, address)
        self.reg_write(RX_PW_P0, self.payload_size)

    # address should be a bytes object 5 bytes long
    # pipe 0 and 1 have 5 byte address
    # pipes 2-5 use same 4 most-significant bytes as pipe 1, plus 1 extra byte
    def open_rx_pipe(self, pipe_id, address):
        assert len(address) == 5
        assert 0 <= pipe_id <= 5
        if pipe_id == 0:
            self.pipe0_read_addr = address
        if pipe_id < 2:
            self.reg_write_bytes(RX_ADDR_P0 + pipe_id, address)
        else:
            self.reg_write(RX_ADDR_P0 + pipe_id, address[0])
        self.reg_write(RX_PW_P0 + pipe_id, self.payload_size)
        self.reg_write(EN_RXADDR, self.reg_read(EN_RXADDR) | (1 << pipe_id))

    def start_listening(self):
        self.reg_write(CONFIG, self.reg_read(CONFIG) | PWR_UP | PRIM_RX)
        self.reg_write(STATUS, RX_DR | TX_DS | MAX_RT)

        if self.pipe0_read_addr is not None:
            self.reg_write_bytes(RX_ADDR_P0, self.pipe0_read_addr)

        self.flush_rx()
        self.flush_tx()
        self.ce(1)
        utime.sleep_us(130)

    def stop_listening(self):
        self.ce(0)
        self.flush_tx()
        self.flush_rx()

    # returns True if any data available to recv
    def any(self):
        return not bool(self.reg_read(FIFO_STATUS) & RX_EMPTY)

    def recv(self):
        # get the data
        self.cs(0)
        self.spi.readinto(self.buf, R_RX_PAYLOAD)
        buf = self.spi.read(self.payload_size)
        self.cs(1)
        # clear RX ready flag
        self.reg_write(STATUS, RX_DR)

        return buf

    # blocking wait for tx complete
    def send(self, buf, timeout=500):
        self.send_start(buf)
        start = utime.ticks_ms()
        result = None
        while result is None and utime.ticks_diff(utime.ticks_ms(), start) < timeout:
            result = self.send_done()  # 1 == success, 2 == fail
        if result == 2:
            raise OSError("send failed")

    # non-blocking tx
    def send_start(self, buf):
        # power up
        self.reg_write(CONFIG, (self.reg_read(CONFIG) | PWR_UP) & ~PRIM_RX)
        utime.sleep_us(150)
        # send the data
        self.cs(0)
        self.spi.readinto(self.buf, W_TX_PAYLOAD)
        self.spi.write(buf)
        if len(buf) < self.payload_size:
            self.spi.write(b"\x00" * (self.payload_size - len(buf)))  # pad out data
        self.cs(1)

        # enable the chip so it can send the data
        self.ce(1)
        utime.sleep_us(15)  # needs to be >10us
        self.ce(0)

    # returns None if send still in progress, 1 for success, 2 for fail
    def send_done(self):
        if not (self.reg_read(STATUS) & (TX_DS | MAX_RT)):
            return None  # tx not finished

        # either finished or failed: get and clear status flags, power down
        status = self.reg_write(STATUS, RX_DR | TX_DS | MAX_RT)
        self.reg_write(CONFIG, self.reg_read(CONFIG) & ~PWR_UP)
        return 1 if status & TX_DS else 2

 

 

3. Oraz kod programu: nadajnik.py. Dla nadajnika: odkomentowujemy linię role = "send". Dla odbiornika role = "receive"

# demo.py
# Kevin McAleer
# test the nRF24L01 modules to send and receive data
# Watch this video for more information about that library https://www.youtube.com/watch?v=aP8rSN-1eT0

from nrf24l01 import NRF24L01
from machine import SPI, Pin
from time import sleep
import struct
led = Pin(25, Pin.OUT)

csn = Pin(14, mode=Pin.OUT, value=1) # Chip Select Not
ce = Pin(17, mode=Pin.OUT, value=0)  # Chip Enable
             # Onboard LED
payload_size = 20

# Define the channel or 'pipes' the radios use.
# switch round the pipes depending if this is a sender or receiver pico

#role = "send"
role = "receive"

if role == "send":
    send_pipe = b"\xe1\xf0\xf0\xf0\xf0"
    receive_pipe = b"\xd2\xf0\xf0\xf0\xf0"
else:
    send_pipe = b"\xd2\xf0\xf0\xf0\xf0"
    receive_pipe = b"\xe1\xf0\xf0\xf0\xf0"

def setup():
    print("Initialising the nRF24L0+ Module")
    nrf = NRF24L01(SPI(0), csn, ce, payload_size=payload_size)
    nrf.open_tx_pipe(send_pipe)
    nrf.open_rx_pipe(1, receive_pipe)
    nrf.start_listening()
    return nrf

def flash_led(times:int=None):
    ''' Flashed the built in LED the number of times defined in the times parameter '''
    for _ in range(times):
        led.value(1)
        sleep(0.01)
        led.value(0)
        sleep(0.01)

def send(nrf, msg):
    print("sending message.", msg)
    nrf.stop_listening()
    for n in range(len(msg)):
        try:
            encoded_string = msg[n].encode()
            byte_array = bytearray(encoded_string)
            buf = struct.pack("s", byte_array)
            nrf.send(buf)
            # print(role,"message",msg[n],"sent")
            flash_led(1)
        except OSError:
            print(role,"Sorry message not sent")
    nrf.send("\n")
    nrf.start_listening()

# main code loop
flash_led(1)
nrf = setup()
nrf.start_listening()
msg_string = ""

while True:
    msg = ""
    if role == "send":
        send(nrf, "Yello world")
        send(nrf, "Test")
    else:
        # Check for Messages
        if nrf.any():
            package = nrf.recv()          
            message = struct.unpack("s",package)
            msg = message[0].decode()
            flash_led(1)

            # Check for the new line character
            if (msg == "\n") and (len(msg_string) <= 20):
                print("full message",msg_string, msg)
                msg_string = ""
            else:
                if len(msg_string) <= 20:
                    msg_string = msg_string + msg
                else:
                    msg_string = ""

 

 

[ethanak]

Nie pamiętam o co wtedy chodziło, przepisałem całą bibliotekę i ruszył. W każdym razie dziękuję - będzie dla potomności.

Kod w C opublikuję na dniach przy okazji opisu najnowszego robota (jak wiadomo, stworzenie opisu to najtrudniejsza rzecz pod słońcem)

Edytowano Sierpień 11, 2022 przez ethanak

[ethanak]

Zgodnie z obietnicą - kod w C do wydłubania z kodu sterownika RC do Krasnoluda.

Nie jest to absolutnie moje dzieło - po prostu popularna biblioteka do Arduino przejechana skryptem w Pythonie, który z C++ zrobił C

 

[trainee]

Odplus-pluskwianie?

[ethanak]

8 godzin temu, trainee napisał:

Odplus-pluskwianie?

Coś w tym stylu czyli wąż co zjada krzyżyki

[krzysiek1985]

Myślę o zrobieniu w domu monitoringu temperatury w kilku pokojach i na zewnątrz, ale nie chcę plątaniny kabli dlatego myślałem o takim wstępnym szkicu:

nadajniki: raspberry pi pico + nrf24l01 + DS18B20 + źródło zasilania (baterie),

odbiornik: raspberry pi 4 + nrf24l01 (jako odbiornik, chyba, że nie potrzebuję i mam wszystko w malinie),

W temacie jestem całkowicie zielony - od ok. 2 tygodniu jestem szczęśliwym posiadaczem kursu RPi, dlatego proszę o ewentualne spostrzeżenia, uwagi, bo jeszcze nic poza malinką i czujnikiem temperatury nie zakupiłem. Może coś można zrobić prościej.

PS. Zanim doczytałem, chciałem używać wbudowanego czujnika w pico, ale wtedy miałbym temperaturę płytki, a nie w pomieszczeniu

[farmaceuta]

@krzysiek1985 to  lepszym i prostszym rozwiązaniem będzie esp32 i protokół esp-now...bez kabli i komplikacji, wystarczy że do maliny pi 4 prześlesz dane przez uart np. (jeśli pi 4 taki posiada) z esp32 który będzie odbiornikiem i będzie odbierał dane od innych ESP (nadajników) ESP ma dużo oszczędnych trybów uśpienia więc zasilanie z jednego ogniwa 18650 wystarczy na dlugo

[ethanak]

32 minuty temu, farmaceuta napisał:

lepszym i prostszym rozwiązaniem będzie esp32

Wystarczy ESP8266 - używam takiego zestawu (RPi <-> UART <-> ESP8266) w centralce i ESP8266 w satelitach.

Niestety, z przyczyn czysto technicznych nie da się zaimplementować esp-now bezpośrednio na malince.

[farmaceuta]

50 minut temu, ethanak napisał:

Niestety, z przyczyn czysto technicznych nie da się zaimplementować esp-now bezpośrednio na malince.

To akurat nie problem, jak sam wspomniałeś można przez uart puścić dane do maliny więc jedyny "problem" to dokupienie jednej sztuki ESP wiecej w roli odbiornika...i na cały dom plus podwórko zasięgu powinno styknąć...

[ethanak]

Przy okazji można jeszcze dwa druty podłączyć do GPIO0 i RST i można programować ESP w układzie bezpośrednio z malinki (albo wrzucic na ESP program wygenerowany na pececie).