传统仪器断电数据丢失，程序实现关键测量，数据自动存入闪存，断电重启不丢失。

智能仪器断电数据保护系统一、实际应用场景描述场景某工业自动化生产线的在线质量检测工位部署了20台智能测量仪器游标卡尺、千分尺、数字量具等用于实时检测产品尺寸精度。这些仪器每天24小时不间断工作经常因产线电力维护、突发停电或设备重启导致测量数据丢失。业务痛点当测量到关键尺寸数据如公差边缘产品、首件检验、不良品复测时一旦断电这些数据立即消失。质检员需要在断电后重新测量严重影响生产效率且可能导致批次质量问题追溯困难。需求开发一套断电数据保护系统实现关键测量数据自动存入闪存EEPROM/Flash断电重启后数据完整保留支持历史数据查询和导出。二、引入痛点痛点 具体影响 经济损失估算断电数据丢失 关键测量数据瞬间消失需重复测量 单次测量损失30分钟工时批次追溯困难 断电前后数据断层无法完整追溯 质量问题定位时间增加4小时人工备份繁琐 操作员需手动抄录重要数据 每人每天额外工作1小时数据一致性差 断电重启后仪器零点漂移 测量误差导致废品率上升2%合规审计风险 缺乏完整数据链不符合ISO9001 认证审核不通过风险解决方案通过Python编写断电数据保护程序结合硬件Flash存储芯片实现测量即保存断电不丢失支持自动备份、数据加密和历史查询。三、核心逻辑讲解┌─────────────┐ ┌─────────────┐ ┌─────────────┐│ 仪器测量 │────▶│ 判断关键数据│────▶│ 写入Flash │└─────────────┘ └─────────────┘ └─────────────┘│▼┌─────────────┐ ┌─────────────┐ ┌─────────────┐│ 数据持久化 │◀────│ Flash状态 │◀────│ 断电检测 │└─────────────┘ └─────────────┘ └─────────────┘│▼┌─────────────┐ ┌─────────────┐ ┌─────────────┐│ 数据恢复 │◀────│ 启动时扫描 │◀────│ 电源恢复 │└─────────────┘ └─────────────┘ └─────────────┘核心技术点1. 断电检测机制监控电源电压检测到电压跌落趋势时触发紧急保存2. Flash存储策略采用环形缓冲区结构循环写入自动覆盖旧数据3. 掉电保护算法检测到断电信号后预留足够时间完成最后一批数据写入4. 数据完整性校验CRC16校验确保数据正确性损坏数据自动标记5. 分层存储架构关键数据首检/终检→Flash常规数据→RAM缓存四、代码模块化实现项目结构power_loss_protection/├── main.py # 主程序入口├── flash_manager.py # Flash存储管理模块├── power_monitor.py # 断电检测模块├── data_encoder.py # 数据编码加密模块├── measurement_interface.py # 测量数据接口模块├── recovery_manager.py # 数据恢复管理模块├── config.json # 配置文件├── requirements.txt # 依赖清单└── README.md # 说明文档1. config.json - 配置文件{flash: {start_address: 0x0000,total_size: 65536,sector_size: 4096,write_unit: 256},data_storage: {critical_threshold: 0.02,max_records: 1000,backup_interval: 300},power_monitor: {voltage_drop_threshold: 4.5,warning_voltage: 4.8,check_interval: 0.1},measurement: {tool_id: CALIPER_001,product_line: LINE_A,operator_id: OP001}}2. requirements.txt - 依赖清单pyserial3.5numpy1.21.0cryptography3.4.8crccheck0.63. flash_manager.py - Flash存储管理模块Flash存储管理模块功能封装Flash芯片操作实现断电安全的环形缓冲数据存储硬件接口SPI Flash (W25Q64/W25Q128等型号)import structimport timefrom typing import Optional, List, Tuplefrom dataclasses import dataclassfrom enum import Enumclass FlashStatus(Enum):Flash操作状态码SUCCESS 0WRITE_ERROR 1ERASE_ERROR 2READ_ERROR 3FULL 4CHECKSUM_ERROR 5dataclassclass StorageRecord:存储记录数据结构record_id: int # 记录唯一IDtimestamp: float # Unix时间戳tool_id: str # 仪器编号product_line: str # 生产线编号operator_id: str # 操作员编号measurement_value: float # 测量值tolerance_min: float # 公差下限tolerance_max: float # 公差上限is_critical: bool # 是否关键数据crc16: int # CRC校验码class FlashManager:Flash存储管理器类实现断电安全的环形缓冲区数据存储# Flash指令集W25Q系列兼容CMD_WRITE_ENABLE 0x06CMD_WRITE_DISABLE 0x04CMD_READ_STATUS 0x05CMD_READ_DATA 0x03CMD_PAGE_PROGRAM 0x02CMD_SECTOR_ERASE 0x20CMD_CHIP_ERASE 0xC7def __init__(self, config_path: str config.json):初始化Flash管理器参数config_path: 配置文件路径self.config self._load_config(config_path)self.flash_config self.config[flash]# 计算存储参数self.total_sectors self.flash_config[total_size] // self.flash_config[sector_size]self.records_per_sector self.flash_config[sector_size] // self.flash_config[write_unit]# 环形缓冲区指针self.head_ptr 0 # 写入位置头指针self.tail_ptr 0 # 读取位置尾指针self.record_count 0# 硬件接口模拟SPI Flashself._spi_interface Noneself._initialized Falseprint(f[Flash管理] 初始化完成总容量: {self.flash_config[total_size]}字节)def _load_config(self, config_path: str) - dict:加载配置文件import jsontry:with open(config_path, r, encodingutf-8) as f:return json.load(f)except FileNotFoundError:print(f[Flash管理] 警告配置文件不存在使用默认配置)return self._get_default_config()def _get_default_config(self) - dict:获取默认配置return {flash: {start_address: 0x0000,total_size: 65536,sector_size: 4096,write_unit: 256}}def connect_hardware(self, port: str /dev/spidev0.0) - bool:连接Flash硬件参数port: SPI设备路径返回连接是否成功try:# 实际项目中这里会初始化spidev# import spidev# self._spi_interface spidev.SpiDev()# self._spi_interface.open(0, 0)# self._spi_interface.max_speed_hz 8000000# 模拟连接成功self._spi_interface simulatedself._initialized Trueprint(f[Flash管理] 硬件连接成功: {port})return Trueexcept Exception as e:print(f[Flash管理] 硬件连接失败: {e})return Falsedef _wait_for_ready(self, timeout: float 1.0) - bool:等待Flash就绪参数timeout: 超时时间秒返回Flash是否就绪start_time time.time()while time.time() - start_time timeout:# 读取状态寄存器status self._read_status_register()if status 0x01 0: # WIP位为0表示就绪return Truetime.sleep(0.001)return Falsedef _read_status_register(self) - int:读取Flash状态寄存器# 实际SPI通信代码# self._spi_interface.xfer2([self.CMD_READ_STATUS])# result self._spi_interface.readbytes(1)# return result[0]# 模拟返回就绪状态return 0x00def _send_command(self, cmd: int, address: int 0, data: bytes b) - bool:发送Flash命令参数cmd: 命令字节address: 地址可选data: 写入数据可选返回操作是否成功if not self._initialized:print([Flash管理] 错误Flash未初始化)return Falsetry:# 实际SPI通信代码# tx_buffer [cmd]# if address 0:# tx_buffer.extend([(address 16) 0xFF,# (address 8) 0xFF,# address 0xFF])# if data:# tx_buffer.extend(list(data))# self._spi_interface.xfer2(tx_buffer)# 模拟成功time.sleep(0.001) # 模拟传输延迟return Trueexcept Exception as e:print(f[Flash管理] 命令发送失败: {e})return Falsedef _calculate_crc16(self, data: bytes) - int:计算CRC16校验码使用CRC-16/MODBUS算法参数data: 待校验数据返回CRC16校验码crc 0xFFFFpolynomial 0xA001for byte in data:crc ^ bytefor _ in range(8):if crc 0x0001:crc (crc 1) ^ polynomialelse:crc 1return crcdef _encode_record(self, record: StorageRecord) - bytes:编码记录为二进制格式参数record: 存储记录对象返回编码后的字节数据# 使用struct打包二进制数据# 格式说明# I: unsigned int (4字节) - record_id# d: double (8字节) - timestamp# 32s: 32字节字符串 - tool_id# H: unsigned short (2字节) - crc16encoded struct.pack(Id32s32s32sdffff?H,record.record_id,record.timestamp,record.tool_id.encode(ascii),record.product_line.encode(ascii),record.operator_id.encode(ascii),record.measurement_value,record.tolerance_min,record.tolerance_max,record.is_critical,record.crc16)# 填充到固定长度padded encoded.ljust(self.flash_config[write_unit], b\x00)return padded[:self.flash_config[write_unit]]def _decode_record(self, raw_data: bytes) - Optional[StorageRecord]:解码二进制数据为记录对象参数raw_data: 原始二进制数据返回存储记录对象或None解码失败try:unpacked struct.unpack(Id32s32s32sdffff?H, raw_data)return StorageRecord(record_idunpacked[0],timestampunpacked[1],tool_idunpacked[2].decode(ascii).rstrip(\x00),product_lineunpacked[3].decode(ascii).rstrip(\x00),operator_idunpacked[4].decode(ascii).rstrip(\x00),measurement_valueunpacked[5],tolerance_minunpacked[6],tolerance_maxunpacked[7],is_criticalunpacked[8],crc16unpacked[9])except Exception as e:print(f[Flash管理] 解码失败: {e})return Nonedef write_record(self, measurement_value: float,tolerance_min: float,tolerance_max: float,is_critical: bool False) - Tuple[FlashStatus, int]:写入一条测量记录到Flash参数measurement_value: 测量值tolerance_min: 公差下限tolerance_max: 公差上限is_critical: 是否为关键数据返回(操作状态, 记录ID)if not self._initialized:return FlashStatus.WRITE_ERROR, -1# 生成记录ID基于时间戳record_id int(time.time() * 1000) % 0xFFFFFFFF# 创建记录对象record StorageRecord(record_idrecord_id,timestamptime.time(),tool_idself.config[measurement][tool_id],product_lineself.config[measurement][product_line],operator_idself.config[measurement][operator_id],measurement_valuemeasurement_value,tolerance_mintolerance_min,tolerance_maxtolerance_max,is_criticalis_critical,crc160 # 临时值后面计算)# 编码并计算CRCencoded_data self._encode_record(record)record.crc16 self._calculate_crc16(encoded_data)encoded_data self._encode_record(record)# 计算Flash地址address self.flash_config[start_address] self.head_ptr * self.flash_config[write_unit]# 检查是否需要擦除扇区新扇区首次写入sector_num (address - self.flash_config[start_address]) // self.flash_config[sector_size]if self.head_ptr % self.records_per_sector 0:if not self._erase_sector(sector_num):return FlashStatus.ERASE_ERROR, -1# 写入数据if not self._send_write_enable():return FlashStatus.WRITE_ERROR, -1if not self._send_command(self.CMD_PAGE_PROGRAM, address, encoded_data):return FlashStatus.WRITE_ERROR, -1if not self._wait_for_ready():return FlashStatus.WRITE_ERROR, -1# 更新环形缓冲区指针self.head_ptr (self.head_ptr 1) % self.config[data_storage][max_records]if self.record_count self.config[data_storage][max_records]:self.record_count 1print(f[Flash管理] 记录写入成功 ID{record_id} 0x{address:06X})return FlashStatus.SUCCESS, record_iddef _send_write_enable(self) - bool:发送写使能命令return self._send_command(self.CMD_WRITE_ENABLE)def _erase_sector(self, sector_num: int) - bool:擦除指定扇区参数sector_num: 扇区编号返回擦除是否成功address self.flash_config[start_address] sector_num * self.flash_config[sector_size]print(f[Flash管理] 擦除扇区 {sector_num} 0x{address:06X})if not self._send_write_enable():return Falseif not self._send_command(self.CMD_SECTOR_ERASE, address):return Falsereturn self._wait_for_ready(timeout5.0) # 扇区擦除较慢def read_records(self, count: int 10,only_critical: bool False) - List[StorageRecord]:从Flash读取记录参数count: 读取记录数量only_critical: 是否只读取关键数据返回存储记录列表records []read_ptr self.tail_ptrattempts 0max_attempts self.record_countwhile len(records) count and attempts max_attempts:address self.flash_config[start_address] read_ptr * self.flash_config[write_unit]# 读取原始数据raw_data self._read_flash_data(address, self.flash_config[write_unit])if raw_data:record self._decode_record(raw_data)if record:# CRC校验calculated_crc self._calculate_crc16(raw_data[:-2])if calculated_crc record.crc16:if not only_critical or record.is_critical:records.append(record)else:print(f[Flash管理] CRC校验失败 ID{record.record_id})read_ptr (read_ptr 1) % self.config[data_storage][max_records]attempts 1return recordsdef _read_flash_data(self, address: int, length: int) - Optional[bytes]:从Flash读取原始数据参数address: 起始地址length: 读取长度返回读取的数据字节或Nonetry:# 实际SPI通信代码# self._spi_interface.xfer2([self.CMD_READ_DATA])# self._spi_interface.xfer2([# (address 16) 0xFF,# (address 8) 0xFF,# address 0xFF# ])# data self._spi_interface.readbytes(length)# return bytes(data)# 模拟读取返回随机数据用于测试import osreturn os.urandom(length)except Exception as e:print(f[Flash管理] 读取失败 0x{address:06X}: {e})return Nonedef get_storage_status(self) - dict:获取存储状态信息返回存储状态字典used_sectors (self.record_count * self.flash_config[write_unit]) // self.flash_config[sector_size]if used_sectors 0 and self.record_count 0:used_sectors 1return {total_capacity: self.flash_config[total_size],used_capacity: self.record_count * self.flash_config[write_unit],available_capacity: (self.config[data_storage][max_records] - self.record_count) * self.flash_config[write_unit],record_count: self.record_count,max_records: self.config[data_storage][max_records],usage_percent: (self.record_count / self.config[data_storage][max_records]) * 100,head_pointer: self.head_ptr,tail_pointer: self.tail_ptr,used_sectors: used_sectors,total_sectors: self.total_sectors}def emergency_flush(self) - FlashStatus:紧急刷新缓冲区数据断电前调用返回刷新状态print([Flash管理] 执行紧急数据刷新...)# 确保所有挂起的操作完成if not self._wait_for_ready(timeout0.5):print([Flash管理] 警告Flash未及时就绪)# 强制写入写保护禁用self._send_write_enable()print([Flash管理] 紧急刷新完成)return FlashStatus.SUCCESSdef format_storage(self) - FlashStatus:格式化存储空间擦除所有数据返回格式化状态print([Flash管理] 开始格式化存储...)total_sectors self.total_sectorsfor sector in range(total_sectors):if not self._erase_sector(sector):print(f[Flash管理] 扇区 {sector} 擦除失败)return FlashStatus.ERASE_ERRORprint(f进度: {sector1}/{total_sectors}, end\r)# 重置指针self.head_ptr 0self.tail_ptr 0self.record_count 0print(f\n[Flash管理] 格式化完成共擦除 {total_sectors} 个扇区)return FlashStatus.SUCCESSif __name__ __main__:# 测试Flash管理器print(*50)print(Flash存储管理模块测试)print(*50)# 初始化fm FlashManager(config.json)if fm.connect_hardware():# 显示存储状态status fm.get_storage_status()print(f存储状态: {status})# 写入测试记录print(\n写入测试记录...)for i in range(3):status, rec_id fm.write_record(measurement_value25.36 i * 0.1,tolerance_min25.0,tolerance_max25.5,is_critical(i 1) # 第二条为关键数据)print(f 记录 {i1}: 状态{status.name}, ID{rec_id})# 读取记录print(\n读取记录...)records fm.read_records(count5)for rec in records:critical_mark 关键 if rec.is_critical else ⚪普通print(f {critical_mark} ID{rec.record_id} f值{rec.measurement_value:.4f} f时间{time.strftime(%Y-%m-%d %H:%M:%S, time.localtime(rec.timestamp))})# 再次显示状态status fm.get_storage_status()print(f\n存储状态: {status})4. power_monitor.py - 断电检测模块断电检测模块功能实时监控电源电压检测断电趋势触发紧急数据保存import timeimport threadingfrom typing import Callable, Optionalfrom dataclasses import dataclassfrom enum import Enumfrom queue import Queueclass PowerState(Enum):电源状态枚举NORMAL normalWARNING warningCRITICAL criticalPOWER_LOSS power_lossdataclassclass PowerEvent:电源事件数据结构event_type: PowerStatevoltage: floattimestamp: floatmessage: strclass PowerMonitor:断电检测器类实时监控电源电压多级预警机制def __init__(self, config_path: str config.json):初始化断电检测器参数config_path: 配置文件路径self.config self._load_config(config_path)self.power_config self.config[power_monitor]# 电源参数self.voltage_drop_threshold self.power_config[voltage_drop_threshold]self.warning_voltage self.power_config[warning_voltage]self.check_interval self.power_config[check_interval]# 状态变量self.current_voltage 5.0 # 模拟当前电压self.current_state PowerState.NORMALself.is_monitoring Falseself.monitor_thread: Optional[threading.Thread] None# 回调函数self.emergency_call利用AI解决实际问题如果你觉得这个工具好用欢迎关注长安牧笛