Python用字典模拟switch功能处理MQTT消息控制4路继电器求教
Python MQTT控制4路继电器优雅实现方案
核心实现思路
使用两层字典替代嵌套的条件判断,第一层完成MQTT Topic到位掩码的映射,第二层完成Payload到位运算逻辑的映射,全程无冗余代码,时间复杂度为O(1)。
代码实现
基础版(全局状态)
# 继电器全局状态,初始全关(4位均为0) relay_state = 0b0000 # 第一层映射:Topic对应继电器的位掩码 TOPIC_MASK_MAP = { "Z1": 1 << 0, # 对应第1位,掩码0b0001 "Z2": 1 << 1, # 对应第2位,掩码0b0010 "Z3": 1 << 2, # 对应第3位,掩码0b0100 "Z4": 1 << 3 # 对应第4位,掩码0b1000 } # 第二层映射:Payload对应的位运算逻辑,入参为当前状态、掩码,返回新状态 PAYLOAD_OP_MAP = { "ON": lambda current, mask: current | mask, "OFF": lambda current, mask: current & ~mask } # MQTT消息处理函数 def handle_mqtt_message(receive_topic: str, receive_payload: str): global relay_state # 非法消息直接过滤,可自行添加日志打印 if receive_topic not in TOPIC_MASK_MAP: return mask = TOPIC_MASK_MAP[receive_topic] # 统一转大写避免Payload大小写问题 payload_upper = receive_payload.upper() if payload_upper not in PAYLOAD_OP_MAP: return # 执行运算更新状态 relay_state = PAYLOAD_OP_MAP[payload_upper](relay_state, mask) # 此处添加你自己的硬件控制逻辑,将relay_state写入继电器寄存器
面向对象版(无全局变量)
如果需要多实例控制,或者不想使用全局变量,可以封装为控制类,逻辑完全一致:
class RelayController: def __init__(self): # 初始状态全关 self.relay_state = 0b0000 self.topic_mask_map = { "Z1": 1 << 0, "Z2": 1 << 1, "Z3": 1 << 2, "Z4": 1 << 3 } self.payload_op_map = { "ON": lambda current, mask: current | mask, "OFF": lambda current, mask: current & ~mask } def handle_mqtt_message(self, receive_topic: str, receive_payload: str): if receive_topic not in self.topic_mask_map: return mask = self.topic_mask_map[receive_topic] payload_upper = receive_payload.upper() if payload_upper not in self.payload_op_map: return self.relay_state = self.payload_op_map[payload_upper](self.relay_state, mask) # 硬件控制逻辑写在这里
方案优势
- 无冗余嵌套判断,执行效率远高于多层if/else
- 扩展性极强:后续新增继电器通道仅需在
topic_mask_map新增映射即可,新增Payload操作(比如翻转TOGGLE)仅需在payload_op_map新增对应运算逻辑,不需要修改核心处理代码 - 规则集中管理,后续修改映射规则不需要梳理逻辑分支,维护成本极低
内容的提问来源于stack exchange,提问作者HeneryH
相关产品推荐
相关产品推荐

