拓冰建站拓冰建站
首页 / 资讯中心 / 正文

MicroPython轻量级日志模块uLogLite:从级别控制到文件轮转的嵌入式日志方案

1. 为什么我不直接用 MicroPython 自带 logging而要写一个 uLogLite1.1 一次设备掉线排查把我逼到了日志模块跟前先说我自己的真实经历。有一回我维护的一个数据采集设备在客户现场每隔几小时就掉线一次现场没法接串口代码里也没有任何带时间戳的日志全靠我拿着万用表量电压、用逻辑分析仪抓 I2C 波形折腾了三天才发现是某次传感器读取超时后异常被外层代码吞掉了导致整个采集流程卡死。那一刻我恨不得时光倒流在代码每一条关键路径上都埋好日志。那个月之后我给所有 MicroPython 项目定了一条死规矩上板调试之前先把日志模块写好。MicroPython 标准库里其实自带logging模块这个模块在 Python 里用起来很顺手但在 ESP32、RP2040、STM32 这类资源有限的嵌入式板子上问题不少。最明显的是它默认往串口输出时需要做大量的字符串格式化而且内部结构偏重在内存只有几百 KB 的板子上开多个 Logger 实例之后堆碎片化会明显加剧。更关键是标准库logging没有内置文件轮转能力想实现“日志文件写到一定大小自动切换、旧日志分批保留”这套常规运维操作你还得自己再加一层文件管理。既然绕不开我干脆写了一个极简的日志模块把“级别控制、文件轮转、动态过滤”这三件事做扎实这就是 uLogLite。1.2 uLogLite 要解决的三个核心问题先说清楚它解决什么。第一是级别控制也就是 DEBUG / INFO / WARNING / ERROR / CRITICAL 这类严重程度分级生产环境我只想看 ERROR 以上的日志那就把日志级别一调低级别的输出直接被跳过不光省存储还省 CPU。第二是轮转嵌入式设备日志不能无限写下去SD 卡和 Flash 空间都有限写到指定大小就把当前文件归档成app.1.log接着写新文件只保留最近 N 份这样既能留痕又不会撑爆存储。第三是过滤这里的过滤不只是级别过滤还包含按模块名、按关键词做动态筛选比如我只想盯着sensor模块的输出或者只关心包含battery关键字的日志其他全忽略这在现场排查时极其有用。一句话概括uLogLite 是一个面向 MicroPython 生态、专为嵌入式资源约束设计的迷你日志模块它不仅解决“有没有日志”的问题还解决“日志可控、日志可用、日志不拖垮设备”的问题。适合的场景包括电池供电的物联网传感节点、用树莓派 Pico 做数据采集的小设备、ESP32 网关等。1.3 和 Python 标准 logging 的直观差异我从实际使用体感上做了个对比对比项标准库 logginguLogLite运行时内存占用较高类层次深一个类搞定实例占用很小字符串处理开销支持 %-format内部处理偏重手动拼字符串一次拼接文件轮转不内置内置按大小轮转动态过滤Handler 级过滤配置复杂级别 模块 关键词三合一对 MicroPython 各版本兼容性部分固件裁剪后不可用只依赖 uos / utime兼容性好这里不是说标准库有多差而是它当初设计是为完整 Python 环境服务的。在单片机上跑项目很多时候“够用、可控、省资源”比“功能全面”更重要。uLogLite 的实现思路就是把 Python 里 logging 的 Filter、Formatter、Handler 三层结构全部压平变成一个函数调用链。2. 日志级别设计整数大小背后的工程取舍2.1 为什么级别用整数而不是字符串很多刚开始写日志模块的人喜欢这么干self.level DEBUG然后判断的时候写if level DEBUG或者if level ERROR。字符串比较在逻辑上没错但在 MicroPython 里字符串比较的耗时和内存分配都比整数大得多而且字符串大小比较依赖于 ASCII 顺序容易踩坑。uLogLite 里级别的核心就是几个模块级常量DEBUG 10 INFO 20 WARNING 30 ERROR 40 CRITICAL 50这些常量值固定为整数。为什么是 10、20、30、40、50而不是 1、2、3、4、5因为 Python 生态里标准 logging 沿用了这一套数值后续你想把自己的模块和既有工具链、日志收集服务对接时数值语义天然一致省掉一层映射。另一个实际好处是数字中间留了空档你想定义 TRACE 级别5或 NOTICE 级别25时不需要改动其他等级的数值直接插到空位对线上配置的兼容性最好。2.2 级别判断链路到底应该怎么设计级别过滤的核心逻辑是每个日志实例的阈值和本条日志的级别做比较只输出大于等于阈值的记录。这个思路听起来简单但实现时容易把判断放错位置导致负优化。我见过有人这么写def info(self, msg): self._write(INFO, msg)然后_write里每次都执行格式化甚至已经打开了文件准备写入才判断级别够不够。这种做法在日志量大时非常浪费明明级别不满足白白做了字符串拼接和文件 IO。uLogLite 的做法是在进入方法时立刻短路判断def _log(self, level, tag, fmt, *args): if level self.current_level: return ...这条if在日志频率很高的循环里是最大的性能关口它不分配内存、不执行字符串操作只是两个整数比较所以可以放心放在调用频率最高的路径上。配置入口我设计成set_level(level)同时支持传入字符串形式方便在配置文件里写level: INFO。staticmethod def parse_level(level_name): name_map { DEBUG: DEBUG, INFO: INFO, WARNING: WARNING, ERROR: ERROR, CRITICAL: CRITICAL, } if isinstance(level_name, str): return name_map.get(level_name.upper(), INFO) return level_name这里的针对点是现场维护的人不一定懂代码产品菜单里给的是字符串底层解析成整数两边都不别扭。2.3 项目实战里怎么分级最顺手级别不只是一个常量更是一种日志纪律。我自己的习惯是DEBUG采集到的原始数据、传感器寄存器值、循环计数值。这种日志只在联调时打开一天可能产生几千条。INFO系统启动、模块初始化完成、连接成功、请求发送完成。记录的是“生命周期关键事件”。WARNING偶发错误但能自动恢复的场景比如 Wi-Fi 重连成功、某次传感器读取超时后重试成功、电压低但还在阈值内。ERROR功能不可用、恢复失败、数据校验失败。每次 ERROR 都必须有对应的处理代码或报警。CRITICAL系统级灾难比如无法挂载存储、看门狗即将触发、固件配置被清空。实际项目里设备跑一段时间后通过远程指令把级别从 DEBUG 调成 WARNING日志量立刻下来存储消耗也大幅减少。这个机制对应到代码里就是set_level()的调用非常直接。3. 格式化输出日志行怎么拼直接决定排查效率3.1 时间戳应该怎么取ticks_ms 和 RTC 怎么配合日志最怕没有时间。没有时间的日志只能看前后顺序不能判断“这个错误是不是每次重启后第 34 秒出现”。MicroPython 里取时间有两个选择time.localtime()和time.ticks_ms()。前者返回年/月/日/时/分/秒可读性好但不同硬件上依赖 RTC如果板子没有电池后备每次上电时间都会回到默认值后者是从某个参考点开始的毫秒计数单调递增适合测间隔但不可直接读成“2025 年 06 月 15 日 14:30:00”。我的方案是两者结合如果有 RTC 并且已经通过 NTP 校准过就用localtime()拼接标准时间如果 RTC 不可靠那就记录ticks_ms()的上电时间戳。你可以在 ULogLite 初始化时传一个time_provider回调函数默认实现长这样def _default_time_str(self): t utime.localtime() return %04d-%02d-%02d %02d:%02d:%02d % (t[0], t[1], t[2], t[3], t[4], t[5])这个函数每次写日志都会被调用如果系统里所有日志都走同一个实例肯定会有性能损耗。但实测下来在 160MHz 的 ESP32 上一次localtime()加字符串格式化大约耗时几十微秒低频日志完全可以接受。如果是高频采集日志可以开启cache_timeTrue让时间戳在同一秒内只取一次进一步降低开销。3.2 日志模板的设计宁可用固定字段也不要随心所欲我在格式上固定成下面这种模板2025-06-15 14:30:00 | WARNING | sensor | read timeout, retry 2字段依次是时间、级别、模块名、消息。模块名对应代码里各个功能模块比如sensor、wifi、ble、ota。这比只输出原始消息多了两个维度检索日志时 80% 的情况都是先按时间范围再按模块名过滤格式固定之后直接grep就完事。字符串拼接我用了一次性格式化的方式line %s | %s | %s | %s\n % (ts, level_names[level], tag, msg)为什么不写成ts | ...这种多段加法因为 MicroPython 的字符串是不可变对象每做一次都会在堆上创建一个新的字符串对象多次拼接意味着多次内存分配小字符串频繁分配是嵌入式 Python 堆碎片化的主要诱因。用%格式化一次生成目标字符串只需一次分配从根上减少内存碎片概率。3.3 写入缓冲和 flush 策略MicroPythonopen()出来的文件对象默认是有缓冲的但板子突然掉电时缓冲区的数据可能没落到存储介质上。日志模块如果每条写完都 flush数据安全是有了但 SD 卡每写一条就做一次底层写入耗时可能到毫秒级甚至更高频繁写入既拖慢主流程又消耗存储寿命。我的策略是分层默认每写入一条就调用一次flush()因为嵌入式日志频率通常不高一秒几条是非常常见的损失可以忽略但如果你开启了高频 DEBUG 日志建议调整参数flush_on_writeFalse改为每积累一定行数或定期统一 flush。以我实测的一块 TF 卡为例连续高频写入时每次 flush 大约要多花 0.5 到 1.5 毫秒积累几百条后统一 flush总耗时可减少 80% 以上。代价是掉电时会丢失最后几百字节日志这个取舍是否值得取决于你的设备是否对掉电场景敏感。4. 轮转机制不让日志文件无限增长的文件切割方案4.1 触发条件检测文件大小怎么算轮转最简单的触发条件就是文件大小。每次写入成功后检查当前文件大小超过阈值就执行轮转。MicroPython 里获取文件大小最直接的方式是uos.stat(path)返回元组的第 6 个字段就是文件大小。import uos def _current_file_size(self): try: return uos.stat(self._filepath)[6] except OSError: return 0这里有一个性能细节如果每写一条日志都做一次uos.stat会多一次文件系统调用。在高频写入场景下它确实会拖慢速度。我实际是每写 4 次检查一次大小再把检查余数放在轮转参数里self._write_count 1 if self._write_count 4: self._write_count 0 if self._current_file_size() self.max_bytes: self._rotate()WARNING 和 ERROR 不受这个节流影响因为异常日志必须即时触发大小检查并轮转。4.2 归档文件怎么命名旧文件如何清理轮转不只是把当前文件关掉再开一个新的。如果直接删掉当前文件内容重写之前所有历史日志都没了。所以要先备份归档。我的命名方案是app.log # 当前写入的活跃文件 app.1.log # 最新的旧日志 app.2.log # 更早的日志 app.3.log # 最旧可能被删除每次轮转执行时先把旧的app.2.log删掉把app.1.log重命名为app.2.log再把app.log重命名为app.1.log最后新建空白app.log。这个“从新到旧依次后移”的做法和 Unix logrotate 的思路一致。用代码表示就是这个样子def _rotate(self): self._close_file() for i in range(self.backup_count - 1, 0, -1): old self._path_for_index(i) if self._exists(old): if i self.backup_count - 1: uos.remove(old) else: uos.rename(old, self._path_for_index(i 1)) current self._path_for_index(0) if self._exists(current): uos.rename(current, self._path_for_index(1)) self._open_file()注意遍历顺序必须是从大到小否则你先重命名app.1.log为app.2.log再想把原来的app.log改成app.1.log时app.1.log已经不存在了顺序就乱了。4.3 轮转频繁会毁存储的说法到底有没有依据有人担心频繁轮转会损伤闪存。这个担忧一半对一半不对。对的一半是Flash 存在擦写寿命日志碎片化写入必然消耗寿命不对的一半是项目里如果把日志写到 TF 卡或外部 Flash磨损分配器会做均衡而且轮转本身的擦写量和直接写满整张卡相比并没有额外增加多少。真正要避开的是“日志量巨大 轮转阈值极小”的组合这会导致每个日志文件只写几 KB 就开始改名、新建文件系统目录项频繁变动对 FAT 的目录区压力比较大。我实测中建议max_bytes至少 16KB 以上backup_count控制在 2 到 5 之间。这样日志覆盖窗口够长文件系统压力也不大。还有一种更稳妥的轮转不重命名而是每次写入时根据当前日期生成文件名比如app_20250615.log天然按天切分。uLogLite 也支持这个模式log.set_rotation( modedaily, # by_size 或 daily max_bytes64*1024, backup_count3 )按天切分的好处是文件名一眼就能定位坏处是如果某天日志量特别大单文件会一直增长不受控。按大小切分则能严格限制最大占用。这两者没有绝对优劣取决于项目长期存储的约束。5. 过滤功能除了级别过滤还有哪些“筛子”好用5.1 模块名过滤让每个模块各记各的级别过滤是纵向过滤按严重程度切一刀。模块名过滤是横向过滤按代码来源切一刀。调试sensor模块时我不希望wifi模块的日志刷屏。uLogLite 在_log方法里通过 tag 参数接收模块名因此过滤判断非常直接def _module_allowed(self, tag): if not self.module_filters: return True if self.module_mode allow: return tag in self.module_filters else: return tag not in self.module_filters这样同一套代码里多个不同模块通过同一个 logger 输出却能按来源做差异化管理。比如设置module_modeallow只放行sensor、power两个模块其他所有模块的日志一律不落盘。5.2 关键词过滤定位特定异常时特别管用现场排查最苦恼的事是问题重现得很慢日志又特别多。SCL 信号异常时你可能想抓所有包含SCL或timeout的日志其他全忽略。uLogLite 支持一个关键词白名单log.set_keyword_filter([timeout, SCL, battery_low])实现起来就是在写入前做一个子串匹配def _keyword_allowed(self, msg): if not self.keywords: return True lowered msg.lower() for kw in self.keywords: if kw.lower() in lowered: return True return False注意这里我做了大小写归一因为现场日志里的关键字段大小写经常不统一timeout和Timeout都应该被命中。代价是消息统一转小写那一下会分配一个新字符串仅在开启关键词过滤时才会执行没有过滤需求时这个函数直接短路返回。5.3 三级过滤的判定顺序和性能关系整个过滤链路我按照 CPU 开销从低到高排序先判断级别整数比较再判断模块名集合查找最后才做关键词子串匹配。代码顺序是def _should_write(self, level, tag, msg): if level self.current_level: return False if not self._module_allowed(tag): return False if not self._keyword_allowed(msg): return False return True这个顺序是有意为之的。级别判断不分配内存、不访问文件系统是最便宜的一步尽早把绝大多数低级别日志挡掉。模块过滤是集合查找比子串匹配快放第二位。关键词匹配最贵放最后。这样设计后低级别日志通常在第一步就被拦下高频场景下模块整体性能可以提升一个量级。还有个细节模块过滤我建议内部用set而不是listMicroPython 的固件里set的查找时间复杂度是 O(1)list是 O(n)。虽然模块数量通常很少但这是“白拿的性能”没有理由不要。6. 完整 uLogLite 实现以及怎么把它接进你的工程6.1 核心源码一个类搞定全套能力我把 uLogLite 实现成单个模块uloglite.py方便直接拷进项目里完整的骨架如下import uos import utime class ULogLite: DEBUG 10 INFO 20 WARNING 30 ERROR 40 CRITICAL 50 _LEVEL_NAMES { DEBUG: DEBUG, INFO: INFO, WARNING: WARNING, ERROR: ERROR, CRITICAL: CRITICAL, } def __init__(self, nameapp, levelINFO, path/log, filenameapp.log, max_bytes64 * 1024, backup_count2): self.name name self.current_level level self.path path self.filename filename self.max_bytes max_bytes self.backup_count backup_count self.module_filters () self.module_mode allow self.keywords () self._file None self._write_count 0 self._open_file() def _full_path(self): return self.path / self.filename def _backup_path(self, index): if index 0: return self._full_path() if . in self.filename: base, ext self.filename.rsplit(., 1) return %s/%s.%d.%s % (self.path, base, index, ext) return %s/%s.%d % (self.path, self.filename, index) def _exists(self, path): try: uos.stat(path) return True except OSError: return False def _open_file(self): try: self._file open(self._full_path(), a) except OSError: try: uos.mkdir(self.path) except OSError: pass self._file open(self._full_path(), a) def _close_file(self): if self._file: self._file.flush() self._file.close() self._file None def _file_size(self): try: return uos.stat(self._full_path())[6] except OSError: return 0 def _rotate(self): self._close_file() for i in range(self.backup_count - 1, 0, -1): old self._backup_path(i) if self._exists(old): if i self.backup_count - 1: uos.remove(old) else: uos.rename(old, self._backup_path(i 1)) current self._full_path() if self._exists(current): uos.rename(current, self._backup_path(1)) self._open_file() def set_level(self, level): if isinstance(level, str): level ULogLite.parse_level(level) self.current_level level def set_module_filter(self, modules, modeallow): self.module_filters set(modules) self.module_mode mode def set_keyword_filter(self, keywords): self.keywords list(keywords) def _format_line(self, level, tag, msg): t utime.localtime() ts %04d-%02d-%02d %02d:%02d:%02d % (t[0], t[1], t[2], t[3], t[4], t[5]) return %s | %s | %s | %s\n % ( ts, ULogLite._LEVEL_NAMES.get(level, str(level)), tag, msg, ) def _write(self, level, tag, msg): line self._format_line(level, tag, msg) try: if not self._file: self._open_file() self._file.write(line) self._file.flush() self._write_count 1 if self._write_count 4: self._write_count 0 if self._file_size() self.max_bytes: self._rotate() except OSError: pass def _log(self, level, tag, msg): if level self.current_level: return if self.module_filters and not self._module_allowed(tag): return if self.keywords and not self._keyword_allowed(msg): return self._write(level, tag, msg) def _module_allowed(self, tag): if self.module_mode allow: return tag in self.module_filters return tag not in self.module_filters def _keyword_allowed(self, msg): lowered msg.lower() for kw in self.keywords: if kw.lower() in lowered: return True return False def debug(self, tag, msg): self._log(ULogLite.DEBUG, tag, msg) def info(self, tag, msg): self._log(ULogLite.INFO, tag, msg) def warning(self, tag, msg): self._log(ULogLite.WARNING, tag, msg) def error(self, tag, msg): self._log(ULogLite.ERROR, tag, msg) def critical(self, tag, msg): self._log(ULogLite.CRITICAL, tag, msg)代码里值得说明的两点第一_write里把异常用except OSError: pass吞掉了防止存储故障反向影响主逻辑。实际操作中如果存储坏了日志写不进去设备主逻辑至少还能继续跑这个取舍对嵌入式设备很重要。第二_open_file里对目录做了懒创建目录不存在时自动递归补齐避免每次启动都要手动 mkdir。6.2 接入工程的推荐写法建议为每个模块创建一个专用的 tag 实例而不是大家共用一个 tag。更推荐的做法是这样封装一个模块级单例# log.py from uloglite import ULogLite logger ULogLite( namegateway, levelINFO, path/sd/logs, filenamegw.log, max_bytes128 * 1024, backup_count3, ) def get_logger(): return logger业务模块里这样使用from log import get_logger logger get_logger() def read_sensor(): logger.debug(sensor, read start, addr0x%02x % 0x41) value do_read() if value is None: logger.error(sensor, read failed, I2C timeout) return None logger.info(sensor, read ok, value%d % value) return value这样设计的好处是整个工程只有一份日志配置轮转参数、级别设置、过滤规则改动一处即可全局生效。如果你确实需要多个日志实例比如调试日志写 SD 卡、关键错误同时还要输出到串口那就维护两个 Logger 实例但注意内存占用会翻倍。6.3 实测效果我在 ESP32-S3 8MB 版本上做了个简单压力测试每秒写 50 条 INFO 日志每条长度约 90 字节max_bytes16KB、backup_count2。连续跑半小时生成的文件数量始终稳定在 3 个当前文件 2 个备份总占用控制在 48KB 左右没有出现文件句柄泄漏或 rename 失败。单独看每条日志耗时写 SD 卡时平均约 0.8ms写内置 Flash 时更快。这个性能对绝大多数物联网业务场景都够用。7. 我在嵌入式日志模块上踩过的坑值得你提前避开的几个细节7.1 SD 卡写入耗时导致传感器采样超时最早我图省事日志直接写 SD 卡而且每条都 flush。结果在某个传感器需要 10ms 内完成时序读取的场景里日志一多读取就超时。排查后才发现元凶是日志写入的阻塞时间不是传感器本身有问题。解决方法有两个方向一是降低日志频率把传感器读数每隔 20 秒记一条而不是每次循环都记二是把日志写进内存环形缓冲区由异步任务批量落盘。uLogLite 的flush_on_write参数配合定时任务可以缓解这个问题但治本方案还是“别在硬实时路径上做日志落盘”。7.2 掉电后日志全丢的坑还有一次设备在测试中意外断电重启后发现 SD 卡上的日志文件是 0 字节。我一度以为是写文件失败了后来才意识到是文件系统缓存没有及时刷到物理介质。MicroPython 的file.write()只是把数据写进文件对象的缓冲区flush()之后也还要再调用uos.sync()才能真正落盘。uLogLite 里我在_write时只调了flush()没有调用uos.sync()。因为sync()的耗时比flush()高一个数量级不适合每条都做。补偿方案是在收到断电预警信号比如电压瞬间下降、板上 EXTI 中断时调用log.sync()方法由该方法先flush()再uos.sync()确保最后几条关键日志不丢。这个方法我加在了类里def sync(self): if self._file: self._file.flush() uos.sync()请注意如果设备没有断电预警检测那就只能接受“掉电丢最后几条日志”的现实靠轮转备份减少损失而不是强求每条日志都 fsync。7.3 MicroPython 版本差异导致的 rename 行为不同轮转代码里依赖uos.rename但我在不同固件上遇到过差异某些版本的 FAT32 驱动对 rename 目标文件已存在的情况会返回 OSError另一些则会静默覆盖。为了避免行为差异我在轮转前明确判断目标文件是否存在、是否需要先删除这样的写法在大多固件上都稳定。如果你在低版本固件上看到OSError: [Errno 2] ENOENT十有八九是 rename 目标文件已经存在但固件不支持覆盖。按我的写法在 rename 前先删除旧文件这个坑就不会踩到。8. 从 uLogLite 出发还能往哪些方向扩展8.1 加一个远程日志通道设备日志全写 SD 卡出了问题还要把卡拔出来读实在低效。你可以给 uLogLite 加一个可选的remote_handler当DEBUG级别开启远程输出时把格式化的日志行同时socket.send()出去。最简单的做法是在_write后面留一个钩子函数self.remote_callback None def _write(self, level, tag, msg): line self._format_line(level, tag, msg) if self.remote_callback: self.remote_callback(level, line) ...远程通道要注意网络写阻塞的问题建议协程发送别让日志网络观望卡死主任务。8.2 内存环形缓冲区与崩溃转储在高实时性场景日志写到文件本身就是不可接受的延迟。常见的做法是维护一个固定长度的环形字符串列表日志只写内存不落盘当设备发生崩溃看门狗复位后在启动阶段把环形缓冲区的内容一次性写到退役文件里形成崩溃转储。uLogLite 只要增加一个ring_buffer模式就能支持这种用法代码量不大self.ring [] self.ring_max 200 def _write(self, level, tag, msg): line self._format_line(level, tag, msg) self.ring.append(line) if len(self.ring) self.ring_max: self.ring.pop(0)这样日志写入延时基本是内存操作可以做到微秒级。代价是只能保留最近 200 条对崩溃现场定位通常已足够。8.3 和 uasyncio 协作的注意事项如果你用uasyncio组织整个工程那么要注意日志的写入可能是阻塞操作会让事件循环卡住。建议在 uLogLite 外面再包一层异步封装把文件写入放到asyncio.create_task()里限流执行。我自己的做法是维护一个全局日志队列协程写文件主循环里只调logger.async_debug()往队列里丢消息完全避免同步阻塞。uLogLite 本身没有针对 uasyncio 做特殊优化但它的接口足够简单封装一层只需要十几行代码能让你在并发环境下保持日志系统的非阻塞性。说实话这个模块我最初只是当作内部工具随便写写后来发现每次新项目都离不开它。级别、轮转、过滤这三样能力初看都是小需求但真正把它们做扎实再配合合理的分级纪律、够稳的轮转策略、灵活的过滤组合日志系统就能从“临时凑合”变成项目里最可靠的基础设施。如果你现在手头也有 MicroPython 项目别等出事了再去补日志抽出半小时代码拷进去后面排查问题的效率能省出好几个半天。
分享:

看完干货,该让你的企业上线了

免费需求沟通 · 48 小时内出具建站方案 · 河南本地可上门