Apprenez à établir une communication bidirectionnelle entre deux planches ESP32 à l’aide du protocole de communication ESP-Now. Tout d’abord, nous allons tester un exemple simple pour vous montrer comment implémenter une communication bidirectionnelle. Enfin, nous allons construire un projet plus complexe, dans lequel nous échangeons des lectures de capteurs entre les cartes et affichons les résultats sur un écran OLED.
Utiliser Arduino IDE? Suivez ce tutoriel à la place: ESP-Now Communication bidirectionnelle entre les cartes ESP32.
Table des matières
Dans ce tutoriel, nous couvrirons les sujets suivants:
Prérequis – Firmware Micropython
Pour suivre ce didacticiel, vous avez besoin de micro-firmware Micropython installé sur vos cartes ESP32 ou ESP8266. Vous avez également besoin d’un IDE pour écrire et télécharger le code sur votre carte. Nous suggérons d’utiliser Thonny IDE:
Nouveau sur Micropython? Consultez notre ebook: Micropython Programming avec ESP32 et ESP8266 Ebook (2e édition)
Présentation ESP-NOW
ESP-Now est un protocole de communication sans fil développé par ESPRESSIF qui permet à plusieurs cartes ESP32 ou ESP8266 d’échanger de petites quantités de données sans utiliser Wi-Fi ou Bluetooth. ESP-Now ne nécessite pas de connexion Wi-Fi complète (bien que le contrôleur Wi-Fi doit être activé), ce qui le rend idéal pour les applications de faible puissance et de faible latence comme les réseaux de capteurs, les télécommandes ou l’échange de données entre les cartes.
ESP-Now utilise un modèle de communication sans connexion, ce qui signifie que les périphériques peuvent envoyer et recevoir des données sans se connecter à un routeur ou configurer un point d’accès (contrairement à la communication HTTP entre les cartes). Il prend en charge unicast (envoyer des données à un appareil spécifique à l’aide de son adresse MAC) et diffuser (envoyer des données à tous les appareils à l’aide d’une adresse MAC diffusée).
Nouveau sur ESP-NOW? Lisez notre guide de démarrage pour Micropython en utilisant ESP-Now avec ESP32.
ESP32: obtenir une adresse MAC à la carte
Pour communiquer via ESP-Now, vous devez connaître l’adresse MAC de vos conseils. Pour obtenir l’adresse MAC de votre carte, vous pouvez copier le code suivant sur Thonny IDE et l’exécuter sur votre carte.
# Rui Santos & Sara Santos - Raspberryme.com
# Complete project details at https://Raspberryme.com/micropython-esp-now-esp32/
import network
wlan = network.WLAN(network.STA_IF)
wlan.active(True)
# Get MAC address (returns bytes)
mac = wlan.config('mac')
# Convert to human-readable format
mac_address=":".join('%02x' % b for b in mac)
print("MAC Address:", mac_address)Afficher le code brut
Après avoir exécuté le code, il devrait imprimer l’adresse MAC de la carte sur le shell.
Nous vous recommandons d’ajouter une étiquette ou un autocollant à chaque carte afin que vous puissiez les identifier clairement.
Nous utiliserons une carte ESP32 DOIT V1 et un Devkitc ESP32 S3, mais cela devrait être compatible avec n’importe quel modèle ESP32.
Communication bidirectionnelle ESP-Now entre deux planches ESP32 (Micropython)
Dans cette section, nous vous montrerons un exemple de base sur la façon d’échanger des messages simples entre les cartes ESP32 à l’aide du protocole de communication ESP-Now. Chaque carte est en même temps un récepteur et un émetteur, nous pouvons donc les appeler émetteurs-récepteurs.
Il existe deux bibliothèques Micropython ESP-Now incluses par défaut maintenant dans les versions de firmware MicropyThon actuelles: ESPNow et Aioespnow. Nous utiliserons l’Aioespnow, qui est la version asynchrone de la bibliothèque ESPNow.
Copiez le code suivant sur Thonny IDE.
# Rui Santos & Sara Santos - Raspberryme.com
# Complete project details at https://Raspberryme.com/micropython-esp32-esp-now-two-way/
import network
import aioespnow
import asyncio
import time
# Initialize Wi-Fi in station mode
sta = network.WLAN(network.STA_IF)
sta.active(True)
sta.config(channel=1) # Set channel explicitly if packets are not received
sta.disconnect()
# Initialize AIOESPNow
e = aioespnow.AIOESPNow()
try:
e.active(True)
except OSError as err:
print("Failed to initialize AIOESPNow:", err)
raise
# Peer MAC address (replace with the actual MAC of the other board)
peer_mac = b'\xff\xff\xff\xff\xff\xff' # Example peer MAC for unicast
# Add peer for unicast reliability
try:
e.add_peer(peer_mac)
except OSError as err:
print("Failed to add peer:", err)
raise
# Stats tracking
last_stats_time = time.time()
stats_interval = 10 # Print stats every 10 seconds
# Async function to send messages
async def send_messages(e, peer):
message_count = 0
while True:
try:
message = f"Hello from ESP32 #{message_count}"
if await e.asend(peer, message, sync=True):
print(f"Sent message: {message}")
else:
print("Failed to send message")
message_count += 1
await asyncio.sleep(1) # Send every 1 second
except OSError as err:
print("Send error:", err)
await asyncio.sleep(5)
# Async function to receive messages
async def receive_messages(e):
while True:
try:
async for mac, msg in e:
print(f"Received from {mac.hex()}: {msg.decode()}")
except OSError as err:
print("Receive error:", err)
await asyncio.sleep(5)
# Async function to print stats periodically
async def print_stats(e):
global last_stats_time
while True:
if time.time() - last_stats_time >= stats_interval:
stats = e.stats()
print("\nESP-NOW Statistics:")
print(f" Packets Sent: {stats[0]}")
print(f" Packets Delivered: {stats[1]}")
print(f" Packets Dropped (TX): {stats[2]}")
print(f" Packets Received: {stats[3]}")
print(f" Packets Dropped (RX): {stats[4]}")
last_stats_time = time.time()
await asyncio.sleep(1) # Check every second
# Main async function
async def main(e, peer):
# Run send, receive, and stats tasks concurrently
await asyncio.gather(send_messages(e, peer), receive_messages(e), print_stats(e))
# Run the async program
try:
asyncio.run(main(e, peer_mac))
except KeyboardInterrupt:
print("Stopping transceiver...")
e.active(False)
sta.active(False)
Afficher le code brut
Ce code définit la carte ESP32 en tant que récepteur ESP-Now et émetteur. Dans le code, vous devez insérer l’adresse MAC de la carte à laquelle vous souhaitez envoyer des données.
Par exemple, si l’adresse MAC de la carte du récepteur est de 30: AE: A4: F6: 7d: 4c. Vous devez le convertir au format octets comme suit:
- 30: ae: a4: f6: 7d: 4c »b ‘\ x30 \ xae \ xa4 \ xf6 \ x7d \ x4c’
# Peer MAC address
receiver_mac = b'\x30\xae\xa4\xf6\x7d\x4c'Téléchargez ce code sur les deux cartes, mais ajustez l’adresse MAC.
Comment fonctionne le code?
Pour mieux comprendre ce code, nous vous recommandons de vous familiariser avec la programmation asynchrone Micropython en premier. Vous pouvez lire le guide suivant:
Modules d’importation
Tout d’abord, importez les modules requis.
import network
import aioespnow
import asyncio
import timeInitialiser l’interface Wi-Fi
Ensuite, nous devons initialiser le Wi-Fi (même si nous ne l’utilisons pas) pour utiliser ESP-Now. Nous pouvons utiliser la station (STA_IF) ou le mode Point d’accès (AP_IF).
# Initialize Wi-Fi in station mode
sta = network.WLAN(network.STA_IF)
sta.active(True)
sta.config(channel=1) # Set channel
sta.disconnect()Initialiser ESP-now
Ensuite, nous pouvons initialiser ESP-Now. Tout d’abord, créez une instance aioespnow appelée e. Puis activez-le en utilisant la méthode active () et en passant la valeur vraie en tant qu’argument.
# Initialize AIOESPNow
e = aioespnow.AIOESPNow()
try:
e.active(True)
print("AIOESPNow initialized")
except OSError as err:
print("Failed to initialize AIOESPNow:", err)
raiseNous activons ESP-NOW à l’intérieur d’un essai et sauf des instructions afin que nous puissions prendre des erreurs en cas d’échec de l’initialisation.
Ajouter un pair ESP-now
Insérez l’adresse MAC par les pairs (l’adresse MAC de la carte à laquelle vous souhaitez envoyer des données).
# Receiver MAC address (the board you want to send data to)
peer_mac = b'\x68\xb6\xb3\x22\x9e\x60'Ensuite, nous pouvons ajouter l’adresse MAC du récepteur en tant que pair à l’aide de la méthode add_peer ().
try:
e.add_peer(peer_mac)
except OSError as err:
print("Failed to add peer:", err)
raiseFonction pour envoyer des messages ESP-Now
La fonction suivante envoie des messages ESP-Now au pair. Passez comme argument l’instance ESP-Now E et le pair. À titre d’exemple, nous enverrons un message Hello de ESP32 suivi d’un compteur. Nous envoyons un nouveau message à chaque seconde.
# Async function to send messages
async def send_messages(e, peer):
message_count = 0
while True:
try:
message = f"Hello from ESP32 #{message_count}"
if await e.asend(peer, message, sync=True):
print(f"Sent message: {message}")
else:
print("Failed to send message")
message_count += 1
await asyncio.sleep(1) # Send every 1 second
except OSError as err:
print("Send error:", err)
await asyncio.sleep(5)Fonction pour recevoir des messages ESP-Now
Nous devons également créer une fonction pour recevoir les messages ESP-Now. Lorsqu’il y a un nouveau message, nous l’avons imprimé dans le shell Micropython.
# Async function to receive messages
async def receive_messages(e):
while True:
try:
async for mac, msg in e:
print(f"Received from {mac.hex()}: {msg.decode()}")
except OSError as err:
print("Receive error:", err)
await asyncio.sleep(5)Imprimez les statistiques ESP-Now
Ensuite, nous créons un message à appeler plus tard dans le code, appelé print_stats () qui imprime les statistiques actuelles sur les paquets ESP-Now. Pour obtenir le nombre de paquets envoyés / reçus et perdus, nous pouvons appeler la méthode statistiques () sur l’objet ESP-Now E.
Cela renvoie un 5-Tuple contenant le nombre de paquets envoyés / reçus / perdus:
(tx_pkts, tx_responses, tx_failures, rx_packets, rx_dropped_packets)Il s’agit de la fonction asynchrone complète.
async def print_stats(e):
global last_stats_time
while True:
if time.time() - last_stats_time >= stats_interval:
stats = e.stats()
print("\nESP-NOW Statistics:")
print(f" Packets Sent: {stats[0]}")
print(f" Packets Delivered: {stats[1]}")
print(f" Packets Dropped (TX): {stats[2]}")
print(f" Packets Received: {stats[3]}")
print(f" Packets Dropped (RX): {stats[4]}")
last_stats_time = time.time()
await asyncio.sleep(1) # Check every secondFonction asynchrone principale
La ligne suivante définit une fonction asynchrone Main () qui prend l’objet ESP-Now E et l’adresse MAC par les pairs comme arguments.
async def main(e, peer):Ensuite, nous utilisons Asyncio.gather pour exécuter trois tâches asynchrones simultanément:
- Send_Messages (E, Peer): Envoie des données de capteur BME280 au pair;
- reçoile_messages (e): écoute les données entrantes du pair;
- print_stats (e): imprime périodiquement les statistiques ESP-Now.
await asyncio.gather(send_messages(e, peer), receive_messages(e), print_stats(e))Exécution du programme asynchrone
Enfin, les lignes suivantes exécuteront le programme de manière asynchrone.
# Run the async program
try:
asyncio.run(main(e, peer_mac))
except KeyboardInterrupt:
print("Stopping transceiver...")
e.active(False)
sta.active(False)Cette commande asyncio.run (main (e, peer_mac)) démarre la boucle d’événement MicropyThon Asyncio, exécutant la fonction Main ().
Cela va à l’intérieur d’essayer et d’exaucer les instructions pour attraper un clavier Interrup (arrêtant manuellement le code). Lorsque cela se produit, nous désactivons ESP-now et désactivons le Wi-Fi.
Exécution du code
Après avoir connecté la carte à votre ordinateur et établi une communication avec Thonny IDE, vous pouvez télécharger le code en tant que main.py sur la carte ou l’exécuter à l’aide du bouton d’exécution verte. Assurez-vous d’avoir inséré l’adresse MAC du récepteur dans le code.
Il commencera à imprimer des messages dans le shell. Il ne parviendra pas à envoyer le message tandis que l’autre carte n’est toujours pas prête.
Ouvrez une autre instance de Thonny IDE et établissez une communication avec l’autre conseil d’administration.
Activer deux instances de thonny ide
Accédez à des outils> Options et décochez l’option autorisez uniquement l’instance Thonny.
Copiez le même code, mais insérez l’autre adresse MAC de la carte. Après quelques secondes, il commencera à envoyer et à recevoir des messages ESP-Now.
Les captures d’écran suivantes montrent la coque micropython pour chacune de mes planches.
Vous pouvez voir qu’après avoir alimenté les deux planches, ils commenceront à échanger des messages entre eux.
ESP32 ESP-NOW Communication bidirectionnelle – Échange de lectures de capteurs et affichage sur OLED
Dans cette section, nous allons construire un projet avec une application du monde réel dans lequel les deux cartes ESP32 échangent des lectures de capteurs et afficher les données reçues sur un écran OLED.
Aperçu du projet
Le diagramme suivant montre un aperçu de haut niveau du projet que nous construirons.
- Dans ce projet, nous aurons deux planches ESP32. Chaque carte est connectée à un écran OLED et à un capteur BME280;
- Chaque planche obtient des lectures de température, d’humidité et de pression de ses capteurs correspondants;
- Chaque planche envoie ses lectures à l’autre tableau via ESP-Now;
- Lorsqu’une carte reçoit les lectures, elle les affiche sur l’écran OLED;
- Après avoir envoyé les lectures, la carte s’affiche sur l’OLED si le message a été livré avec succès;
- Chaque carte doit connaître l’autre adresse MAC pour envoyer le message.
Dans cet exemple, nous utilisons une communication bidirectionnelle entre deux planches, mais vous pouvez ajouter plus de planches à cette configuration, et avoir toutes les cartes communiquant entre elles.
Pièces requises
Pour ce tutoriel, vous avez besoin des parties suivantes:
Vous pouvez utiliser les liens précédents ou aller directement à makeradvisor.com/tools pour trouver toutes les pièces de vos projets au meilleur prix!
Diagramme schématique
Tirez un écran OLED et un capteur BME280 à chaque carte ESP32. Suivez le diagramme schématique suivant (ajustez si vous utilisez un modèle ESP32 avec une épingle différente).
Vous pouvez utiliser le tableau suivant comme référence lors du câblage du capteur BME280.
| BME280 | ESP32 |
| Vin | 3.3 V |
| GND | GND |
| SCL | GPIO 22 |
| SDA | GPIO 21 |
Vous pouvez également suivre la table suivante pour câbler l’écran OLED à l’ESP32.
| Affichage OLED | ESP32 |
| GND | GND |
| VDD / VCC | Vin |
| SCK / SCL | GPIO 22 |
| SDA | GPIO 21 |
Importation de bibliothèques
Les bibliothèques pour s’interfacer avec l’affichage OLED et les données BME280 ne font pas partie du package Micropython standard. Nous devons donc importer ces modules dans nos conseils.
Module SSD1306.PY Micropython
1. Copiez le code suivant dans un nouveau fichier sur Thonny IDE.
# MicroPython SSD1306 OLED driver, I2C and SPI interfaces created by Adafruit
import time
import framebuf
# register definitions
SET_CONTRAST = const(0x81)
SET_ENTIRE_ON = const(0xa4)
SET_NORM_INV = const(0xa6)
SET_DISP = const(0xae)
SET_MEM_ADDR = const(0x20)
SET_COL_ADDR = const(0x21)
SET_PAGE_ADDR = const(0x22)
SET_DISP_START_LINE = const(0x40)
SET_SEG_REMAP = const(0xa0)
SET_MUX_RATIO = const(0xa8)
SET_COM_OUT_DIR = const(0xc0)
SET_DISP_OFFSET = const(0xd3)
SET_COM_PIN_CFG = const(0xda)
SET_DISP_CLK_DIV = const(0xd5)
SET_PRECHARGE = const(0xd9)
SET_VCOM_DESEL = const(0xdb)
SET_CHARGE_PUMP = const(0x8d)
class SSD1306:
def __init__(self, width, height, external_vcc):
self.width = width
self.height = height
self.external_vcc = external_vcc
self.pages = self.height // 8
# Note the subclass must initialize self.framebuf to a framebuffer.
# This is necessary because the underlying data buffer is different
# between I2C and SPI implementations (I2C needs an extra byte).
self.poweron()
self.init_display()
def init_display(self):
for cmd in (
SET_DISP | 0x00, # off
# address setting
SET_MEM_ADDR, 0x00, # horizontal
# resolution and layout
SET_DISP_START_LINE | 0x00,
SET_SEG_REMAP | 0x01, # column addr 127 mapped to SEG0
SET_MUX_RATIO, self.height - 1,
SET_COM_OUT_DIR | 0x08, # scan from COM[N] to COM0
SET_DISP_OFFSET, 0x00,
SET_COM_PIN_CFG, 0x02 if self.height == 32 else 0x12,
# timing and driving scheme
SET_DISP_CLK_DIV, 0x80,
SET_PRECHARGE, 0x22 if self.external_vcc else 0xf1,
SET_VCOM_DESEL, 0x30, # 0.83*Vcc
# display
SET_CONTRAST, 0xff, # maximum
SET_ENTIRE_ON, # output follows RAM contents
SET_NORM_INV, # not inverted
# charge pump
SET_CHARGE_PUMP, 0x10 if self.external_vcc else 0x14,
SET_DISP | 0x01): # on
self.write_cmd(cmd)
self.fill(0)
self.show()
def poweroff(self):
self.write_cmd(SET_DISP | 0x00)
def contrast(self, contrast):
self.write_cmd(SET_CONTRAST)
self.write_cmd(contrast)
def invert(self, invert):
self.write_cmd(SET_NORM_INV | (invert & 1))
def show(self):
x0 = 0
x1 = self.width - 1
if self.width == 64:
# displays with width of 64 pixels are shifted by 32
x0 += 32
x1 += 32
self.write_cmd(SET_COL_ADDR)
self.write_cmd(x0)
self.write_cmd(x1)
self.write_cmd(SET_PAGE_ADDR)
self.write_cmd(0)
self.write_cmd(self.pages - 1)
self.write_framebuf()
def fill(self, col):
self.framebuf.fill(col)
def pixel(self, x, y, col):
self.framebuf.pixel(x, y, col)
def scroll(self, dx, dy):
self.framebuf.scroll(dx, dy)
def text(self, string, x, y, col=1):
self.framebuf.text(string, x, y, col)
class SSD1306_I2C(SSD1306):
def __init__(self, width, height, i2c, addr=0x3c, external_vcc=False):
self.i2c = i2c
self.addr = addr
self.temp = bytearray(2)
# Add an extra byte to the data buffer to hold an I2C data/command byte
# to use hardware-compatible I2C transactions. A memoryview of the
# buffer is used to mask this byte from the framebuffer operations
# (without a major memory hit as memoryview doesn't copy to a separate
# buffer).
self.buffer = bytearray(((height // 8) * width) + 1)
self.buffer[0] = 0x40 # Set first byte of data buffer to Co=0, D/C=1
self.framebuf = framebuf.FrameBuffer1(memoryview(self.buffer)[1:], width, height)
super().__init__(width, height, external_vcc)
def write_cmd(self, cmd):
self.temp[0] = 0x80 # Co=1, D/C#=0
self.temp[1] = cmd
self.i2c.writeto(self.addr, self.temp)
def write_framebuf(self):
# Blast out the frame buffer using a single I2C transaction to support
# hardware I2C interfaces.
self.i2c.writeto(self.addr, self.buffer)
def poweron(self):
pass
class SSD1306_SPI(SSD1306):
def __init__(self, width, height, spi, dc, res, cs, external_vcc=False):
self.rate = 10 * 1024 * 1024
dc.init(dc.OUT, value=0)
res.init(res.OUT, value=0)
cs.init(cs.OUT, value=1)
self.spi = spi
self.dc = dc
self.res = res
self.cs = cs
self.buffer = bytearray((height // 8) * width)
self.framebuf = framebuf.FrameBuffer1(self.buffer, width, height)
super().__init__(width, height, external_vcc)
def write_cmd(self, cmd):
self.spi.init(baudrate=self.rate, polarity=0, phase=0)
self.cs.high()
self.dc.low()
self.cs.low()
self.spi.write(bytearray([cmd]))
self.cs.high()
def write_framebuf(self):
self.spi.init(baudrate=self.rate, polarity=0, phase=0)
self.cs.high()
self.dc.high()
self.cs.low()
self.spi.write(self.buffer)
self.cs.high()
def poweron(self):
self.res.high()
time.sleep_ms(1)
self.res.low()
time.sleep_ms(10)
self.res.high()
Afficher le code brut
2. Allez dans le fichier> Enregistrer AS et sélectionner le périphérique Micropython.
3. Nommez le fichier ssd1306.py et cliquez sur OK pour enregistrer le fichier sur le système de fichiers ESP.
Et c’est tout. La bibliothèque a été téléchargée sur votre carte.
Module BME280.py Micropython
1. Copiez le code suivant dans un nouveau fichier sur Thonny IDE.
from machine import I2C
import time
# BME280 default address.
BME280_I2CADDR = 0x76
# Operating Modes
BME280_OSAMPLE_1 = 1
BME280_OSAMPLE_2 = 2
BME280_OSAMPLE_4 = 3
BME280_OSAMPLE_8 = 4
BME280_OSAMPLE_16 = 5
# BME280 Registers
BME280_REGISTER_DIG_T1 = 0x88 # Trimming parameter registers
BME280_REGISTER_DIG_T2 = 0x8A
BME280_REGISTER_DIG_T3 = 0x8C
BME280_REGISTER_DIG_P1 = 0x8E
BME280_REGISTER_DIG_P2 = 0x90
BME280_REGISTER_DIG_P3 = 0x92
BME280_REGISTER_DIG_P4 = 0x94
BME280_REGISTER_DIG_P5 = 0x96
BME280_REGISTER_DIG_P6 = 0x98
BME280_REGISTER_DIG_P7 = 0x9A
BME280_REGISTER_DIG_P8 = 0x9C
BME280_REGISTER_DIG_P9 = 0x9E
BME280_REGISTER_DIG_H1 = 0xA1
BME280_REGISTER_DIG_H2 = 0xE1
BME280_REGISTER_DIG_H3 = 0xE3
BME280_REGISTER_DIG_H4 = 0xE4
BME280_REGISTER_DIG_H5 = 0xE5
BME280_REGISTER_DIG_H6 = 0xE6
BME280_REGISTER_DIG_H7 = 0xE7
BME280_REGISTER_CHIPID = 0xD0
BME280_REGISTER_VERSION = 0xD1
BME280_REGISTER_SOFTRESET = 0xE0
BME280_REGISTER_CONTROL_HUM = 0xF2
BME280_REGISTER_CONTROL = 0xF4
BME280_REGISTER_CONFIG = 0xF5
BME280_REGISTER_PRESSURE_DATA = 0xF7
BME280_REGISTER_TEMP_DATA = 0xFA
BME280_REGISTER_HUMIDITY_DATA = 0xFD
class Device:
"""Class for communicating with an I2C device.
Allows reading and writing 8-bit, 16-bit, and byte array values to
registers on the device."""
def __init__(self, address, i2c):
"""Create an instance of the I2C device at the specified address using
the specified I2C interface object."""
self._address = address
self._i2c = i2c
def writeRaw8(self, value):
"""Write an 8-bit value on the bus (without register)."""
value = value & 0xFF
self._i2c.writeto(self._address, value)
def write8(self, register, value):
"""Write an 8-bit value to the specified register."""
b=bytearray(1)
b[0]=value & 0xFF
self._i2c.writeto_mem(self._address, register, b)
def write16(self, register, value):
"""Write a 16-bit value to the specified register."""
value = value & 0xFFFF
b=bytearray(2)
b[0]= value & 0xFF
b[1]= (value>>8) & 0xFF
self.i2c.writeto_mem(self._address, register, value)
def readRaw8(self):
"""Read an 8-bit value on the bus (without register)."""
return int.from_bytes(self._i2c.readfrom(self._address, 1),'little') & 0xFF
def readU8(self, register):
"""Read an unsigned byte from the specified register."""
return int.from_bytes(
self._i2c.readfrom_mem(self._address, register, 1),'little') & 0xFF
def readS8(self, register):
"""Read a signed byte from the specified register."""
result = self.readU8(register)
if result > 127:
result -= 256
return result
def readU16(self, register, little_endian=True):
"""Read an unsigned 16-bit value from the specified register, with the
specified endianness (default little endian, or least significant byte
first)."""
result = int.from_bytes(
self._i2c.readfrom_mem(self._address, register, 2),'little') & 0xFFFF
if not little_endian:
result = ((result << 8) & 0xFF00) + (result >> 8)
return result
def readS16(self, register, little_endian=True):
"""Read a signed 16-bit value from the specified register, with the
specified endianness (default little endian, or least significant byte
first)."""
result = self.readU16(register, little_endian)
if result > 32767:
result -= 65536
return result
def readU16LE(self, register):
"""Read an unsigned 16-bit value from the specified register, in little
endian byte order."""
return self.readU16(register, little_endian=True)
def readU16BE(self, register):
"""Read an unsigned 16-bit value from the specified register, in big
endian byte order."""
return self.readU16(register, little_endian=False)
def readS16LE(self, register):
"""Read a signed 16-bit value from the specified register, in little
endian byte order."""
return self.readS16(register, little_endian=True)
def readS16BE(self, register):
"""Read a signed 16-bit value from the specified register, in big
endian byte order."""
return self.readS16(register, little_endian=False)
class BME280:
def __init__(self, mode=BME280_OSAMPLE_1, address=BME280_I2CADDR, i2c=None,
**kwargs):
# Check that mode is valid.
if mode not in [BME280_OSAMPLE_1, BME280_OSAMPLE_2, BME280_OSAMPLE_4,
BME280_OSAMPLE_8, BME280_OSAMPLE_16]:
raise ValueError(
'Unexpected mode value {0}. Set mode to one of '
'BME280_ULTRALOWPOWER, BME280_STANDARD, BME280_HIGHRES, or '
'BME280_ULTRAHIGHRES'.format(mode))
self._mode = mode
# Create I2C device.
if i2c is None:
raise ValueError('An I2C object is required.')
self._device = Device(address, i2c)
# Load calibration values.
self._load_calibration()
self._device.write8(BME280_REGISTER_CONTROL, 0x3F)
self.t_fine = 0
def _load_calibration(self):
self.dig_T1 = self._device.readU16LE(BME280_REGISTER_DIG_T1)
self.dig_T2 = self._device.readS16LE(BME280_REGISTER_DIG_T2)
self.dig_T3 = self._device.readS16LE(BME280_REGISTER_DIG_T3)
self.dig_P1 = self._device.readU16LE(BME280_REGISTER_DIG_P1)
self.dig_P2 = self._device.readS16LE(BME280_REGISTER_DIG_P2)
self.dig_P3 = self._device.readS16LE(BME280_REGISTER_DIG_P3)
self.dig_P4 = self._device.readS16LE(BME280_REGISTER_DIG_P4)
self.dig_P5 = self._device.readS16LE(BME280_REGISTER_DIG_P5)
self.dig_P6 = self._device.readS16LE(BME280_REGISTER_DIG_P6)
self.dig_P7 = self._device.readS16LE(BME280_REGISTER_DIG_P7)
self.dig_P8 = self._device.readS16LE(BME280_REGISTER_DIG_P8)
self.dig_P9 = self._device.readS16LE(BME280_REGISTER_DIG_P9)
self.dig_H1 = self._device.readU8(BME280_REGISTER_DIG_H1)
self.dig_H2 = self._device.readS16LE(BME280_REGISTER_DIG_H2)
self.dig_H3 = self._device.readU8(BME280_REGISTER_DIG_H3)
self.dig_H6 = self._device.readS8(BME280_REGISTER_DIG_H7)
h4 = self._device.readS8(BME280_REGISTER_DIG_H4)
h4 = (h4 << 24) >> 20
self.dig_H4 = h4 | (self._device.readU8(BME280_REGISTER_DIG_H5) & 0x0F)
h5 = self._device.readS8(BME280_REGISTER_DIG_H6)
h5 = (h5 << 24) >> 20
self.dig_H5 = h5 | (
self._device.readU8(BME280_REGISTER_DIG_H5) >> 4 & 0x0F)
def read_raw_temp(self):
"""Reads the raw (uncompensated) temperature from the sensor."""
meas = self._mode
self._device.write8(BME280_REGISTER_CONTROL_HUM, meas)
meas = self._mode << 5 | self._mode << 2 | 1
self._device.write8(BME280_REGISTER_CONTROL, meas)
sleep_time = 1250 + 2300 * (1 << self._mode)
sleep_time = sleep_time + 2300 * (1 << self._mode) + 575
sleep_time = sleep_time + 2300 * (1 << self._mode) + 575
time.sleep_us(sleep_time) # Wait the required time
msb = self._device.readU8(BME280_REGISTER_TEMP_DATA)
lsb = self._device.readU8(BME280_REGISTER_TEMP_DATA + 1)
xlsb = self._device.readU8(BME280_REGISTER_TEMP_DATA + 2)
raw = ((msb << 16) | (lsb << 8) | xlsb) >> 4
return raw
def read_raw_pressure(self):
"""Reads the raw (uncompensated) pressure level from the sensor."""
"""Assumes that the temperature has already been read """
"""i.e. that enough delay has been provided"""
msb = self._device.readU8(BME280_REGISTER_PRESSURE_DATA)
lsb = self._device.readU8(BME280_REGISTER_PRESSURE_DATA + 1)
xlsb = self._device.readU8(BME280_REGISTER_PRESSURE_DATA + 2)
raw = ((msb << 16) | (lsb << 8) | xlsb) >> 4
return raw
def read_raw_humidity(self):
"""Assumes that the temperature has already been read """
"""i.e. that enough delay has been provided"""
msb = self._device.readU8(BME280_REGISTER_HUMIDITY_DATA)
lsb = self._device.readU8(BME280_REGISTER_HUMIDITY_DATA + 1)
raw = (msb << 8) | lsb
return raw
def read_temperature(self):
"""Get the compensated temperature in 0.01 of a degree celsius."""
adc = self.read_raw_temp()
var1 = ((adc >> 3) - (self.dig_T1 << 1)) * (self.dig_T2 >> 11)
var2 = ((
(((adc >> 4) - self.dig_T1) * ((adc >> 4) - self.dig_T1)) >> 12) *
self.dig_T3) >> 14
self.t_fine = var1 + var2
return (self.t_fine * 5 + 128) >> 8
def read_pressure(self):
"""Gets the compensated pressure in Pascals."""
adc = self.read_raw_pressure()
var1 = self.t_fine - 128000
var2 = var1 * var1 * self.dig_P6
var2 = var2 + ((var1 * self.dig_P5) << 17)
var2 = var2 + (self.dig_P4 << 35)
var1 = (((var1 * var1 * self.dig_P3) >> 8) +
((var1 * self.dig_P2) >> 12))
var1 = (((1 << 47) + var1) * self.dig_P1) >> 33
if var1 == 0:
return 0
p = 1048576 - adc
p = (((p << 31) - var2) * 3125) // var1
var1 = (self.dig_P9 * (p >> 13) * (p >> 13)) >> 25
var2 = (self.dig_P8 * p) >> 19
return ((p + var1 + var2) >> 8) + (self.dig_P7 << 4)
def read_humidity(self):
adc = self.read_raw_humidity()
# print 'Raw humidity = {0:d}'.format (adc)
h = self.t_fine - 76800
h = (((((adc << 14) - (self.dig_H4 << 20) - (self.dig_H5 * h)) +
16384) >> 15) * (((((((h * self.dig_H6) >> 10) * (((h *
self.dig_H3) >> 11) + 32768)) >> 10) + 2097152) *
self.dig_H2 + 8192) >> 14))
h = h - (((((h >> 15) * (h >> 15)) >> 7) * self.dig_H1) >> 4)
h = 0 if h < 0 else h
h = 419430400 if h > 419430400 else h
return h >> 12
@property
def temperature(self):
"Return the temperature in degrees."
t = self.read_temperature()
ti = t // 100
td = t - ti * 100
return "{}.{:02d}C".format(ti, td)
@property
def pressure(self):
"Return the temperature in hPa."
p = self.read_pressure() // 256
pi = p // 100
pd = p - pi * 100
return "{}.{:02d}hPa".format(pi, pd)
@property
def humidity(self):
"Return the humidity in percent."
h = self.read_humidity()
hi = h // 1024
hd = h * 100 // 1024 - hi * 100
return "{}.{:02d}%".format(hi, hd)
Afficher le code brut
2. Allez dans le fichier> Enregistrer sous…
3. Sélectionnez Enregistrer sur «Micropython Device»:
4. Nommez votre fichier comme bme280.py et appuyez sur le bouton OK:
Et c’est tout. La bibliothèque a été téléchargée sur votre carte.
ESP32 Échange de lectures de capteurs BME280 via ESP-NOW – Code micropython
Maintenant que vous avez tous les modules requis téléchargés sur vos planches, vous pouvez télécharger le code suivant sur chacun de vos planches.
Important: n’oubliez pas d’insérer l’adresse MAC du récepteur sur le code.
# Rui Santos & Sara Santos - Raspberryme.com
# Complete project details at https://Raspberryme.com/micropython-esp32-esp-now-two-way/
import network
import aioespnow
import asyncio
import time
import ujson
from machine import Pin, I2C
import BME280
import ssd1306
# Initialize I2C for BME280 and SSD1306
i2c = I2C(0, scl=Pin(22), sda=Pin(21))
# Initialize BME280 sensor
try:
bme = BME280.BME280(i2c=i2c, address=0x76)
print("BME280 initialized")
except Exception as err:
print("Failed to initialize BME280:", err)
raise
# Initialize SSD1306 OLED display
try:
display = ssd1306.SSD1306_I2C(128, 64, i2c, addr=0x3C)
print("SSD1306 initialized")
except Exception as err:
print("Failed to initialize SSD1306:", err)
raise
# Initialize Wi-Fi in station mode
sta = network.WLAN(network.STA_IF)
sta.active(True)
sta.config(channel=1) # set Wi-Fi channel for more stable communication
sta.disconnect()
print("Wi-Fi initialized")
# Initialize AIOESPNow
e = aioespnow.AIOESPNow()
try:
e.active(True)
print("AIOESPNow initialized")
except OSError as err:
print("Failed to initialize AIOESPNow:", err)
raise
# Receiver MAC address (the board you want to send data to)
peer_mac = b'\xff\xff\xff\xff\xff\xff'
# Add peer
try:
e.add_peer(peer_mac)
except OSError as err:
print("Failed to add peer:", err)
raise
# Variables to store readings and status
last_send_status = " "
incoming_readings = {'temp': 0.0, 'hum': 0.0, 'pres': 0.0}
# Function to get BME280 readings
def get_readings():
try:
temp = float(bme.temperature[:-1]) # Remove 'C'
hum = float(bme.humidity[:-1]) # Remove '%'
pres = float(bme.pressure[:-3]) # Remove 'hPa'
print("BME280 readings:", temp, hum, pres)
return temp, hum, pres
except Exception as err:
print("Error reading BME280:", err)
return 0.0, 0.0, 0.0
# Function to update OLED display
def update_display():
try:
display.fill(0)
display.text("INCOM. READINGS", 0, 0)
display.text("Temp: {:.1f} C".format(incoming_readings['temp']), 0, 15)
display.text("Hum: {:.1f} %".format(incoming_readings['hum']), 0, 25)
display.text("Pres: {:.1f} hPa".format(incoming_readings['pres']), 0, 35)
display.text(last_send_status, 0, 55)
display.show()
print("Display updated")
except Exception as err:
print("Error updating display:", err)
# Async function to send messages
async def send_messages(e, peer):
global last_send_status
while True:
try:
print("Sending data")
temp, hum, pres = get_readings()
# Create JSON string
data_dict = {"temp": temp, "hum": hum, "pres": pres}
json_str = ujson.dumps(data_dict)
data = json_str.encode('utf-8') # Convert to bytes
print("Sending JSON:", json_str)
if await e.asend(peer, data, sync=True):
print("Sent with success")
last_send_status = "Delivery Success :)"
else:
print("Send failed")
last_send_status = "Delivery Fail :("
update_display()
print("Sending task complete")
await asyncio.sleep(10) # Send every 10 seconds
except OSError as err:
print("Send error:", err)
last_send_status = "Delivery Fail :("
update_display()
await asyncio.sleep(0.1) # Shorter delay in case of error
# Async function to receive messages
async def receive_messages(e):
global incoming_readings
while True:
try:
print("Checking for messages")
async for mac, msg in e:
try:
# Decode bytes to string and parse JSON
json_str = msg.decode('utf-8')
data_dict = ujson.loads(json_str)
temp = data_dict['temp']
hum = data_dict['hum']
pres = data_dict['pres']
incoming_readings['temp'] = temp
incoming_readings['hum'] = hum
incoming_readings['pres'] = pres
print("\nINCOMING READINGS")
print("Temperature: {:.1f} ºC".format(temp))
print("Humidity: {:.1f} %".format(hum))
print("Pressure: {:.1f} hPa".format(pres))
update_display()
except (ValueError, KeyError) as err:
print("Error parsing JSON:", err)
await asyncio.sleep(0.01) # Yield if no received messages
except OSError as err:
print("Receive error:", err)
await asyncio.sleep(0.1) # Shorter delay in case of error
# Main async function
async def main(e, peer):
print("Starting main loop")
await asyncio.gather(send_messages(e, peer), receive_messages(e))
# Run the async program
try:
print("Starting transceiver...")
asyncio.run(main(e, peer_mac))
except KeyboardInterrupt:
print("Stopping transceiver...")
e.active(False)
sta.active(False)
except Exception as err:
print("Main loop error:", err)
finally:
print("Cleaning up...")
e.active(False)
sta.active(False)
Afficher le code brut
Comment fonctionne le code?
Voyons rapidement comment fonctionne le code. Alternativement, vous pouvez passer à la section de démonstration.
Importation de bibliothèques
Nous commençons par importer les modules requis.
import network
import aioespnow
import asyncio
import time
import ujson
from machine import Pin, I2C
import BME280
import ssd1306Communication I2C
Initialisez une communication I2C sur GPIOS 22 et 21. Ajustez si vous utilisez différentes broches.
i2c = I2C(0, scl=Pin(22), sda=Pin(21))BME280
Initialisez le capteur BME280.
# Initialize BME280 sensor
try:
bme = BME280.BME280(i2c=i2c, address=0x76)
print("BME280 initialized")
except Exception as err:
print("Failed to initialize BME280:", err)
raiseAffichage OLED
Initialiser l’affichage OLED.
# Initialize SSD1306 OLED display
try:
display = ssd1306.SSD1306_I2C(128, 64, i2c, addr=0x3C)
print("SSD1306 initialized")
except Exception as err:
print("Failed to initialize SSD1306:", err)
raiseInitialiser l’interface Wi-Fi
Ensuite, nous devons initialiser le Wi-Fi (même si nous ne l’utilisons pas) pour utiliser ESP-Now. Nous pouvons utiliser la station (STA_IF) ou le mode Point d’accès (AP_IF).
# Initialize Wi-Fi in station mode
sta = network.WLAN(network.STA_IF)
sta.active(True)
sta.config(channel=1) # Set channel
sta.disconnect()Initialiser ESP-now
Ensuite, nous pouvons initialiser ESP-Now. Tout d’abord, créez une instance aioespnow appelée e. Puis activez-le en utilisant la méthode active () et en passant la valeur vraie en tant qu’argument.
# Initialize AIOESPNow
e = aioespnow.AIOESPNow()
try:
e.active(True)
print("AIOESPNow initialized")
except OSError as err:
print("Failed to initialize AIOESPNow:", err)
raiseNous activons ESP-NOW à l’intérieur d’un essai et sauf des instructions afin que nous puissions prendre des erreurs en cas d’échec de l’initialisation.
Ajouter un pair ESP-now
Insérez l’adresse MAC par les pairs (l’adresse MAC de la carte à laquelle vous souhaitez envoyer des données).
# Receiver MAC address (the board you want to send data to)
peer_mac = b'\x68\xb6\xb3\x22\x9e\x60'Ensuite, nous pouvons ajouter l’adresse MAC du récepteur en tant que pair à l’aide de la méthode add_peer ().
try:
e.add_peer(peer_mac)
except OSError as err:
print("Failed to add peer:", err)
raiseVariables
Créez des variables pour stocker les lectures des capteurs entrants et l’état du processus d’envoi. Nous économisons les lectures de capteurs entrants dans un dictionnaire.
last_send_status = " "
incoming_readings = {'temp': 0.0, 'hum': 0.0, 'pres': 0.0}Obtenez des lectures de capteurs BME280
La fonction get_readings () renvoie les lectures du capteur BME280 dans cet ordre, la température, l’humidité et la pression. Les fonctions de la bibliothèque BME280 renvoient les résultats avec le caractère de l’unité, c’est pourquoi nous supprimons les caractères à la fin des lectures.
Dans le cas où il y a une erreur en lisant le capteur, il renverra 0,0 pour toutes les lectures.
# Function to get BME280 readings
def get_readings():
try:
temp = float(bme.temperature[:-1]) # Remove 'C'
hum = float(bme.humidity[:-1]) # Remove '%'
pres = float(bme.pressure[:-3]) # Remove 'hPa'
print("BME280 readings:", temp, hum, pres)
return temp, hum, pres
except Exception as err:
print("Error reading BME280:", err)
return 0.0, 0.0, 0.0Mettre à jour l’écran OLED
La fonction suivante met à jour l’affichage avec les lectures actuelles des capteurs entrants. Il affiche également le dernier statut d’envoi.
# Function to update OLED display
def update_display():
try:
display.fill(0)
display.text("INCOM. READINGS", 0, 0)
display.text("Temp: {:.1f} C".format(incoming_readings['temp']), 0, 15)
display.text("Hum: {:.1f} %".format(incoming_readings['hum']), 0, 25)
display.text("Pres: {:.1f} hPa".format(incoming_readings['pres']), 0, 35)
display.text(last_send_status, 0, 55)
display.show()
print("Display updated")
except Exception as err:
print("Error updating display:", err)Pour en savoir plus sur l’utilisation de l’affichage OLED avec l’ESP32 programmé avec Micropython, vérifiez notre guide: Micropython: Affichage OLED avec ESP32 et ESP8266.
La fonction Send_Message (E, Peer) envoie un message au pair spécifié à l’aide d’une instance ESPNow e.
# Async function to send messages
async def send_messages(e, peer):
global last_send_status
while True:
try:
print("Sending data")
temp, hum, pres = get_readings()
# Create JSON string
data_dict = {"temp": temp, "hum": hum, "pres": pres}
json_str = ujson.dumps(data_dict)
data = json_str.encode('utf-8') # Convert to bytes
print("Sending JSON:", json_str)
if await e.asend(peer, data, sync=True):
print("Sent with success")
last_send_status = "Delivery Success :)"
else:
print("Send failed")
last_send_status = "Delivery Fail :("
update_display()
print("Sending task complete")
await asyncio.sleep(10) # Send every 10 seconds
except OSError as err:
print("Send error:", err)
last_send_status = "Delivery Fail :("
update_display()
await asyncio.sleep(0.1) # Shorter delay in case of errorDans cette fonction, nous commençons par appeler get_readings () pour obtenir les lectures de température, d’humidité et de pression actuelles. Ensuite, nous mettons à jour le dictionnaire avec les lectures actuelles. Nous le convertissons en une chaîne JSON puis en octets.
temp, hum, pres = get_readings()
# Create JSON string
data_dict = {"temp": temp, "hum": hum, "pres": pres}
json_str = ujson.dumps(data_dict)
data = json_str.encode('utf-8') # Convert to bytesEnsuite, nous utilisons la même procédure que nous avons décrite précédemment pour envoyer des données. Au cours de ce processus, nous mettons également à jour l’affichage avec le résultat de l’envoi des données.
if await e.asend(peer, data, sync=True):
print("Sent with success")
last_send_status = "Delivery Success :)"
else:
print("Send failed")
last_send_status = "Delivery Fail :("
update_display()
print("Sending task complete")
await asyncio.sleep(10) # Send every 10 seconds
except OSError as err:
print("Send error:", err)
last_send_status = "Delivery Fail :("
update_display()
await asyncio.sleep(0.1) # Shorter delay in case of errorRecevoir des données entrantes
La fonction reçoit_messages () recevra les données de l’autre carte.
# Async function to receive messages
async def receive_messages(e):
global incoming_readings
while True:
try:
print("Checking for messages")
async for mac, msg in e:
try:
# Decode bytes to string and parse JSON
json_str = msg.decode('utf-8')
data_dict = ujson.loads(json_str)
temp = data_dict['temp']
hum = data_dict['hum']
pres = data_dict['pres']
incoming_readings['temp'] = temp
incoming_readings['hum'] = hum
incoming_readings['pres'] = pres
print("\nINCOMING READINGS")
print("Temperature: {:.1f} ºC".format(temp))
print("Humidity: {:.1f} %".format(hum))
print("Pressure: {:.1f} hPa".format(pres))
update_display()
except (ValueError, KeyError) as err:
print("Error parsing JSON:", err)
await asyncio.sleep(0.01) # Yield if no received messages
except OSError as err:
print("Receive error:", err)
await asyncio.sleep(0.1) # Shorter delay in case of errorLorsque nous recevons les données, nous la convertissons en une chaîne JSON. Ensuite, nous mettons à jour le dictionnaire entrant_readings avec les données reçues.
Enfin, nous imprimons les lectures dans le moniteur série et appelons également la fonction Update_Display () pour mettre à jour l’écran avec les dernières lectures de capteurs.
Création de la fonction de boucle asynchrone et exécution du programme
Enfin, nous créons la fonction de boucle asynchrone principale et exécutons le programme de manière asynchrone comme nous l’avons fait dans l’exemple précédent.
# Run the async program
try:
print("Starting transceiver...")
asyncio.run(main(e, peer_mac))
except KeyboardInterrupt:
print("Stopping transceiver...")
e.active(False)
sta.active(False)
except Exception as err:
print("Main loop error:", err)
finally:
print("Cleaning up...")
e.active(False)
sta.active(False)Téléchargement du code sur vos planches
Remarque importante: le simple fait d’exécuter le fichier avec Thonny ne le copie pas en permanence au système de fichiers de la carte. Cela signifie que si vous le débranchez à partir de votre ordinateur et appliquez de l’alimentation à la carte, rien ne se passera car il n’a pas de fichier Python enregistré sur son système de fichiers. La fonction Thonny IDE Run est utile pour tester le code, mais si vous souhaitez le télécharger en permanence sur votre carte, vous devez créer et enregistrer un fichier sur le système de fichiers de la carte.
Pour exécuter le code sur vos planches sans être connecté à votre ordinateur, vous devez le télécharger sur le système de fichiers de la carte avec le nom main.py.
Allez dans Fichier> Enregistrer sous … Appareil micropython.
Appelez ce fichier main.py et enregistrez-le sur la carte.
Démonstration
Après avoir téléchargé le code sur les deux cartes, vous devriez voir l’OLED afficher les lectures du capteur de l’autre carte, ainsi qu’un message «Succès de livraison».
Emballage
Nous espérons que vous avez trouvé ce guide utile. Le protocole de communication sans fil ESP-Now est l’une des méthodes les plus faciles à communiquer entre les cartes ESP32 à distance sans avoir besoin d’un routeur Wi-Fi.
Dans ce tutoriel, vous avez appris à établir une communication bidirectionnelle entre les conseils. Vous pouvez facilement ajouter plus de cartes à votre configuration en ajoutant plus de pairs pour créer un réseau de planches ESP32 qui communiquent entre elles.
Si vous ne faites que commencer avec ESP-Now, nous vous recommandons de commencer avec notre guide ESP-Now Getting pour le programme ESP32 avec Micropython.
Vous voulez en savoir plus sur Micropython avec l’ESP32? Consultez nos ressources:
Merci d’avoir lu.
Plongez dans l’histoire de Raspberry Pi avec cette vidéo :
-
ELEGOO 3PCS Carte de Développement ESP-32 Type-C, 2,4 GHz WiFi + Bluetooth -
ELEGOO 2 Pièces ESP-32 Carte de Développement avec Port USB Micro