基于 Adafruit CircuitPython 10.x 及 ESP32-C3 平台的调试经验总结
CircuitPython 的 ESP-NOW 更简洁，但必须严格遵守 信道锁定 + Peer 对象 + 发送传入 peer 的范式，切勿生搬 MicroPython 的用法。

## 一、核心差异：MicroPython vs CircuitPython

CircuitPython 的 espnow 模块 API 与 MicroPython **不兼容**，切勿混用。

| 操作 | MicroPython | CircuitPython |

|------|-------------|---------------|

| **激活** | `esp.active(True)` | **无需调用**，创建实例即激活 |

| **添加对等端** | `esp.add_peer(mac)` | 创建 `Peer` 对象并加入 `esp.peers` 列表 |

| **发送** | `esp.send(message)` 或 `esp.send(mac, message)` | `esp.send(message, peer)`（**必须传入 peer 对象**） |

| **广播地址** | 可直接使用 | **必须作为 Peer 添加** |

## 二、正确的初始化流程

```python

import wifi

import espnow

# 1. 激活 WiFi 硬件（无需连接 AP）

wifi.radio.enabled = True

# 2. 获取信道（优先使用已连接 AP 的信道）

channel = 6

if wifi.radio.ap_info is not None:

    channel = wifi.radio.ap_info.channel

# 3. 锁定信道（ESP32-C3 必须，通过临时 AP）

wifi.radio.start_ap("temp", "", channel=channel, max_connections=0)

wifi.radio.stop_ap()

# 4. 创建 ESP-NOW 实例

e = espnow.ESPNow()

# 5. 添加广播对等端（广播地址必须显式添加）

broadcast_mac = b'\xff\xff\xff\xff\xff\xff'

peer = espnow.Peer(mac=broadcast_mac, channel=channel)

e.peers.append(peer)

# 6. 发送消息（必须传入 peer）

message = b'\x01\x00\x01\x02\x03\x07'   # 自定义报文

e.send(message, peer)

```

## 三、关键注意事项

### 3.1 信道锁定是必需的

即使已获取到当前信道，也必须通过 `start_ap() + stop_ap()` 的"空 AP"方式让 ESP-NOW 锁定到该信道。

若不锁定，发送可能成功但接收方收不到数据，或直接报错 `0x3069`。

```python

# 正确做法

wifi.radio.start_ap("_temp_espnow", "", channel=channel, max_connections=0)

wifi.radio.stop_ap()

```

### 3.2 广播地址必须作为 Peer 添加

即使发送目标是广播地址 `FF:FF:FF:FF:FF:FF`，也必须创建 Peer 对象并加入 `peers` 列表。

否则会返回 `ESP-NOW error 0x3069`（底层错误：对等端未找到）。

```python

# 必须创建 Peer 对象

peer = espnow.Peer(b'\xff\xff\xff\xff\xff\xff', channel=channel)

e.peers.append(peer)

```

### 3.3 发送时需传入 peer 参数

**正确**：`e.send(message, peer)`

**错误**：`e.send(message)` 或 `e.send(peer, message)`（参数顺序反了）

### 3.4 避免重复初始化

使用 `initialized` 标志位，确保 `init()` 只执行一次。

CircuitPython 的 `espnow.ESPNow()` 实例创建即激活，再次调用 `active(True)` 会抛出 `Already running` 异常（因为该方法不存在）。

```python

class ESPNowManager:

    def __init__(self):

        self.initialized = False

      def init(self):

        if self.initialized:

            return True

        # ... 初始化逻辑 ...

        self.initialized = True

        return True

```

### 3.5 信道一致性

ESP-NOW 通信双方必须在同一信道。

若设备同时连接 WiFi 路由器，ESP-NOW 的信道必须与路由器信道一致，否则 WiFi 连接会强制切换信道，导致 ESP-NOW 失效。

## 四、常见错误及解决方法

| 错误信息 | 原因 | 解决方案 |

|----------|------|----------|

| `ESP-NOW error 0x3069` | 未添加对等端，或信道不匹配 | 按正确流程添加 Peer，锁定信道 |

| `object with buffer protocol required` | `send()` 参数错误（传入了 Peer 而非 message） | 使用 `send(message, peer)` |

| `'ESPNow' object has no attribute 'active'` | 混用 MicroPython API | 删除 `.active(True)` 调用 |

| `Already running` | 尝试重复激活（或之前存在残留实例） | 检查初始化逻辑，确保只调用一次 |

| `ssid length must be 1-32` | `start_ap()` 的 SSID 为空字符串 | 提供临时 SSID，如 `"temp"` |

## 五、调试建议

### 5.1 先用独立脚本验证

编写最小测试脚本（只包含 WiFi 激活 + ESP-NOW 初始化 + 发送），排除调度器和其他模块干扰。

### 5.2 打印关键状态

- 本机 MAC 地址

- 使用的信道

- 是否成功添加 Peer

- 每次发送的结果

### 5.3 接收端测试

在另一台设备上运行接收示例（同样需要正确锁定信道），确认广播能被收到。

### 5.4 减少日志输出

稳定运行后，可将详细日志的触发间隔调大（如每 50 或 100 次打印一次），避免串口拥塞。

## 六、完整示例参考

### 6.1 最小发送示例

```python

# test_espnow.py - 最小发送示例

import wifi, espnow

# 信道锁定

channel = 6   # 或从 wifi.radio.ap_info.channel 获取

wifi.radio.enabled = True

wifi.radio.start_ap("test", "", channel=channel, max_connections=0)

wifi.radio.stop_ap()

# 初始化

e = espnow.ESPNow()

peer = espnow.Peer(b'\xff\xff\xff\xff\xff\xff', channel=channel)

e.peers.append(peer)

# 发送

msg = b'\x01\x00\x01\x02\x03\x07'

e.send(msg, peer)

print("Sent")

```

### 6.2 完整类示例

# test_espnow.py - ESP-NOW 测试脚本 (CircuitPython 版本，已修正)

import espnow
import wifi

print("[TEST] 激活WiFi硬件...")
wifi.radio.enabled = True

mac = wifi.radio.mac_address
mac_str = ":".join("{:02x}".format(b) for b in mac)
print(f"[TEST] 本机MAC: {mac_str}")

# 关键步骤1：获取并锁定信道
channel = 6
if wifi.radio.ap_info is not None:
    channel = wifi.radio.ap_info.channel
    print(f"[TEST] 检测到已连接WiFi，使用其信道: {channel}")
else:
    print(f"[TEST] 未连接WiFi，使用默认信道: {channel}")

print("[TEST] 为ESP32-C3锁定ESP-NOW信道...")
wifi.radio.start_ap("espnow_temp", "", channel=channel, max_connections=0)
wifi.radio.stop_ap()

print("[TEST] 初始化ESP-NOW...")
e = espnow.ESPNow()

# 关键步骤2：添加广播对等端
broadcast_mac = b'\xff\xff\xff\xff\xff\xff'
peer = espnow.Peer(mac=broadcast_mac, channel=channel)
e.peers.append(peer)
print(f"[TEST] 已添加广播对等端: {broadcast_mac.hex()}, 信道: {channel}")

# 构造测试消息
test_message = b'\x01\x00\x01\x02\x03\x07'
print(f"[TEST] 发送消息 (hex): {test_message.hex()}")

# 关键步骤3：发送时传入peer参数
try:
    e.send(test_message, peer)
    print("[TEST] 消息发送成功!")
except Exception as err:
    print(f"[TEST] 消息发送失败: {err}")

# espnow_message.py - ESP-NOW 广播消息发送（CircuitPython版本）
# 支持四种模拟数据：
# - F1线圈状态 (类型0x01)
# - F2离散量输入 (类型0x02)
# - F3保持寄存器 (类型0x03)
# - F4输入寄存器 (类型0x04)
# 报文格式：
# - F1/F2: 1字节类型 + 4字节数据 + 1字节校验和
# - F3/F4: 1字节类型 + 32字节数据 + 1字节校验和

import espnow
import time
import random
import struct
import wifi

# ESP-NOW 配置
BROADCAST_INTERVAL = 1000      # 发送间隔(ms)
BROADCAST_MAC = b'\xff\xff\xff\xff\xff\xff'   # 广播地址

# 数据类型定义
TYPE_F1 = 0x01  # F1线圈状态 (Coils)
TYPE_F2 = 0x02  # F2离散量输入 (Discrete Inputs)
TYPE_F3 = 0x03  # F3保持寄存器 (Holding Registers - 4xxxx区)
TYPE_F4 = 0x04  # F4输入寄存器 (Input Registers - 3xxxx区)


class ESPNowMessage:
    """ESP-NOW 消息发送器（CircuitPython版本）"""

    def __init__(self):
        self.espnow = None
        self.peer = None          # 广播对等端对象
        self.msg_count = 0
        self.msg_count_f1 = 0
        self.msg_count_f2 = 0
        self.msg_count_f3 = 0
        self.msg_count_f4 = 0
        self.initialized = False
        self.enabled = False
        self.wifi_ready = False
        self.current_type = TYPE_F1
        self.start_delay_ms = 5000
        self.start_time_ms = 0
        self.mac_address = None

    def init(self):
        """初始化ESP-NOW（CircuitPython规范）"""
        if self.initialized:
            return True

        try:
            print('[ESP-NOW] 初始化...')

            # 1. 激活WiFi硬件
            if not wifi.radio.enabled:
                print('[ESP-NOW] 启用WiFi硬件...')
                wifi.radio.enabled = True
            print('[ESP-NOW] WiFi状态: 已激活')

            # 2. 确定信道（优先使用已连接AP的信道）
            channel = 6
            if wifi.radio.ap_info is not None:
                channel = wifi.radio.ap_info.channel
                print(f'[ESP-NOW] 当前WiFi信道: {channel}')
            else:
                print(f'[ESP-NOW] 未连接WiFi，使用默认信道: {channel}')

            # 3. 锁定信道（ESP32-C3必须通过临时AP设置）
            print('[ESP-NOW] 锁定ESP-NOW信道...')
            wifi.radio.start_ap("espnow_temp", "", channel=channel, max_connections=0)
            wifi.radio.stop_ap()

            # 4. 创建ESP-NOW实例并添加广播对等端
            self.espnow = espnow.ESPNow()
            self.peer = espnow.Peer(mac=BROADCAST_MAC, channel=channel)
            self.espnow.peers.append(self.peer)
            print(f'[ESP-NOW] 已添加广播对等端: {BROADCAST_MAC.hex()}, 信道: {channel}')

            # 5. 获取本机MAC地址
            mac_bytes = wifi.radio.mac_address
            self.mac_address = ':'.join(f'{b:02x}' for b in mac_bytes)
            print(f'[ESP-NOW] 本机MAC: {self.mac_address}')

            self.initialized = True
            print('[ESP-NOW] 初始化完成 - 广播模式')
            print('[ESP-NOW] 报文格式:')
            print('[ESP-NOW]   - F1/F2: 1字节类型 + 4字节数据 + 1字节校验和')
            print('[ESP-NOW]   - F3/F4: 1字节类型 + 32字节数据 + 1字节校验和')
            print('[ESP-NOW] 支持四种数据类型:')
            print('[ESP-NOW]   - 类型0x01: F1线圈状态 (Coils)')
            print('[ESP-NOW]   - 类型0x02: F2离散量输入 (Discrete Inputs)')
            print('[ESP-NOW]   - 类型0x03: F3保持寄存器 (Holding Registers - 4xxxx区)')
            print('[ESP-NOW]   - 类型0x04: F4输入寄存器 (Input Registers - 3xxxx区)')
            return True

        except Exception as e:
            print(f'[ESP-NOW] 初始化失败: {e}')
            self.espnow = None
            self.peer = None
            self.initialized = False
            return False

    def generate_random_32bit(self):
        """生成32位随机数（用于F1/F2）"""
        return random.getrandbits(32)

    def generate_registers(self):
        """生成32字节寄存器数据（16个16位寄存器）"""
        data = bytearray()
        for _ in range(16):
            reg_value = random.getrandbits(16)
            data.extend(struct.pack('>H', reg_value))
        return bytes(data)

    def calculate_checksum(self, data_bytes):
        """计算校验和（累加取低8位）"""
        return sum(data_bytes) & 0xFF

    def create_message(self, data_type=TYPE_F2):
        """创建消息包"""
        type_byte = bytes([data_type])

        if data_type in (TYPE_F3, TYPE_F4):
            data_bytes = self.generate_registers()
            value = data_bytes
        else:
            value = self.generate_random_32bit()
            data_bytes = struct.pack('>I', value)

        checksum = self.calculate_checksum(type_byte + data_bytes)
        message = type_byte + data_bytes + bytes([checksum])

        label_map = {TYPE_F1: "F1", TYPE_F2: "F2", TYPE_F3: "F3", TYPE_F4: "F4"}
        return message, value, checksum, label_map[data_type]

    def send_broadcast(self):
        """发送广播消息（循环F1→F2→F3→F4）"""
        try:
            if not self.initialized:
                if not self.init():
                    return False

            # 生成消息
            message, value, checksum, type_label = self.create_message(self.current_type)

            # 发送
            try:
                print(f'[ESP-NOW] 发送目标: {BROADCAST_MAC.hex()}')
                print(f'[ESP-NOW] 消息长度: {len(message)} 字节')
                print(f'[ESP-NOW] 消息内容: {message.hex()}')
                self.espnow.send(message, self.peer)   # 必须传入peer
                print(f'[ESP-NOW] 发送成功 #{self.msg_count + 1}')
            except Exception as e:
                print(f'[ESP-NOW] 发送失败: {e}')
                return False

            self.msg_count += 1

            # 计数
            if self.current_type == TYPE_F1:
                self.msg_count_f1 += 1
                type_count = self.msg_count_f1
            elif self.current_type == TYPE_F2:
                self.msg_count_f2 += 1
                type_count = self.msg_count_f2
            elif self.current_type == TYPE_F3:
                self.msg_count_f3 += 1
                type_count = self.msg_count_f3
            else:
                self.msg_count_f4 += 1
                type_count = self.msg_count_f4

            # 每10次打印详细日志
            if type_count % 10 == 0:
                hex_str = ' '.join(f'{b:02x}' for b in message)
                print('-' * 50)
                print(f'[ESP-NOW] {type_label} #{type_count} (总 #{self.msg_count})')

                if self.current_type in (TYPE_F3, TYPE_F4):
                    reg_label = "F3" if self.current_type == TYPE_F3 else "F4"
                    for i in range(16):
                        reg_value = struct.unpack('>H', value[2*i:2*i+2])[0]
                        print(f'[ESP-NOW] {reg_label}[{i}] 0x{reg_value:04X} ({reg_value})')
                else:
                    bits_str = ''.join(f'{b:08b}' for b in message[1:5])
                    print(f'[ESP-NOW] 数据 0x{value:08X}')
                    print(f'[ESP-NOW] 二进制 {bits_str}')

                print(f'[ESP-NOW] 报文 {hex_str}')
                print(f'[ESP-NOW] MAC {self.mac_address}')
                print('-' * 50)

            # 切换数据类型
            if self.current_type == TYPE_F1:
                self.current_type = TYPE_F2
            elif self.current_type == TYPE_F2:
                self.current_type = TYPE_F3
            elif self.current_type == TYPE_F3:
                self.current_type = TYPE_F4
            else:
                self.current_type = TYPE_F1

            return True

        except Exception as e:
            print(f'[ESP-NOW] 发送异常: {e}')
            return False

    def _check_start_condition(self):
        """检查是否满足发送条件（延迟启动）"""
        try:
            if not wifi.radio.enabled:
                print('[ESP-NOW] WiFi硬件未激活')
                return False
            if not self.wifi_ready:
                self.wifi_ready = True
                print('[ESP-NOW] WiFi硬件已激活，准备发送')
        except Exception as e:
            print(f'[ESP-NOW] WiFi检查失败: {e}')
            return False

        if self.start_time_ms == 0:
            self.start_time_ms = time.monotonic() * 1000

        if (time.monotonic() * 1000 - self.start_time_ms) >= self.start_delay_ms:
            if not self.enabled:
                self.enabled = True
                print('[ESP-NOW] 延迟启动完成，开始发送报文')
            return True
        return False

    def task_callback(self):
        """调度器回调函数"""
        if not self._check_start_condition():
            return True
        return self.send_broadcast()


def create_espnow_task(scheduler):
    """创建ESP-NOW发送任务"""
    sender = ESPNowMessage()
    sender.init()   # 预初始化，避免运行时重复初始化

    task = scheduler.add_task(
        name='espnow',
        interval_ms=BROADCAST_INTERVAL,
        callback=sender.task_callback,
        enabled=True
    )
    print(f'[ESP-NOW] 任务已注册 (间隔: {BROADCAST_INTERVAL}ms)')
    return task


def register_tasks(scheduler):
    """注册ESP-NOW任务到调度器"""
    create_espnow_task(scheduler)

## 七、参考资料

- **Adafruit Learn Guide**: ESP-NOW in CircuitPython

- **CircuitPython 文档**: espnow 模块

## 八、总结

**CircuitPython 的 ESP-NOW 更简洁，但必须严格遵守 信道锁定 + Peer 对象 + 发送传入 peer 的范式，切勿生搬 MicroPython 的用法。**

| 要点 | 说明 |

|------|------|

| **信道锁定** | ESP32-C3 上使用 ESP-NOW 的**关键步骤**，必须执行 `start_ap/stop_ap` |

| **Peer 对象** | 必须显式创建并添加到 `peers` 列表，不能直接使用 MAC 地址 |

| **发送参数** | `send(message, peer)`，**必须传入 peer 对象** |

| **初始化控制** | 使用标志位避免重复初始化，防止 "Already running" 错误 |

| **WiFi 连接** | ESP-NOW **不需要 WiFi 连接**，但需要 WiFi 硬件激活 |