1. 项目概述在资源受限的微控制器上驾驭异步与内存如果你和我一样在玩转各种小巧的微控制器比如Adafruit的Feather系列、Raspberry Pi Pico时总会遇到两个绕不开的“天花板”一个是有限的RAM和Flash空间另一个是如何优雅地处理多个并发任务比如一边读取传感器数据一边通过无线电发送同时还得闪个LED灯。传统的“顺序执行延时”写法在稍微复杂点的项目里很快就会变得笨拙且低效。这正是CircuitPython的asyncio库和.mpy文件优化技术大显身手的地方。asyncio不是魔法它本质上是一种“协作式多任务”模型。想象一下你是一个厨师只有一个炉灶单核CPU。传统的做法是煮完一锅汤任务A再去炒菜任务B。而asyncio让你在等汤烧开的间隙I/O等待去切菜执行任务B。通过async/await语法你明确地告诉程序“这里我需要等待你可以先去干点别的。” 这对于需要同时监听按钮、更新屏幕、处理网络请求的物联网设备来说是革命性的。但光有“聪明”的调度还不够我们还得面对物理限制微控制器上那以KB计的内存。一个不经意的import或几行冗余代码就可能引发令人头疼的MemoryError。这时.mpy文件就成了我们的“空间压缩魔法”。它并不是什么高深莫测的东西你可以把它理解为Python源码.py的“预编译精简版”。CircuitPython解释器执行.mpy文件时无需再进行词法分析和语法分析等前端工作直接加载字节码这不仅能加快导入速度更重要的是能显著减少运行时内存占用。对于内存捉襟见肘的SAMD21仅有32KB RAM等平台这往往是项目能否成功运行的关键。本文将深入这两个核心主题不仅告诉你“是什么”和“怎么做”更会结合我实际踩过的坑分享“为什么”要这么做以及如何在实际项目中权衡与优化。无论你是刚接触CircuitPython的新手还是正在为项目内存优化而苦恼的开发者相信都能找到实用的解决方案。2. CircuitPython异步编程asyncio深度解析与实战2.1 asyncio在微控制器上的价值与限制在桌面或服务器环境我们有线程、进程甚至多核CPU来应对并发。但在微控制器世界硬件资源极其有限多个物理线程的开销往往是不可接受的。asyncio提供的是一种在单线程内实现并发的轻量级方案其核心是事件循环Event Loop和协程Coroutine。事件循环就像一个总调度员。它维护着一个任务队列不断检查哪些协程可以运行比如等待的I/O操作完成了哪些需要暂停遇到了await。协程则是一个可以暂停和恢复的函数用async def定义。当你在协程中调用await asyncio.sleep(1)或await some_sensor.read()时你实际上是在对事件循环说“我这个任务现在要等一会儿你先去执行队列里其他准备好了的任务吧。”这种模式在微控制器上优势明显极低的开销协程切换的成本远低于线程切换因为它不涉及操作系统内核和复杂的上下文保存。避免阻塞在等待传感器响应、网络数据包或用户输入时CPU不会被白白占用可以处理其他任务极大提高了系统响应能力和整体吞吐量。代码结构清晰用顺序编写的代码逻辑就能实现并发的效果避免了回调地狱Callback Hell或复杂的状态机。然而CircuitPython的asyncio实现有其特定限制这也是理解它的关键无抢占式调度这是“协作式”的核心。如果一个协程长时间运行而不await例如一个计算密集的循环它会独占事件循环导致其他所有任务“饿死”。你必须主动在代码中插入await asyncio.sleep(0)或分解任务来“让出”控制权。不支持中断IRQ这是很多从Arduino或MicroPython转过来的开发者会困惑的一点。CircuitPython目前不提供硬件中断机制。所有异步响应比如按键检测都需要通过asyncio来轮询实现。这听起来效率低但在事件循环的框架下通过合理的await和短时轮询完全可以实现实时性要求不苛刻的响应。注意asyncio并非万能。对于绝对硬实时的任务如精确控制步进电机脉冲你可能仍然需要借助time.monotonic_ns()进行精细的时间轮询或者考虑其他实时性更强的平台。asyncio更适合处理“人类时间尺度”或“网络时间尺度”的并发。2.2 核心概念与基础用法从零搭建异步任务让我们从一个最简单的例子开始创建两个交替闪烁的LED任务。这是理解事件循环和任务创建的最佳起点。首先你需要确保你的CircuitPython固件版本是7.1.0或更高。你可以通过连接到串行REPL并输入import os; os.uname()来查看版本。import asyncio import board import digitalio # 初始化两个LED led1 digitalio.DigitalInOut(board.LED) led1.direction digitalio.Direction.OUTPUT led2 digitalio.DigitalInOut(board.D13) # 假设D13是另一个可用的LED引脚 led2.direction digitalio.Direction.OUTPUT async def blink(led, interval, name): 一个简单的协程用于以指定间隔闪烁LED while True: led.value not led.value print(f{name}: {ON if led.value else OFF}) # 关键使用asyncio.sleep来暂停此协程并让出控制权 await asyncio.sleep(interval) async def main(): 主协程用于创建并运行多个任务 # 创建两个任务Task。任务是对协程的封装由事件循环调度。 task1 asyncio.create_task(blink(led1, 0.5, LED1)) task2 asyncio.create_task(blink(led2, 0.7, LED2)) # 等待所有任务完成在这个例子里两个都是无限循环所以这里会一直运行 # 更常见的做法是这里可以等待一个特定的结束信号 await asyncio.gather(task1, task2) # 启动事件循环并运行主协程 # 在CircuitPython中通常使用asyncio.run()来启动顶层入口 asyncio.run(main())代码解析与避坑指南asyncio.create_task(): 这个函数将协程如blink(led1, 0.5, “LED1”)包装成一个Task对象并立即将其加入事件循环的调度队列。任务一旦创建就会在后台开始运行无需显式await它。await asyncio.gather(task1, task2)的作用是挂起main协程直到task1和task2都完成本例中不会完成。await asyncio.sleep(): 这是协作式多任务的核心。它告诉事件循环“我这个协程要休眠interval秒在此期间你可以去运行其他就绪的协程。” 如果这里用普通的time.sleep()整个事件循环都会被阻塞另一个LED就无法闪烁。asyncio.run(main()): 这是启动异步程序的推荐方式。它会创建一个新的事件循环运行传入的协程main并在其完成后关闭循环。在code.py中这通常是最后一行。实操心得在微控制器上打印print到串行控制台是一个相对较慢的I/O操作。在频繁执行的协程中大量使用print可能会轻微影响其他任务的时序。在性能要求高的场景可以考虑将日志信息缓存定期批量输出或者通过一个专门的日志任务来管理输出。2.3 高级模式任务通信、取消与资源管理实际项目很少只是简单闪烁LED。我们经常需要任务间通信例如传感器任务将数据发送给网络上传任务或者一个UI任务响应按钮事件。1. 使用asyncio.Queue进行任务间通信 队列是生产者-消费者模式的完美实现。下面模拟一个温度传感器读取任务和一个数据上传任务。import asyncio import random import time async def sensor_producer(queue): 模拟传感器每隔随机时间产生一个数据并放入队列 sensor_id 0 while True: # 模拟读取传感器数据 temperature 20.0 random.uniform(-2, 2) data {id: sensor_id, temp: temperature, timestamp: time.monotonic()} await queue.put(data) # 将数据放入队列如果队列满则会等待 print(fProduced: {data}) sensor_id 1 await asyncio.sleep(random.uniform(0.5, 2)) # 模拟不规律的读取间隔 async def data_consumer(queue): 从队列中取出数据并处理例如上传到云端 while True: data await queue.get() # 从队列获取数据如果队列空则会等待 # 模拟网络上传耗时 print(fConsuming: {data}, simulating upload...) await asyncio.sleep(0.8) print(fUploaded: {data}) queue.task_done() # 通知队列该任务已完成 async def main(): queue asyncio.Queue(maxsize5) # 创建一个最大容量为5的队列 producer_task asyncio.create_task(sensor_producer(queue)) consumer_task asyncio.create_task(data_consumer(queue)) # 让它们运行一段时间比如20秒 await asyncio.sleep(20) # 取消任务 producer_task.cancel() consumer_task.cancel() # 等待任务被妥善取消处理CancelledError try: await producer_task except asyncio.CancelledError: print(Producer task cancelled.) try: await consumer_task except asyncio.CancelledError: print(Consumer task cancelled.) print(Done.) asyncio.run(main())2. 任务的取消与清理 如示例所示使用task.cancel()可以请求取消一个任务。被取消的任务在下一个await点会抛出asyncio.CancelledError。务必在顶层捕获这个异常来完成清理工作比如关闭文件句柄、释放硬件资源将GPIO设为输入模式等。不清理资源可能会导致硬件处于不确定状态。3. 使用asyncio.Event或asyncio.Lock进行同步Event: 用于通知一个或多个任务某个事件已发生。比如一个网络连接成功的事件可以唤醒多个等待发送数据的任务。connection_ready asyncio.Event() async def wait_for_connection(): await connection_ready.wait() # 挂起直到event被set print(Connection is ready!) # 在某个地方如连接成功的回调里 connection_ready.set()Lock: 用于保护共享资源防止多个协程同时访问。在CircuitPython中对于硬件资源如SPI总线、特定I2C设备的访问使用锁是很好的实践。i2c_lock asyncio.Lock() async def read_sensor(sensor_addr): async with i2c_lock: # 确保同一时间只有一个协程在使用I2C总线 # ... 使用I2C总线读取传感器 ... await asyncio.sleep(0.01)重要提示在资源极度受限的板上如SAMD21创建大量任务或使用深度递归的协程可能会耗尽内存。务必根据实际情况合理设计任务数量并在不需要时及时取消任务以释放资源。3. 深入CircuitPython内存管理与.mpy文件优化3.1 理解MemoryError内存分配与碎片化当你看到MemoryError时意味着CircuitPython无法从堆heap中分配出足够连续的内存块来满足请求。微控制器的RAM通常很小32KB-256KB常见这块内存需要同时存放Python对象你代码中创建的所有变量、列表、字典、字符串等。字节码正在执行的.py或.mpy文件编译后的代码对象。库代码导入的库所占用的内存。运行时栈用于函数调用和协程切换。内存碎片化是另一个隐形杀手。想象你的内存是一块空白画布。你先分配了一个大对象A占一块然后分配了一些小对象B、C之后释放了A。现在画布上有一大块空闲空间但它被B和C“夹在”中间。如果接下来要分配一个比这个空闲空间更大、但比原来A小的对象仍然可能失败因为找不到足够大的连续空间。导入顺序会影响内存布局进而影响碎片化程度。这就是为什么官方文档会提到导入顺序可能影响内存可用性。如何诊断内存问题检查空闲内存在REPL或代码中定期使用import gc; print(gc.mem_free())。观察在程序不同阶段初始化后、运行一段时间后内存的变化。使用gc.collect()手动触发垃圾回收。虽然CircuitPython有自动回收机制但在已知会创建大量临时对象如循环中拼接大字符串后手动调用gc.collect()有时可以立即释放内存缓解压力。留意状态LED很多CircuitPython板载有一个RGB NeoPixel作为状态指示灯。不同的颜色闪烁模式可能指示内存分配失败或其他错误。查阅你的开发板对应文档了解其含义。3.2 .mpy文件原理、创建与部署.mpyMicroPython字节码文件是CircuitPython内存优化的利器。它的优势来自于“预编译”节省RAM.py文件在导入时需要被解析Parse和编译Compile成字节码这个过程本身会在堆上创建语法树、代码对象等临时结构消耗RAM。而.mpy文件直接包含编译好的字节码导入时省去了编译步骤直接加载减少了临时内存开销。节省Flash/磁盘空间字节码通常比源代码文本更紧凑。加快导入速度省去了编译时间对于大型库尤其明显。保护源代码.mpy是二进制格式逆向工程难度远高于.py文件。如何创建.mpy文件你需要一个叫mpy-cross的交叉编译器。它运行在你的开发电脑Windows/macOS/Linux上将.py文件编译为.mpy文件。步骤详解获取mpy-cross前往CircuitPython的GitHub发布页面找到与你目标板CircuitPython固件版本完全一致的mpy-cross工具。版本不匹配可能导致生成的.mpy文件无法运行。编译单个文件打开终端或命令提示符导航到mpy-cross所在目录。# Linux/macOS ./mpy-cross path/to/your_library.py # Windows mpy-cross.exe path\to\your_library.py这将在your_library.py的同目录下生成your_library.mpy。部署到设备将生成的.mpy文件而非.py文件复制到你的CIRCUITPY驱动器的lib文件夹中。在你的主程序code.py中像导入普通模块一样导入它import your_library。批量编译与高级用法 对于包含多个模块的库你需要分别编译每个.py文件。注意mpy-cross不会自动处理包包含__init__.py的文件夹。你需要单独编译__init__.py以及包内的其他模块文件并保持原有的目录结构复制到lib下。# 假设你的库结构如下 # my_lib/ # __init__.py # sensor.py # utils.py # 编译 ./mpy-cross my_lib/__init__.py ./mpy-cross my_lib/sensor.py ./mpy-cross my_lib/utils.py # 在CIRCUITPY的lib文件夹下创建my_lib目录并将生成的三个.mpy文件放入。3.3 系统性的内存优化策略仅靠.mpy文件可能不足以解决所有内存问题。下面是一套组合拳优先使用.mpy格式的库从Adafruit的CircuitPython Library Bundle下载的库本身就提供了.mpy版本。务必使用与你的固件版本匹配的bundle。将你自己的代码模块化并编译为.mpy如果你的code.py很长可以将其中的函数和类拆分到独立的模块文件中编译成.mpy后再导入。甚至可以将整个code.py编译成code.mpy注意编译后你将无法直接在板上编辑它。优化代码习惯避免在全局作用域执行大量代码或分配大对象全局变量在导入时就会初始化并占用内存。将初始化代码放入函数中在需要时调用。谨慎使用字符串字符串在Python中是不可变的频繁拼接尤其是用会产生大量临时对象。对于路径或日志考虑使用””.join(list_of_parts)的方式。使用bytearray代替大list如果需要存储大量数值数据bytearray或array模块的数组类型比list更节省内存。及时删除大对象的引用对于不再需要的大列表、字典显式地del它们并调用gc.collect()。简化或移除注释在最终部署时.py文件中的注释也会被加载到内存尽管影响较小。编译为.mpy后注释会被彻底移除。管理导入按需导入在函数内部导入模块而不是在文件顶部。这样模块只在函数被调用时才加载。使用from module import specific_function而不是import module这样可以避免将整个模块的命名空间引入当前作用域。但要注意这并不会显著减少内存占用因为模块本身已经被完整加载。注意导入顺序虽然没有银弹规则但一个常见的建议是先导入较大的、基础的核心库再导入较小的、应用特定的库。这有助于减少内存碎片。如果遇到奇怪的MemoryError尝试调整import语句的顺序有时会有意外收获。4. 实战构建一个异步环境监测站让我们综合运用所学设计一个简单的环境监测站。它需要异步读取温湿度传感器如DHT22或AHT20。异步读取光照强度传感器。将数据异步发送到串行控制台模拟上传到网络。所有任务互不阻塞并且要高效利用内存。硬件假设主控板Adafruit Feather RP2040内存相对宽裕。传感器AHT20I2C温湿度BH1750I2C光照强度。代码结构lib/ adafruit_ahtx0.mpy # 使用.mpy版本库 adafruit_bh1750.mpy code.py (或 code.mpy)核心代码实现import asyncio import board import busio import gc from digitalio import DigitalInOut, Direction import adafruit_ahtx0 import adafruit_bh1750 # 初始化I2C总线 i2c busio.I2C(board.SCL, board.SDA) # 初始化传感器对象 # 注意实际硬件操作如I2C读取是阻塞的但我们将它们包裹在async函数中 # 并通过asyncio.sleep(0)或短时等待来让出控制权。 try: aht20 adafruit_ahtx0.AHTx0(i2c) bh1750 adafruit_bh1750.BH1750(i2c) except Exception as e: print(fSensor init failed: {e}) # 在实际项目中这里可能需要更优雅的错误处理比如进入安全模式 aht20 bh1750 None # 创建一个锁来保护对I2C总线的访问防止多个传感器同时访问冲突 i2c_lock asyncio.Lock() # 创建一个队列用于从传感器任务向上传任务传递数据 data_queue asyncio.Queue(maxsize10) async def read_aht20(): 异步读取温湿度传感器 if not aht20: return while True: async with i2c_lock: # 实际的传感器读取是阻塞的但时间很短通常几毫秒 temperature aht20.temperature humidity aht20.relative_humidity data {sensor: AHT20, temp: temperature, humidity: humidity} try: # 非阻塞方式放入队列如果队列满则跳过本次数据根据需求调整策略 data_queue.put_nowait(data) except asyncio.QueueFull: print(Warning: Data queue full, dropping AHT20 data.) # 让出控制权并等待下一次读取 await asyncio.sleep(2) # 每2秒读一次 async def read_bh1750(): 异步读取光照传感器 if not bh1750: return while True: async with i2c_lock: lux bh1750.lux data {sensor: BH1750, lux: lux} try: data_queue.put_nowait(data) except asyncio.QueueFull: print(Warning: Data queue full, dropping BH1750 data.) await asyncio.sleep(1) # 每1秒读一次 async def upload_data(): 模拟数据上传任务此处打印到串口 while True: data await data_queue.get() # 等待数据到来 # 模拟上传过程网络延迟 await asyncio.sleep(0.5) print(fUploaded: {data}) data_queue.task_done() # 偶尔手动触发垃圾回收观察效果 if gc.mem_free() 15000: # 假设空闲内存低于15KB时触发 gc.collect() print(fGC triggered. Free mem: {gc.mem_free()} bytes) async def monitor_memory(): 后台内存监控任务 while True: await asyncio.sleep(10) free_mem gc.mem_free() print(f[Memory Monitor] Free: {free_mem} bytes) if free_mem 10000: print(f[Memory Monitor] WARNING: Memory low!) async def main(): print(Environmental Station Starting...) print(fInitial free memory: {gc.mem_free()} bytes) # 创建所有任务 tasks [ asyncio.create_task(read_aht20()), asyncio.create_task(read_bh1750()), asyncio.create_task(upload_data()), asyncio.create_task(monitor_memory()), ] # 这里我们让程序一直运行。在实际应用中你可能需要一个停止信号。 # 例如等待一个按键事件然后取消所有任务。 try: await asyncio.gather(*tasks) except asyncio.CancelledError: # 处理任务取消进行必要的清理 print(Tasks cancelled. Cleaning up.) # 例如将GPIO设为安全状态 finally: print(Station stopped.) # 启动异步应用 asyncio.run(main())项目优化与部署编译为.mpy将上述code.py保存并使用mpy-cross编译为code.mpy。然后将code.mpy和所需的.mpy格式库文件一起放入CIRCUITPY根目录。板子会自动运行code.mpy。内存监控monitor_memory任务让你能实时了解内存使用情况便于发现内存泄漏如果空闲内存持续下降。错误处理增强示例中的错误处理比较简单。生产代码应增加传感器断线重连、队列满时的更优策略如丢弃旧数据、网络上传失败重试等逻辑。低功耗考虑在电池供电场景可以在传感器读取间隙使用await asyncio.sleep()让CPU进入空闲状态并结合板子的深层睡眠模式如果支持来进一步省电。通过这个实战项目你应该能体会到结合asyncio的并发能力和.mpy的内存优化我们可以在资源有限的微控制器上构建出响应迅速、功能相对复杂的应用。关键在于理解协作式多任务的本质并养成良好的内存管理习惯。