import machine from machine import I2C, Pin import utime as time import uselect import usys # Pin & Address Constants RSTB_PIN_NUM = 35 # GPIO35, Pin 39 PKT_PIN_NUM = 34 # GPIO34, Pin 38 (用作接收包中断指示引脚) SCL_PIN_NUM = 24 # GPIO24, Pin 67 SDA_PIN_NUM = 23 # GPIO23, Pin 66 I2C_SLAVE_ADDR = 0x38 # Target 7-bit slave address REG_MODE = 0x07 REG_PKT_FLAG = 0x30 REG_CLEAR_FIFO = 0x34 CMD_READ_FIFO = 0x32 # I2C 模式下读 FIFO 的寄存器地址是 0x32 CHANNEL = 30 # Channel 30 RX = 2 TX = 1 INIT_RETRY_COUNT = 3 MAX_CONSECUTIVE_FAILURES = 3 RETRY_DELAY_MS = 100 PKT_DEBOUNCE_MS = 2 class PL1166ADriver: def __init__(self): self.i2c = None self.rst_pin = None self.pkt_pin = None self.addr = I2C_SLAVE_ADDR self.fc_mode = None self.consecutive_failures = 0 def reset_chip(self): if self.rst_pin: print("[RESET] 正在执行 PL1166A 硬件复位 (GPIO35/Pin39)...") self.rst_pin.write(0) time.sleep_ms(500) self.rst_pin.write(1) time.sleep_ms(500) print("[RESET] PL1166A 硬件复位完成。") def write_reg(self, reg, value): addr_buf = bytearray([reg & 0x7F]) data_buf = bytearray([(value >> 8) & 0xFF, value & 0xFF]) ret = self.i2c.write(self.addr, addr_buf, 1, data_buf, 2) time.sleep_ms(2) # 写入完成后强制等待 2ms，避免连续指令写入造成从机溢出 if ret != 0: raise OSError("I2C write failed, reg=0x%02X, ret=%d" % (reg, ret)) return ret def read_reg(self, reg): addr_buf = bytearray([reg & 0x7F]) r_buf = bytearray(2) if self.i2c.read(self.addr, addr_buf, 1, r_buf, 2, 0) == 0: val = (r_buf[0] << 8) | r_buf[1] time.sleep_ms(2) # 读取完成后等待 2ms 保护 return val else: time.sleep_ms(2) raise OSError("I2C read failed, reg=0x%02X" % reg) def clear_fifo(self): self.write_reg(REG_CLEAR_FIFO, 0x8080) def set_idle(self): if self.fc_mode != 0: self.write_reg(REG_MODE, CHANNEL) self.fc_mode = 0 def set_rx(self): self.clear_fifo() self.write_reg(REG_MODE, 0x0080 | CHANNEL) self.fc_mode = RX def has_packet(self): try: # 方式二：读取物理接收引脚 PKT (GPIO34/Pin38) 电平判定是否有包就绪 # 无线收包就绪后，引脚会被芯片自动拉低 (LOW/0) # 这种方式在等待期间不产生任何 I2C 物理信号，让总线和芯片免受高频通信轰炸 return self.pkt_pin.read() == 0 except Exception: return False def record_i2c_failure(self, reason): self.consecutive_failures += 1 self.fc_mode = None print("[I2C ERROR] %s (consecutive=%d)" % (reason, self.consecutive_failures)) if self.consecutive_failures >= MAX_CONSECUTIVE_FAILURES: print("[RECOVERY] 连续 I2C 失败，复位 PL1166A 并重新初始化...") self.consecutive_failures = 0 if not self.configure_registers(): print("[RECOVERY] PL1166A 重新初始化失败，停止本次访问。") def configure_registers(self): regs_to_write = [ (0x00, 0x6FE0), (0x01, 0x5681), (0x02, 0x6617), (0x04, 0x9CC9), (0x05, 0x6637), (0x08, 0x6C90), (0x09, 0x19C0), (0x0A, 0x7FFD), (0x0B, 0x0008), (0x0C, 0x0000), (0x0D, 0x48BD), (0x16, 0x00FF), (0x17, 0x8005), (0x18, 0x0067), (0x19, 0x1659), (0x1A, 0x19E0), (0x1B, 0x1300), (0x1C, 0x1800), (0x20, 0x5800), (0x21, 0x3FC7), (0x22, 0x2000), (0x23, 0x0080), (0x24, 0x4231), (0x25, 0x8675), (0x26, 0x9A0B), (0x27, 0xDECF), (0x28, 0x4401), (0x29, 0x9400), (0x2A, 0xFDFF), (0x2B, 0x000F) ] try: self.reset_chip() print("[PL1166A] 正在写入寄存器参数...") for reg, val in regs_to_write: self.write_reg(reg, val) print("寄存器写入完成，开始读取校验...") for reg, expected in regs_to_write: val = self.read_reg(reg) if val != expected: raise OSError("Reg 0x%02X mismatch: 0x%04X != 0x%04X" % (reg, val, expected)) self.set_rx() self.consecutive_failures = 0 print("[成功] 所有参数校验一致，已进入接收模式。") return True except Exception as e: self.fc_mode = None print("[INIT ERROR] %s" % e) return False def init_gpios(self): print("正在开启模组外设 IO 电压域...") try: machine.io_set_volt(1) print("IO 电压域开启成功 (1.8V)") except Exception as e: print("开启 IO 电压域失败:", e) time.sleep(1) # RSTB Pin self.rst_pin = Pin(RSTB_PIN_NUM, Pin.OUT, Pin.PULL_DISABLE, 1) # PKT Pin self.pkt_pin = Pin(PKT_PIN_NUM, Pin.IN, Pin.PULL_PU, 0) print("GPIO 引脚初始化成功 (RSTB=Pin39, PKT=Pin38)。") def init_i2c_and_registers(self): # 初始化软件模拟 I2C print("正在初始化软件模拟 I2C (SCL=GPIO24/Pin67, SDA=GPIO23/Pin66, 频率为 400K)...") self.i2c = I2C( I2C.I2C0, I2C.STANDARD_MODE, 1 ) for attempt in range(INIT_RETRY_COUNT): if self.configure_registers(): return True print("[INIT] 第 %d/%d 次初始化失败。" % (attempt + 1, INIT_RETRY_COUNT)) time.sleep_ms(RETRY_DELAY_MS) print("[INIT] 初始化重试已耗尽，请检查接线、供电或 RSTB 电平。") return False def read_fifo_once(self): try: self.set_idle() raw = bytearray(20) addr_buf = bytearray([CMD_READ_FIFO]) ret = self.i2c.read(self.addr, addr_buf, 1, raw, 20, 1) if ret != 0: raise OSError("I2C FIFO read failed, ret=%d" % ret) if raw[0] == 0x00 and raw[1] == 0x00: return None bitnum = raw[1] & 0x0F data_len = 2 + bitnum if data_len >= len(raw): raise ValueError("Invalid FIFO length: %d" % data_len) sum_rx = 0 for i in range(data_len): sum_rx = (sum_rx + raw[i]) & 0xFF rx_checksum = raw[data_len] if sum_rx != rx_checksum: print("[RX ERROR] Checksum mismatch! Raw data: %s" % " ".join(["%02X" % x for x in raw[:data_len + 1]])) return None self.consecutive_failures = 0 rx_data = raw[:data_len] hex_data = " ".join(["%02X" % x for x in rx_data]) print("[RX SUCCESS] Data: %s (len=%d)" % (hex_data, data_len)) return rx_data except Exception as e: self.record_i2c_failure(e) return None finally: try: self.set_rx() except Exception as e: self.record_i2c_failure(e) if __name__ == "__main__": print("==================================================") print(" PL1166A 软件模拟 I2C 收发测试程序") print("==================================================") spoll = uselect.poll() spoll.register(usys.stdin, uselect.POLLIN) driver = PL1166ADriver() try: driver.init_gpios() init_ok = driver.init_i2c_and_registers() if init_ok: print("\n系统进入监听状态。") while True: # 检查串口重启指令 if spoll.poll(0): char = usys.stdin.read(1) if char in ('R', 'r'): print("\n[REBOOT] 软重启模组...") time.sleep_ms(100) machine.reset() # 检查接收数据 if driver.has_packet(): time.sleep_ms(PKT_DEBOUNCE_MS) if driver.has_packet(): driver.read_fifo_once() time.sleep_ms(2) # 物理引脚扫描，设为 2ms 间隔，既有极高实时性，又极大节省模组 CPU 开销 else: print("\n【失败】校验未通过，已退出监听。") finally: # 清理资源 print("\n正在释放 GPIO 资源...") try: if driver.rst_pin: del driver.rst_pin if driver.pkt_pin: del driver.pkt_pin import gc gc.collect() print("资源已安全释放。") except Exception: pass