主逻辑(通信暂时只有抽象定义)

Makefile

@@ -1,2 +0,0 @@
-swag:
-	python -m grpc_tools.protoc -I./proto --python_out=./proto ./proto/client.proto

client_pb.py

@@ -1,306 +0,0 @@
-#!/usr/bin/env python
-# -*- coding: utf-8 -*-
-
-"""
-根据client.proto生成的解析代码
-适用于ESP32 MicroPython环境
-"""
-
-import struct
-
-# MethodType枚举
-METHOD_TYPE_SWITCH = 0
-METHOD_TYPE_COLLECT = 1
-
-def encode_varint(value):
-    """编码varint值"""
-    buf = bytearray()
-    while value >= 0x80:
-        buf.append((value & 0x7F) | 0x80)
-        value >>= 7
-    buf.append(value & 0x7F)
-    return buf
-
-def decode_varint(buf, pos=0):
-    """解码varint值"""
-    result = 0
-    shift = 0
-    while pos < len(buf):
-        byte = buf[pos]
-        pos += 1
-        result |= (byte & 0x7F) << shift
-        if not (byte & 0x80):
-            break
-        shift += 7
-    return result, pos
-
-def encode_string(value):
-    """编码字符串"""
-    value_bytes = value.encode('utf-8')
-    length = encode_varint(len(value_bytes))
-    return length + value_bytes
-
-def decode_string(buf, pos=0):
-    """解码字符串"""
-    length, pos = decode_varint(buf, pos)
-    value = buf[pos:pos+length].decode('utf-8')
-    pos += length
-    return value, pos
-
-def encode_instruction(method, data):
-    """
-    编码Instruction消息
-    
-    Args:
-        method: 方法类型 (int)
-        data: 数据 (bytes)
-        
-    Returns:
-        bytearray: 编码后的数据
-    """
-    result = bytearray()
-    
-    # 编码method字段 (field_number=1, wire_type=0)
-    result.extend(encode_varint((1 << 3) | 0))  # tag
-    result.extend(encode_varint(method))        # value
-    
-    # 编码data字段 (field_number=2, wire_type=2)
-    result.extend(encode_varint((2 << 3) | 2))  # tag
-    result.extend(encode_varint(len(data)))     # length
-    result.extend(data)                         # value
-    
-    return result
-
-def decode_instruction(buf):
-    """
-    解码Instruction消息
-    
-    Args:
-        buf: 编码后的数据 (bytes)
-        
-    Returns:
-        dict: 解码后的消息
-    """
-    result = {}
-    pos = 0
-    
-    while pos < len(buf):
-        # 读取标签
-        tag, pos = decode_varint(buf, pos)
-        field_number = tag >> 3
-        wire_type = tag & 0x07
-        
-        if field_number == 1:  # method字段
-            if wire_type == 0:  # varint类型
-                value, pos = decode_varint(buf, pos)
-                result['method'] = value
-        elif field_number == 2:  # data字段
-            if wire_type == 2:  # 长度分隔类型
-                length, pos = decode_varint(buf, pos)
-                value = buf[pos:pos+length]
-                pos += length
-                result['data'] = value
-        else:
-            # 跳过未知字段
-            if wire_type == 0:  # varint
-                _, pos = decode_varint(buf, pos)
-            elif wire_type == 2:  # 长度分隔
-                length, pos = decode_varint(buf, pos)
-                pos += length
-            else:
-                pos += 1
-                
-    return result
-
-def encode_switch(device_action, bus_number, bus_address, relay_channel):
-    """
-    编码Switch消息
-    
-    Args:
-        device_action: 设备动作指令 (str)
-        bus_number: 总线号 (int)
-        bus_address: 总线地址 (int)
-        relay_channel: 继电器通道号 (int)
-        
-    Returns:
-        bytearray: 编码后的数据
-    """
-    result = bytearray()
-    
-    # 编码device_action字段 (field_number=1, wire_type=2)
-    result.extend(encode_varint((1 << 3) | 2))      # tag
-    action_bytes = encode_string(device_action)     # value (length + string)
-    result.extend(action_bytes)
-    
-    # 编码bus_number字段 (field_number=2, wire_type=0)
-    result.extend(encode_varint((2 << 3) | 0))      # tag
-    result.extend(encode_varint(bus_number))        # value
-    
-    # 编码bus_address字段 (field_number=3, wire_type=0)
-    result.extend(encode_varint((3 << 3) | 0))      # tag
-    result.extend(encode_varint(bus_address))       # value
-    
-    # 编码relay_channel字段 (field_number=4, wire_type=0)
-    result.extend(encode_varint((4 << 3) | 0))      # tag
-    result.extend(encode_varint(relay_channel))     # value
-    
-    return result
-
-def decode_switch(buf):
-    """
-    解码Switch消息
-    
-    Args:
-        buf: 编码后的数据 (bytes)
-        
-    Returns:
-        dict: 解码后的消息
-    """
-    result = {}
-    pos = 0
-    
-    while pos < len(buf):
-        # 读取标签
-        tag, pos = decode_varint(buf, pos)
-        field_number = tag >> 3
-        wire_type = tag & 0x07
-        
-        if field_number == 1:  # device_action字段
-            if wire_type == 2:  # 字符串类型
-                value, pos = decode_string(buf, pos)
-                result['device_action'] = value
-        elif field_number == 2:  # bus_number字段
-            if wire_type == 0:  # varint类型
-                value, pos = decode_varint(buf, pos)
-                result['bus_number'] = value
-        elif field_number == 3:  # bus_address字段
-            if wire_type == 0:  # varint类型
-                value, pos = decode_varint(buf, pos)
-                result['bus_address'] = value
-        elif field_number == 4:  # relay_channel字段
-            if wire_type == 0:  # varint类型
-                value, pos = decode_varint(buf, pos)
-                result['relay_channel'] = value
-        else:
-            # 跳过未知字段
-            if wire_type == 0:  # varint
-                _, pos = decode_varint(buf, pos)
-            elif wire_type == 2:  # 长度分隔
-                length, pos = decode_varint(buf, pos)
-                pos += length
-            else:
-                pos += 1
-                
-    return result
-
-def encode_collect(bus_number, bus_address, value):
-    """
-    编码Collect消息
-    
-    Args:
-        bus_number: 总线号 (int)
-        bus_address: 总线地址 (int)
-        value: 采集值 (float)
-        
-    Returns:
-        bytearray: 编码后的数据
-    """
-    result = bytearray()
-    
-    # 编码bus_number字段 (field_number=1, wire_type=0)
-    result.extend(encode_varint((1 << 3) | 0))      # tag
-    result.extend(encode_varint(bus_number))        # value
-    
-    # 编码bus_address字段 (field_number=2, wire_type=0)
-    result.extend(encode_varint((2 << 3) | 0))      # tag
-    result.extend(encode_varint(bus_address))       # value
-    
-    # 编码value字段 (field_number=3, wire_type=5)
-    result.extend(encode_varint((3 << 3) | 5))      # tag
-    # 将float转换为little-endian的4字节
-    result.extend(struct.pack('<f', value))         # value
-    
-    return result
-
-def decode_collect(buf):
-    """
-    解码Collect消息
-    
-    Args:
-        buf: 编码后的数据 (bytes)
-        
-    Returns:
-        dict: 解码后的消息
-    """
-    result = {}
-    pos = 0
-    
-    while pos < len(buf):
-        # 读取标签
-        tag, pos = decode_varint(buf, pos)
-        field_number = tag >> 3
-        wire_type = tag & 0x07
-        
-        if field_number == 1:  # bus_number字段
-            if wire_type == 0:  # varint类型
-                value, pos = decode_varint(buf, pos)
-                result['bus_number'] = value
-        elif field_number == 2:  # bus_address字段
-            if wire_type == 0:  # varint类型
-                value, pos = decode_varint(buf, pos)
-                result['bus_address'] = value
-        elif field_number == 3:  # value字段
-            if wire_type == 5:  # 32位浮点类型
-                # 从little-endian的4字节解析float
-                value = struct.unpack('<f', buf[pos:pos+4])[0]
-                pos += 4
-                result['value'] = value
-        else:
-            # 跳过未知字段
-            if wire_type == 0:  # varint
-                _, pos = decode_varint(buf, pos)
-            elif wire_type == 5:  # 32位固定长度
-                pos += 4
-            elif wire_type == 2:  # 长度分隔
-                length, pos = decode_varint(buf, pos)
-                pos += length
-            else:
-                pos += 1
-                
-    return result
-
-# 使用示例
-if __name__ == "__main__":
-    # 创建一个Switch消息
-    switch_data = encode_switch("ON", 1, 10, 2)
-    print(f"编码后的Switch消息: {switch_data.hex()}")
-    
-    # 创建一个Instruction消息，包含Switch数据
-    instruction_data = encode_instruction(METHOD_TYPE_SWITCH, switch_data)
-    print(f"编码后的Instruction消息: {instruction_data.hex()}")
-    
-    # 解码Instruction消息
-    decoded_instruction = decode_instruction(instruction_data)
-    print(f"解码后的Instruction消息: {decoded_instruction}")
-    
-    # 解码Switch消息
-    if 'data' in decoded_instruction:
-        decoded_switch = decode_switch(decoded_instruction['data'])
-        print(f"解码后的Switch消息: {decoded_switch}")
-    
-    # 创建一个Collect消息
-    collect_data = encode_collect(1, 20, 25.6)
-    print(f"编码后的Collect消息: {collect_data.hex()}")
-    
-    # 创建一个Instruction消息，包含Collect数据
-    instruction_data2 = encode_instruction(METHOD_TYPE_COLLECT, collect_data)
-    print(f"编码后的Instruction消息(Collect): {instruction_data2.hex()}")
-    
-    # 解码Instruction消息
-    decoded_instruction2 = decode_instruction(instruction_data2)
-    print(f"解码后的Instruction消息: {decoded_instruction2}")
-    
-    # 解码Collect消息
-    if 'data' in decoded_instruction2:
-        decoded_collect = decode_collect(decoded_instruction2['data'])
-        print(f"解码后的Collect消息: {decoded_collect}")

main.py

@@ -1,317 +0,0 @@
-#!/usr/bin/env python
-# -*- coding: utf-8 -*-
-
-"""
-猪舍主控系统主程序入口
-"""
-
-import machine
-import time
-import struct
-import client_pb
-
-# 初始化RS485串口
-# 使用UART2，连接到ESP32的GPIO16(RX)和GPIO17(TX)
-rs485_uart = machine.UART(2, baudrate=9600, bits=8, parity=None, stop=1, rx=16, tx=17)
-rs485_uart.init()
-
-# RS485收发控制引脚
-rs485_re_de_pin = machine.Pin(5, machine.Pin.OUT)
-rs485_re_de_pin.value(0)  # 默认接收模式
-
-# ESP32设备地址（应该唯一标识这个ESP32设备）
-ESP32_ADDRESS = 1
-
-# LoRaWAN模块地址
-LORA_MODULE_ADDRESS = 254
-
-def receive_lora_message():
-    """
-    接收来自LoRaWAN模块的消息
-    返回: 字节数据
-    """
-    # 在共享的RS485总线上监听来自LoRaWAN模块的消息
-    # 需要检查消息是否是发给本设备的
-    buffer = bytearray()
-    
-    # 持续监听，不设置超时
-    while rs485_uart.any():
-        data = rs485_uart.read()
-        if data:
-            buffer.extend(data)
-            # 简单的帧检测逻辑
-            if len(buffer) >= 3 and buffer[0] == 0xAA and buffer[-1] == 0x55:
-                # 检查地址是否匹配
-                if len(buffer) >= 3 and buffer[2] == ESP32_ADDRESS:
-                    # 提取有效数据（去掉帧头、地址、校验和帧尾）
-                    return buffer[3:-2]
-        else:
-            break  # 没有更多数据可读
-    
-    return None
-
-def send_lora_message(data):
-    """
-    通过LoRaWAN模块发送消息
-    
-    参数:
-    data: 要发送的字节数据
-    """
-    # 切换到发送模式
-    rs485_re_de_pin.value(1)
-    time.sleep_ms(10)
-    
-    try:
-        # 构造发送给LoRaWAN模块的RS485帧
-        frame = bytearray()
-        frame.append(0xAA)  # 帧头
-        frame.append(LORA_MODULE_ADDRESS & 0xFF)  # LoRaWAN模块地址
-        frame.append(ESP32_ADDRESS & 0xFF)  # 本设备地址（作为源地址）
-        
-        # 添加数据
-        frame.extend(data)
-        
-        # 计算校验和
-        checksum = sum(frame[1:]) & 0xFF
-        frame.append(checksum)
-        frame.append(0x55)  # 帧尾
-        
-        # 发送命令
-        rs485_uart.write(frame)
-        print(f"通过LoRa发送数据: {frame.hex()}")
-        
-    finally:
-        # 切换回接收模式
-        time.sleep_ms(10)
-        rs485_re_de_pin.value(0)
-
-def send_rs485_command(bus_number, bus_address, command, channel=None):
-    """
-    发送命令到RS485总线上的设备
-    
-    参数:
-    bus_number: 总线号
-    bus_address: 设备地址
-    command: 命令内容
-    channel: 通道号（可选）
-    """
-    # 切换到发送模式
-    rs485_re_de_pin.value(1)
-    time.sleep_ms(10)
-    
-    try:
-        # 构造RS485命令帧
-        frame = bytearray()
-        frame.append(0xAA)  # 帧头
-        frame.append(bus_number & 0xFF)  # 总线号
-        frame.append(bus_address & 0xFF)  # 设备地址
-        
-        # 添加命令数据
-        if isinstance(command, str):
-            frame.extend(command.encode('utf-8'))
-        elif isinstance(command, bytes):
-            frame.extend(command)
-        elif isinstance(command, int):
-            frame.append(command & 0xFF)
-            
-        # 如果有通道号，则添加
-        if channel is not None:
-            frame.append(channel & 0xFF)
-            
-        # 计算校验和
-        checksum = sum(frame[1:]) & 0xFF
-        frame.append(checksum)
-        frame.append(0x55)  # 帧尾
-        
-        # 发送命令
-        rs485_uart.write(frame)
-        print(f"已发送RS485命令: {frame.hex()}")
-        
-    finally:
-        # 切换回接收模式
-        time.sleep_ms(10)
-        rs485_re_de_pin.value(0)
-
-def collect_sensor_data(bus_number, bus_address):
-    """
-    从传感器收集数据
-    
-    参数:
-    bus_number: 总线号
-    bus_address: 传感器地址
-    
-    返回:
-    传感器数据
-    """
-    # 切换到发送模式
-    rs485_re_de_pin.value(1)
-    time.sleep_ms(10)
-    
-    try:
-        # 构造读取传感器数据的命令
-        frame = bytearray([0xAA, bus_number & 0xFF, bus_address & 0xFF, 0x01, 0x00, 0x55])
-        rs485_uart.write(frame)
-        print(f"已发送传感器读取命令: {frame.hex()}")
-        
-        # 等待响应
-        time.sleep_ms(50)  # 短暂等待响应
-        
-        # 读取传感器返回的数据
-        buffer = bytearray()
-        
-        while rs485_uart.any():
-            data = rs485_uart.read()
-            if data:
-                buffer.extend(data)
-                # 简单的帧检测逻辑
-                if len(buffer) >= 3 and buffer[0] == 0xAA and buffer[-1] == 0x55:
-                    # 检查地址是否匹配
-                    if len(buffer) >= 3 and buffer[2] == ESP32_ADDRESS:
-                        # 提取有效数据
-                        if len(buffer) >= 5:
-                            # 模拟一个浮点数值
-                            value = float(buffer[3] + (buffer[4] << 8)) / 100.0
-                            return value
-            else:
-                break
-        
-        if len(buffer) > 0:
-            print(f"传感器响应不完整: {buffer.hex()}")
-        else:
-            print("传感器无响应")
-        return None
-            
-    finally:
-        # 切换回接收模式
-        time.sleep_ms(10)
-        rs485_re_de_pin.value(0)
-
-def parse_instruction(data):
-    """
-    解析来自LoRaWAN的指令
-    
-    参数:
-    data: protobuf编码的指令数据
-    
-    返回:
-    解析后的指令对象，失败时返回None
-    """
-    try:
-        instruction = client_pb.decode_instruction(data)
-        return instruction
-    except Exception as e:
-        print(f"解析指令失败: {e}")
-        return None
-
-def handle_switch_instruction(switch_msg):
-    """
-    处理开关指令
-    
-    参数:
-    switch_msg: Switch消息字典
-    """
-    action = switch_msg.get('device_action', '')
-    bus_number = switch_msg.get('bus_number', 0)
-    bus_address = switch_msg.get('bus_address', 0)
-    channel = switch_msg.get('relay_channel', 0)
-    
-    print(f"处理开关指令: 动作={action}, 总线={bus_number}, 地址={bus_address}, 通道={channel}")
-    
-    if action.upper() == "ON":
-        # 发送开启设备命令
-        send_rs485_command(bus_number, bus_address, 0x01, channel)
-    elif action.upper() == "OFF":
-        # 发送关闭设备命令
-        send_rs485_command(bus_number, bus_address, 0x00, channel)
-    else:
-        # 其他自定义命令
-        send_rs485_command(bus_number, bus_address, action, channel)
-
-def handle_collect_instruction(collect_msg):
-    """
-    处理采集指令
-    
-    参数:
-    collect_msg: Collect消息字典
-    """
-    bus_number = collect_msg.get('bus_number', 0)
-    bus_address = collect_msg.get('bus_address', 0)
-    
-    print(f"处理采集指令: 总线={bus_number}, 地址={bus_address}")
-    
-    # 从传感器采集数据
-    value = collect_sensor_data(bus_number, bus_address)
-    
-    if value is not None:
-        # 构造Collect响应消息
-        collect_data = client_pb.encode_collect(bus_number, bus_address, value)
-        
-        # 构造Instruction消息包装Collect数据
-        instruction_data = client_pb.encode_instruction(client_pb.METHOD_TYPE_COLLECT, collect_data)
-        
-        # 发送回上位机
-        send_lora_message(instruction_data)
-    else:
-        print("采集数据失败")
-
-def process_instruction(instruction):
-    """
-    处理解析后的指令
-    
-    参数:
-    instruction: 解析后的指令字典
-    """
-    method = instruction.get('method', -1)
-    
-    if method == client_pb.METHOD_TYPE_SWITCH:
-        # 处理开关指令
-        if 'data' in instruction:
-            switch_msg = client_pb.decode_switch(instruction['data'])
-            handle_switch_instruction(switch_msg)
-        else:
-            print("开关指令缺少data字段")
-    elif method == client_pb.METHOD_TYPE_COLLECT:
-        # 处理采集指令
-        if 'data' in instruction:
-            collect_msg = client_pb.decode_collect(instruction['data'])
-            handle_collect_instruction(collect_msg)
-        else:
-            print("采集指令缺少data字段")
-    else:
-        print(f"不支持的指令类型: {method}")
-
-def main_loop():
-    """
-    主循环
-    """
-    print("猪舍控制系统启动...")
-    print(f"设备地址: {ESP32_ADDRESS}")
-    
-    while True:
-        # 接收LoRaWAN消息
-        lora_data = receive_lora_message()
-        
-        if lora_data:
-            print(f"收到LoRaWAN消息: {lora_data.hex()}")
-            
-            # 解析指令
-            instruction = parse_instruction(lora_data)
-            
-            if instruction:
-                # 处理指令
-                process_instruction(instruction)
-            else:
-                print("无效的指令数据")
-        
-        # 其他周期性任务可以放在这里
-        # 例如定时采集传感器数据等
-        time.sleep(0.01)  # 短暂休眠避免过度占用CPU
-
-# 程序入口
-if __name__ == "__main__":
-    try:
-        main_loop()
-    except KeyboardInterrupt:
-        print("程序被中断")
-    except Exception as e:
-        print(f"程序异常: {e}")

main/bus/bus_interface.py

@@ -0,0 +1,47 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+
+"""
+总线通信模块的抽象接口定义 (契约)
+
+此接口定义了面向业务操作的方法，将所有实现细节（包括解析）完全封装。
+"""
+
+from abc import ABC, abstractmethod
+
+class IBusManager(ABC):
+    """
+    总线管理器接口。
+    调用方只关心业务，不关心实现。
+    """
+
+    @abstractmethod
+    def execute_raw_command(self, bus_id: int, command: bytes) -> None:
+        """
+        【契约】执行一个“发后不理”的原始指令。
+        
+        Args:
+            bus_id (int): 目标总线的编号。
+            command (bytes): 要发送的原始命令字节。
+        """
+        pass
+
+    @abstractmethod
+    def execute_collect_task(self, task: dict) -> float | None:
+        """
+        【契约】执行一个完整的采集任务，并直接返回最终的数值。
+
+        一个符合本接口的实现必须自己处理所有细节：
+        - 从task字典中解析出 bus_id, command, parser_type。
+        - 发送指令。
+        - 接收响应。
+        - 根据parser_type选择正确的内部解析器进行解析。
+        - 返回最终的float数值，或在任何失败情况下返回None。
+
+        Args:
+            task (dict): 从Protobuf解码出的单个CollectTask消息字典。
+
+        Returns:
+            float | None: 成功解析则返回数值，否则返回None。
+        """
+        pass

main/config.py

@@ -0,0 +1,81 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+
+"""
+项目全局配置文件
+
+集中管理所有硬件引脚、通信参数和软件配置，
+便于统一修改和适配不同的硬件版本。
+"""
+
+# --- LoRa 模块配置 ---
+# 假设LoRa模块使用独立的UART进行通信
+LORA_CONFIG = {
+    # LoRa模块连接的UART总线ID (0, 1, or 2 on ESP32)
+    'uart_id': 2,
+
+    # LoRa模块的通信波特率
+    'baudrate': 9600,
+
+    # LoRa模块连接的GPIO引脚
+    'pins': {
+        'tx': 17,  # UART TX
+        'rx': 16,  # UART RX
+    }
+}
+
+# --- 总线配置 ---
+# 使用字典来定义项目中的所有通信总线
+# key是总线ID (bus_id)，value是该总线的详细配置字典。
+# 这种结构使得 command_processor 可以通过 bus_id 动态获取其配置。
+BUS_CONFIG = {
+    # --- 总线 1 ---
+    1: {
+        # 总线协议类型，用于程序动态选择不同的处理逻辑
+        'protocol': 'RS485',
+
+        # 该总线使用的硬件UART ID
+        'uart_id': 1,
+
+        # 该总线的通信波特率
+        'baudrate': 9600,
+
+        # 该总线使用的GPIO引脚
+        'pins': {
+            'tx': 4,  # RS485 TX
+            'rx': 5,  # RS485 RX
+            'rts': 2,  # RS485 DE/RE 方向控制引脚
+        }
+    },
+
+    # 如果未来有第二条总线，或不同协议的总线，可以直接在这里添加
+    2: {
+        'protocol': 'RS485',
+        'uart_id': 0,
+        'baudrate': 19200,  # 这条总线可以有不同的波特率
+        'pins': {
+            'tx': 25,
+            'rx': 26,
+            'rts': 27,
+        }
+    },
+}
+
+# --- 全局超时设置 (毫秒) ---
+DEFAULT_TIMEOUTS = {
+    'rs485_response': 500,  # 等待RS485设备响应的默认超时时间
+    'lora_at_command': 300,  # 等待LoRa模块AT指令响应的超时时间
+}
+
+# --- 系统参数配置 ---
+SYSTEM_PARAMS = {
+    # 任务队列的最大长度。用于主线程和工作线程之间的缓冲。
+    # 如果LoRa指令瞬间并发量大，可以适当调高此值。
+    # 如果内存紧张，可以适当调低。
+    'task_queue_max_size': 10,
+
+    # 全局调试日志开关
+    # True:  所有 logger.log() 的信息都会被打印到串口。
+    # False: logger.log() 将不执行任何操作，用于发布产品。
+    'debug_enabled': True,
+}

main/logger.py

@@ -0,0 +1,22 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+
+"""
+一个简单的、可配置的日志记录器模块。
+"""
+
+import config
+
+def log(message: str):
+    """
+    打印一条日志消息，是否实际输出取决于配置文件。
+
+    Args:
+        message (str): 要打印的日志消息。
+    """
+    # 从配置文件中获取调试开关的状态
+    # .get()方法可以安全地获取值，如果键不存在，则返回默认值False
+    if config.SYSTEM_PARAMS.get('debug_enabled', False):
+        print(message)
+    
+    # 如果开关为False，此函数会立即返回，不执行任何操作。

main/lora/lora_interface.py

@@ -0,0 +1,53 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+
+"""
+LoRa通信模块的抽象接口定义 (契约)
+
+这个文件定义了一个LoRa处理器应该具备哪些功能，
+但不包含任何具体的实现代码。任何具体的LoRa处理器，
+无论是UART的还是SPI的，都必须实现这里定义的所有方法。
+"""
+
+# abc (Abstract Base Class) 是Python定义接口的标准方式
+from abc import ABC, abstractmethod
+
+class ILoraHandler(ABC):
+    """
+    LoRa处理器接口。
+    它规定了所有LoRa处理器实现类必须提供的功能。
+    """
+
+    @abstractmethod
+    def receive_packet(self):
+        """
+        【契约】非阻塞地检查并接收一个数据包。
+        
+        一个符合本接口的实现必须:
+        - 检查是否有新的数据包。
+        - 如果有，读取、解析并返回负载数据。
+        - 如果没有，必须立刻返回None，不得阻塞。
+
+        Returns:
+            bytes: 如果成功接收到一个数据包，返回该数据包的字节。
+            None:  如果当前没有可读的数据包。
+        """
+        pass
+
+    @abstractmethod
+    def send_packet(self, data_bytes: bytes) -> bool:
+        """
+        【契约】发送一个数据包。
+
+        一个符合本接口的实现必须:
+        - 接收一个bytes类型的参数。
+        - 将这些数据通过LoRa模块发送出去。
+        - 返回一个布尔值表示发送指令是否成功提交。
+        
+        Args:
+            data_bytes (bytes): 需要发送的字节数据。
+        
+        Returns:
+            bool: True表示发送指令已成功提交，False表示因任何原因失败。
+        """
+        pass

main/main.py

@@ -0,0 +1,90 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+
+"""
+程序主入口 (双线程生产者-消费者模型)
+
+主线程 (生产者):
+- 职责：以最高优先级不间断监听LoRa数据，并将数据包放入任务队列。
+- 特点：永远不执行耗时操作，保证LoRa数据接收的实时性。
+
+工作线程 (消费者):
+- 职责：从任务队列中取出数据包，并进行耗时的业务处理。
+- 特点：可能会长时间阻塞，但不影响主线程的数据接收。
+"""
+
+import time
+import _thread
+import config
+import uqueue  # 导入我们自己创建的本地uqueue模块
+
+# 导入接口和实现
+from lora.lora_interface import ILoraHandler
+from bus.bus_interface import IBusManager
+from lora.lora_handler import LoRaHandler
+from bus.rs485_manager import RS485Manager
+from processor import Processor
+
+# 导入工作线程的执行函数
+from worker import worker_task
+from logger import log
+
+# --- 模块级变量定义 (带有类型提示) ---
+lora_controller: ILoraHandler | None = None
+bus_manager: IBusManager | None = None
+processor: Processor | None = None
+task_queue: uqueue.Queue | None = None
+
+
+def setup():
+    """
+    初始化函数，负责创建所有对象实例、共享队列，并启动工作线程。
+    """
+    global lora_controller, bus_manager, processor, task_queue
+
+    log("--- 系统初始化开始 ---")
+
+    # 1. 初始化硬件驱动和业务处理器
+    lora_controller = LoRaHandler()
+    bus_manager = RS485Manager()
+    processor = Processor(lora_handler=lora_controller, bus_manager=bus_manager)
+
+    # 2. 从配置文件读取队列长度，并创建线程安全的队列
+    queue_size = config.SYSTEM_PARAMS.get('task_queue_max_size', 10)
+    task_queue = uqueue.Queue(maxsize=queue_size)
+    log(f"任务队列已创建，最大容量: {queue_size}")
+
+    # 3. 启动工作线程
+    _thread.start_new_thread(worker_task, (task_queue, processor))
+
+    log("--- 系统初始化完成 ---")
+
+
+def loop():
+    """
+    主线程循环函数 (生产者)。
+    只负责监听LoRa，并将数据放入队列。
+    """
+    packet = lora_controller.receive_packet()
+
+    if packet:
+        if task_queue.full():
+            log("警告：任务队列已满，新的LoRa数据包被丢弃！")
+            return
+
+        try:
+            task_queue.put_nowait(packet)
+            log(f"主线程：新LoRa数据包已入队。当前队列大小: {task_queue.qsize()}")
+        except Exception as e:
+            log(f"错误：数据包入队失败: {e}")
+
+    time.sleep_ms(10)
+
+
+# --- 程序主执行区 ---
+if __name__ == "__main__":
+    setup()
+
+    log("--- 主线程进入循环 (LoRa监听) ---")
+    while True:
+        loop()

main/processor.py

@@ -0,0 +1,111 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+
+"""
+核心业务逻辑处理器 (V3 - 面向业务接口)
+
+职责:
+- 编排业务流程：解码指令，并将业务任务分发给相应的管理器。
+- 完全不关心总线通信和数据解析的技术实现细节。
+"""
+
+# 导入我们定义的“契约”（接口）
+from lora.lora_interface import ILoraHandler
+from bus.bus_interface import IBusManager
+
+# 导入Protobuf解析代码
+from proto import client_pb
+
+from logger import log
+
+
+class Processor:
+    """
+    命令处理器类，项目的“大脑”。
+    它依赖于抽象的、面向业务的接口。
+    """
+
+    def __init__(self, lora_handler: ILoraHandler, bus_manager: IBusManager):
+        """
+        构造函数 (依赖注入)。
+        
+        Args:
+            lora_handler (ILoraHandler): 一个实现了LoRa接口的对象。
+            bus_manager (IBusManager): 一个实现了总线接口的对象。
+        """
+        self.lora = lora_handler
+        self.bus = bus_manager
+        log("业务处理器已初始化，准备就绪。")
+
+    def handle_packet(self, packet_bytes: bytes):
+        """
+        处理单个LoRa数据包的入口函数。
+        """
+        log(f"收到待处理数据包: {packet_bytes.hex()}")
+
+        try:
+            instruction = client_pb.decode_instruction(packet_bytes)
+        except Exception as e:
+            log(f"错误：解码指令失败: {e}")
+            return
+
+        # 根据指令类型，分发到不同的业务处理方法
+        if 'raw_485_command' in instruction:
+            self._process_exec_command(instruction['raw_485_command'])
+
+        elif 'batch_collect_command' in instruction:
+            self._process_collect_command(instruction['batch_collect_command'])
+
+        else:
+            log(f"警告：收到未知或不适用于此设备的指令类型: {instruction}")
+
+    def _process_exec_command(self, cmd: dict):
+        """
+        处理“执行命令”业务。
+        """
+        bus_id = cmd['bus_number']
+        command_bytes = cmd['command_bytes']
+        log(f"处理[执行命令]业务：向总线 {bus_id} 下发指令。")
+
+        # 直接调用总线接口的业务方法，不关心实现
+        self.bus.execute_raw_command(bus_id, command_bytes)
+        log("执行指令已下发。")
+
+    def _process_collect_command(self, cmd: dict):
+        """
+        处理“采集命令”业务。
+        """
+        correlation_id = cmd['correlation_id']
+        tasks = cmd['tasks']
+        log(f"处理[采集命令]业务 (ID: {correlation_id})：共 {len(tasks)} 个任务。")
+
+        sensor_values = []
+        for i, task in enumerate(tasks):
+            log(f"  - 执行任务 {i + 1}...")
+
+            # 调用总线接口的业务方法，直接获取最终结果
+            # 我们不再关心task的具体内容，也不关心解析过程
+            value = self.bus.execute_collect_task(task)
+
+            if value is not None:
+                sensor_values.append(value)
+                log(f"    => 成功，获取值为: {value}")
+            else:
+                # 如果返回None，表示任务失败（超时或解析错误）
+                sensor_values.append(-1)  # 添加一个默认/错误值
+                log("    => 失败，任务未返回有效值。")
+
+        # 所有任务执行完毕，构建并发送响应
+        log(f"所有采集任务完成，准备发送响应。采集到的值: {sensor_values}")
+        try:
+            response_payload = {
+                'correlation_id': correlation_id,
+                'values': sensor_values
+            }
+            response_packet = client_pb.encode_instruction('collect_result', response_payload)
+
+            # 通过LoRa接口发送出去
+            self.lora.send_packet(response_packet)
+            log("采集结果已通过LoRa发送。")
+        except Exception as e:
+            log(f"错误：编码或发送采集结果失败: {e}")

main/proto/client.proto

@@ -0,0 +1,51 @@
+syntax = "proto3";
+
+package device;
+
+// import "google/protobuf/any.proto"; // REMOVED: Not suitable for embedded systems.
+
+option go_package = "internal/domain/device/proto";
+
+// --- Concrete Command & Data Structures ---
+
+// 平台生成的原始485指令，单片机直接发送到总线
+message Raw485Command {
+    int32 bus_number = 1;    // 总线号，用于指示单片机将指令发送到哪个总线
+    bytes command_bytes = 2; // 原始485指令的字节数组
+}
+
+// BatchCollectCommand
+// 一个完整的、包含所有元数据的批量采集任务。
+message BatchCollectCommand {
+  string correlation_id = 1;        // 用于关联请求和响应的唯一ID
+  repeated CollectTask tasks = 2;   // 采集任务列表
+}
+
+// CollectTask
+// 定义了单个采集任务的“意图”。
+message CollectTask {
+    Raw485Command command = 2; // 平台生成的原始485指令
+}
+
+// CollectResult
+// 这是设备响应的、极致精简的数据包。
+message CollectResult {
+    string correlation_id = 1;   // 从下行指令中原样返回的关联ID
+    repeated float values = 2;   // 按预定顺序排列的采集值
+}
+
+
+// --- Main Downlink Instruction Wrapper ---
+
+// 指令 (所有从平台下发到设备的数据都应该被包装在这里面)
+// 使用 oneof 来替代 google.protobuf.Any，这是嵌入式环境下的标准做法。
+// 它高效、类型安全，且只解码一次。
+message Instruction {
+    oneof payload {
+        Raw485Command raw_485_command = 1;
+        BatchCollectCommand batch_collect_command = 2;
+        CollectResult collect_result = 3; // ADDED：用于上行数据
+    // 如果未来有其他指令类型，比如开关控制，可以直接在这里添加
+    // SwitchCommand switch_command = 3;
+    }
+}

main/proto/client_pb.py

@@ -0,0 +1,378 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+
+"""
+根据client.proto生成的解析代码
+适用于ESP32 MicroPython环境
+"""
+
+import struct
+
+# --- Protobuf基础类型辅助函数 ---
+
+def encode_varint(value):
+    """编码varint整数"""
+    buf = bytearray()
+    while value >= 0x80:
+        buf.append((value & 0x7F) | 0x80)
+        value >>= 7
+    buf.append(value & 0x7F)
+    return buf
+
+def decode_varint(buf, pos=0):
+    """解码varint整数"""
+    result = 0
+    shift = 0
+    while pos < len(buf):
+        byte = buf[pos]
+        pos += 1
+        result |= (byte & 0x7F) << shift
+        if not (byte & 0x80):
+            break
+        shift += 7
+    return result, pos
+
+def encode_string(value):
+    """编码字符串"""
+    value_bytes = value.encode('utf-8')
+    length = encode_varint(len(value_bytes))
+    return length + value_bytes
+
+def decode_string(buf, pos=0):
+    """解码字符串"""
+    length, pos = decode_varint(buf, pos)
+    value = buf[pos:pos+length].decode('utf-8')
+    pos += length
+    return value, pos
+
+# --- 消息编码/解码函数 ---
+
+def encode_raw_485_command(bus_number, command_bytes):
+    """
+    编码Raw485Command消息
+    Args:
+        bus_number (int): 总线号
+        command_bytes (bytes): 原始485指令
+    Returns:
+        bytearray: 编码后的数据
+    """
+    result = bytearray()
+    # bus_number (field 1, wire type 0)
+    result.extend(encode_varint((1 << 3) | 0))
+    result.extend(encode_varint(bus_number))
+    # command_bytes (field 2, wire type 2)
+    result.extend(encode_varint((2 << 3) | 2))
+    result.extend(encode_varint(len(command_bytes)))
+    result.extend(command_bytes)
+    return result
+
+def decode_raw_485_command(buf):
+    """
+    解码Raw485Command消息
+    Args:
+        buf (bytes): 编码后的数据
+    Returns:
+        dict: 解码后的消息
+    """
+    result = {}
+    pos = 0
+    while pos < len(buf):
+        tag, pos = decode_varint(buf, pos)
+        field_number = tag >> 3
+        wire_type = tag & 0x07
+
+        if field_number == 1:  # bus_number
+            if wire_type == 0:
+                value, pos = decode_varint(buf, pos)
+                result['bus_number'] = value
+        elif field_number == 2:  # command_bytes
+            if wire_type == 2:
+                length, pos = decode_varint(buf, pos)
+                value = buf[pos:pos+length]
+                pos += length
+                result['command_bytes'] = value
+        else:
+            # 跳过未知字段
+            if wire_type == 0: _, pos = decode_varint(buf, pos)
+            elif wire_type == 2: length, pos = decode_varint(buf, pos); pos += length
+            else: pos += 1
+    return result
+
+def encode_collect_task(command_msg):
+    """
+    编码CollectTask消息
+    Args:
+        command_msg (dict): Raw485Command消息字典
+    Returns:
+        bytearray: 编码后的数据
+    """
+    result = bytearray()
+    # command (field 2, wire type 2)
+    encoded_command = encode_raw_485_command(command_msg['bus_number'], command_msg['command_bytes'])
+    result.extend(encode_varint((2 << 3) | 2))
+    result.extend(encode_varint(len(encoded_command)))
+    result.extend(encoded_command)
+    return result
+
+def decode_collect_task(buf):
+    """
+    解码CollectTask消息
+    Args:
+        buf (bytes): 编码后的数据
+    Returns:
+        dict: 解码后的消息
+    """
+    result = {}
+    pos = 0
+    while pos < len(buf):
+        tag, pos = decode_varint(buf, pos)
+        field_number = tag >> 3
+        wire_type = tag & 0x07
+
+        if field_number == 2:  # command
+            if wire_type == 2:
+                length, pos = decode_varint(buf, pos)
+                value_buf = buf[pos:pos+length]
+                pos += length
+                result['command'] = decode_raw_485_command(value_buf)
+        else:
+            if wire_type == 0: _, pos = decode_varint(buf, pos)
+            elif wire_type == 2: length, pos = decode_varint(buf, pos); pos += length
+            else: pos += 1
+    return result
+
+def encode_batch_collect_command(correlation_id, tasks):
+    """
+    编码BatchCollectCommand消息
+    Args:
+        correlation_id (str): 关联ID
+        tasks (list): CollectTask消息字典列表
+    Returns:
+        bytearray: 编码后的数据
+    """
+    result = bytearray()
+    # correlation_id (field 1, wire type 2)
+    result.extend(encode_varint((1 << 3) | 2))
+    result.extend(encode_string(correlation_id))
+    # tasks (field 2, wire type 2) - repeated
+    for task in tasks:
+        encoded_task = encode_collect_task(task['command'])
+        result.extend(encode_varint((2 << 3) | 2))
+        result.extend(encode_varint(len(encoded_task)))
+        result.extend(encoded_task)
+    return result
+
+def decode_batch_collect_command(buf):
+    """
+    解码BatchCollectCommand消息
+    Args:
+        buf (bytes): 编码后的数据
+    Returns:
+        dict: 解码后的消息
+    """
+    result = {'tasks': []}
+    pos = 0
+    while pos < len(buf):
+        tag, pos = decode_varint(buf, pos)
+        field_number = tag >> 3
+        wire_type = tag & 0x07
+
+        if field_number == 1:  # correlation_id
+            if wire_type == 2:
+                value, pos = decode_string(buf, pos)
+                result['correlation_id'] = value
+        elif field_number == 2:  # tasks (repeated)
+            if wire_type == 2:
+                length, pos = decode_varint(buf, pos)
+                value_buf = buf[pos:pos+length]
+                pos += length
+                result['tasks'].append(decode_collect_task(value_buf))
+        else:
+            if wire_type == 0: _, pos = decode_varint(buf, pos)
+            elif wire_type == 2: length, pos = decode_varint(buf, pos); pos += length
+            else: pos += 1
+    return result
+
+def encode_collect_result(correlation_id, values):
+    """
+    编码CollectResult消息
+    Args:
+        correlation_id (str): 关联ID
+        values (list): 采集值列表 (float)
+    Returns:
+        bytearray: 编码后的数据
+    """
+    result = bytearray()
+    # correlation_id (field 1, wire type 2)
+    result.extend(encode_varint((1 << 3) | 2))
+    result.extend(encode_string(correlation_id))
+    # values (field 2, wire type 5) - repeated fixed32
+    for value in values:
+        result.extend(encode_varint((2 << 3) | 5))  # Tag for fixed32
+        result.extend(struct.pack('<f', value))     # 小端序浮点数
+    return result
+
+def decode_collect_result(buf):
+    """
+    解码CollectResult消息
+    Args:
+        buf (bytes): 编码后的数据
+    Returns:
+        dict: 解码后的消息
+    """
+    result = {'values': []}
+    pos = 0
+    while pos < len(buf):
+        tag, pos = decode_varint(buf, pos)
+        field_number = tag >> 3
+        wire_type = tag & 0x07
+
+        if field_number == 1:  # correlation_id
+            if wire_type == 2:
+                value, pos = decode_string(buf, pos)
+                result['correlation_id'] = value
+        elif field_number == 2:  # values (repeated)
+            if wire_type == 5:  # fixed32
+                value = struct.unpack('<f', buf[pos:pos+4])[0]
+                pos += 4
+                result['values'].append(value)
+        else:
+            if wire_type == 0: _, pos = decode_varint(buf, pos)
+            elif wire_type == 5: pos += 4  # fixed32
+            elif wire_type == 2: length, pos = decode_varint(buf, pos); pos += length
+            else: pos += 1
+    return result
+
+def encode_instruction(payload_type, payload_data):
+    """
+    编码Instruction消息 (包含oneof字段)
+    Args:
+        payload_type (str): oneof字段的类型 ('raw_485_command', 'batch_collect_command', 'collect_result')
+        payload_data (dict): 对应类型的消息字典
+    Returns:
+        bytearray: 编码后的数据
+    """
+    result = bytearray()
+    encoded_payload = bytearray()
+
+    if payload_type == 'raw_485_command':
+        encoded_payload = encode_raw_485_command(payload_data['bus_number'], payload_data['command_bytes'])
+        result.extend(encode_varint((1 << 3) | 2))  # field 1, wire type 2
+    elif payload_type == 'batch_collect_command':
+        encoded_payload = encode_batch_collect_command(payload_data['correlation_id'], payload_data['tasks'])
+        result.extend(encode_varint((2 << 3) | 2))  # field 2, wire type 2
+    elif payload_type == 'collect_result':
+        encoded_payload = encode_collect_result(payload_data['correlation_id'], payload_data['values'])
+        result.extend(encode_varint((3 << 3) | 2))  # field 3, wire type 2
+    else:
+        raise ValueError("未知的指令负载类型")
+
+    result.extend(encode_varint(len(encoded_payload)))
+    result.extend(encoded_payload)
+    return result
+
+def decode_instruction(buf):
+    """
+    解码Instruction消息
+    Args:
+        buf (bytes): 编码后的数据
+    Returns:
+        dict: 解码后的消息
+    """
+    result = {}
+    pos = 0
+    while pos < len(buf):
+        tag, pos = decode_varint(buf, pos)
+        field_number = tag >> 3
+        wire_type = tag & 0x07
+
+        if wire_type == 2:  # 所有oneof字段都使用长度分隔类型
+            length, pos = decode_varint(buf, pos)
+            value_buf = buf[pos:pos+length]
+            pos += length
+
+            if field_number == 1:  # raw_485_command
+                result['raw_485_command'] = decode_raw_485_command(value_buf)
+            elif field_number == 2:  # batch_collect_command
+                result['batch_collect_command'] = decode_batch_collect_command(value_buf)
+            elif field_number == 3:  # collect_result
+                result['collect_result'] = decode_collect_result(value_buf)
+        else:
+            # 跳过未知字段
+            if wire_type == 0: _, pos = decode_varint(buf, pos)
+            elif wire_type == 5: pos += 4
+            elif wire_type == 2: length, pos = decode_varint(buf, pos); pos += length
+            else: pos += 1
+    return result
+
+# --- 单元测试与使用范例 ---
+if __name__ == "__main__":
+    print("--- 测试 Raw485Command ---")
+    raw_cmd_data = {'bus_number': 1, 'command_bytes': b'\x01\x03\x00\x00\x00\x02\xc4\x0b'}
+    encoded_raw_cmd = encode_raw_485_command(raw_cmd_data['bus_number'], raw_cmd_data['command_bytes'])
+    print(f"编码后 Raw485Command: {encoded_raw_cmd.hex()}")
+    decoded_raw_cmd = decode_raw_485_command(encoded_raw_cmd)
+    print(f"解码后 Raw485Command: {decoded_raw_cmd}")
+    assert decoded_raw_cmd == raw_cmd_data
+
+    print("\n--- 测试 CollectTask ---")
+    collect_task_data = {'command': raw_cmd_data}
+    encoded_collect_task = encode_collect_task(collect_task_data['command'])
+    print(f"编码后 CollectTask: {encoded_collect_task.hex()}")
+    decoded_collect_task = decode_collect_task(encoded_collect_task)
+    print(f"解码后 CollectTask: {decoded_collect_task}")
+    assert decoded_collect_task == collect_task_data
+
+    print("\n--- 测试 BatchCollectCommand ---")
+    batch_collect_data = {
+        'correlation_id': 'abc-123',
+        'tasks': [
+            {'command': {'bus_number': 1, 'command_bytes': b'\x01\x03\x00\x00\x00\x02\xc4\x0b'}},
+            {'command': {'bus_number': 2, 'command_bytes': b'\x02\x03\x00\x01\x00\x01\xd5\xfa'}}
+        ]
+    }
+    encoded_batch_collect = encode_batch_collect_command(batch_collect_data['correlation_id'], batch_collect_data['tasks'])
+    print(f"编码后 BatchCollectCommand: {encoded_batch_collect.hex()}")
+    decoded_batch_collect = decode_batch_collect_command(encoded_batch_collect)
+    print(f"解码后 BatchCollectCommand: {decoded_batch_collect}")
+    assert decoded_batch_collect == batch_collect_data
+
+    print("\n--- 测试 CollectResult ---")
+    collect_result_data = {
+        'correlation_id': 'res-456',
+        'values': [12.34, 56.78, 90.12]
+    }
+    encoded_collect_result = encode_collect_result(collect_result_data['correlation_id'], collect_result_data['values'])
+    print(f"编码后 CollectResult: {encoded_collect_result.hex()}")
+    decoded_collect_result = decode_collect_result(encoded_collect_result)
+    print(f"解码后 CollectResult: {decoded_collect_result}")
+    # 由于32位浮点数精度问题，直接比较可能会失败，此处设置一个合理的容忍度
+    assert decoded_collect_result['correlation_id'] == collect_result_data['correlation_id']
+    for i in range(len(collect_result_data['values'])):
+        assert abs(decoded_collect_result['values'][i] - collect_result_data['values'][i]) < 1e-5 # 已放宽容忍度
+
+    print("\n--- 测试 Instruction (内含Raw485Command) ---")
+    instruction_raw_485 = encode_instruction('raw_485_command', raw_cmd_data)
+    print(f"编码后 Instruction (Raw485Command): {instruction_raw_485.hex()}")
+    decoded_instruction_raw_485 = decode_instruction(instruction_raw_485)
+    print(f"解码后 Instruction (Raw485Command): {decoded_instruction_raw_485}")
+    assert decoded_instruction_raw_485['raw_485_command'] == raw_cmd_data
+
+    print("\n--- 测试 Instruction (内含BatchCollectCommand) ---")
+    instruction_batch_collect = encode_instruction('batch_collect_command', batch_collect_data)
+    print(f"编码后 Instruction (BatchCollectCommand): {instruction_batch_collect.hex()}")
+    decoded_instruction_batch_collect = decode_instruction(instruction_batch_collect)
+    print(f"解码后 Instruction (BatchCollectCommand): {decoded_instruction_batch_collect}")
+    assert decoded_instruction_batch_collect['batch_collect_command']['correlation_id'] == batch_collect_data['correlation_id']
+    assert len(decoded_instruction_batch_collect['batch_collect_command']['tasks']) == len(batch_collect_data['tasks'])
+
+    print("\n--- 测试 Instruction (内含CollectResult) ---")
+    instruction_collect_result = encode_instruction('collect_result', collect_result_data)
+    print(f"编码后 Instruction (CollectResult): {instruction_collect_result.hex()}")
+    decoded_instruction_collect_result = decode_instruction(instruction_collect_result)
+    print(f"解码后 Instruction (CollectResult): {decoded_instruction_collect_result}")
+    assert decoded_instruction_collect_result['collect_result']['correlation_id'] == collect_result_data['correlation_id']
+    for i in range(len(collect_result_data['values'])):
+        assert abs(decoded_instruction_collect_result['collect_result']['values'][i] - collect_result_data['values'][i]) < 1e-5
+
+    print("\n所有测试均已通过!")

main/uqueue.py

@@ -0,0 +1,81 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+
+"""
+一个适用于MicroPython的、简单的线程安全队列实现。
+
+这个模块提供了一个与标准库 `queue.Queue` 类似的类，
+确保在多线程环境下的数据操作是安全的。
+"""
+
+import _thread
+from collections import deque
+
+class Queue:
+    """一个简单的、线程安全的队列。"""
+
+    def __init__(self, maxsize=0):
+        self.maxsize = maxsize
+        self._lock = _thread.allocate_lock()
+        # 使用deque可以实现更高效的头部弹出操作 (O(1))
+        self._items = deque((), maxsize if maxsize > 0 else 1024)
+
+    def qsize(self):
+        """返回队列中的项目数。"""
+        with self._lock:
+            return len(self._items)
+
+    def empty(self):
+        """如果队列为空，返回True，否则返回False。"""
+        return self.qsize() == 0
+
+    def full(self):
+        """如果队列已满，返回True，否则返回False。"""
+        if self.maxsize <= 0:
+            return False
+        return self.qsize() >= self.maxsize
+
+    def put(self, item, block=True, timeout=None):
+        """将一个项目放入队列。"""
+        if not block:
+            return self.put_nowait(item)
+        
+        # 阻塞式put的简单实现 (在实际应用中更复杂的实现会使用信号量)
+        while True:
+            with self._lock:
+                if not self.full():
+                    self._items.append(item)
+                    return
+            # 如果队列是满的，短暂休眠后重试
+            import time
+            time.sleep_ms(5)
+
+    def put_nowait(self, item):
+        """等同于 put(item, block=False)。"""
+        if self.full():
+            raise OSError("Queue full")
+        
+        with self._lock:
+            self._items.append(item)
+
+    def get(self, block=True, timeout=None):
+        """从队列中移除并返回一个项目。"""
+        if not block:
+            return self.get_nowait()
+
+        # 阻塞式get的简单实现
+        while True:
+            with self._lock:
+                if self._items:
+                    return self._items.popleft()
+            # 如果队列是空的，短暂休眠后重试
+            import time
+            time.sleep_ms(5)
+
+    def get_nowait(self):
+        """等同于 get(item, block=False)。"""
+        if self.empty():
+            raise OSError("Queue empty")
+            
+        with self._lock:
+            return self._items.popleft()

main/worker.py

@@ -0,0 +1,44 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+
+"""
+工作线程模块
+
+职责:
+- 作为一个独立的线程运行。
+- 阻塞式地等待任务队列。
+- 从队列中取出任务，并交给Processor进行耗时处理。
+"""
+
+import uqueue
+from processor import Processor
+from logger import log
+
+
+def worker_task(task_queue: uqueue.Queue, processor: Processor):
+    """
+    工作线程的主函数。
+    
+    Args:
+        task_queue (uqueue.Queue): 共享的任务队列。
+        processor (Processor): 业务处理器实例。
+    """
+    log("工作线程已启动，等待任务...")
+    
+    while True:
+        try:
+            # 1. 阻塞式地从队列中获取任务
+            # 如果队列为空，程序会在这里自动挂起，不消耗CPU
+            # get()方法是线程安全的
+            packet_bytes = task_queue.get()
+            
+            log(f"工作线程：收到新任务，开始处理... 数据: {packet_bytes.hex()}")
+            
+            # 2. 调用processor进行耗时的、阻塞式的处理
+            # 这个处理过程不会影响主线程的LoRa监听
+            processor.handle_packet(packet_bytes)
+            
+            log("工作线程：任务处理完毕，继续等待下一个任务。")
+            
+        except Exception as e:
+            log(f"错误：工作线程在处理任务时发生异常: {e}")

proto/client.proto

@@ -1,32 +0,0 @@
-syntax = "proto3";
-
-package device;
-
-import "google/protobuf/any.proto";
-
-option go_package = "internal/app/service/device/proto";
-
-// 指令类型
-enum MethodType{
-  SWITCH = 0; // 启停
-  COLLECT = 1; // 采集
-}
-
-// 指令
-message Instruction{
-  MethodType method = 1;
-  google.protobuf.Any data = 2;
-}
-
-message Switch{
-  string device_action = 1; // 指令
-  int32 bus_number = 2; // 总线号
-  int32 bus_address = 3; // 总线地址
-  int32 relay_channel = 4; // 继电器通道号
-}
-
-message Collect{
-  int32 bus_number = 1; // 总线号
-  int32 bus_address = 2; // 总线地址
-  float value = 3; // 采集值
-}