树莓派Pico运行DHT11代码时出现InvalidChecksum导入错误

问题描述 #

树莓派Pico与DHT11传感器接线如下：

• 传感器3V3引脚 → Pico 3V3引脚
• 传感器GND引脚 → Pico GND引脚
• 传感器DATA引脚 → Pico G16引脚

在Thonny MicroPython环境中运行代码时，出现错误：ImportError: can't import name InvalidChecksum

涉及代码文件如下：

main temp_hum_DHT_code.py

from machine import Pin
import time
from dht import DHT11, InvalidChecksum
 
sensor = DHT11(Pin(16, Pin.OUT, Pin.PULL_DOWN))
 
while True:
    temp = sensor.temperature
    humidity = sensor.humidity
    print("Temperature: {}°C   Humidity: {:.0f}% ".format(temp, humidity))
    time.sleep(2)

dht.py

import array

import micropython
import utime
from machine import Pin
from micropython import const


class InvalidChecksum(Exception):
    pass


class InvalidPulseCount(Exception):
    pass


MAX_UNCHANGED = const(100)
MIN_INTERVAL_US = const(200000)
HIGH_LEVEL = const(50)
EXPECTED_PULSES = const(84)


class DHT11:
    _temperature: float
    _humidity: float

    def __init__(self, pin):
        self._pin = pin
        self._last_measure = utime.ticks_us()
        self._temperature = -1
        self._humidity = -1

    def measure(self):
        current_ticks = utime.ticks_us()
        if utime.ticks_diff(current_ticks, self._last_measure) < MIN_INTERVAL_US and (
            self._temperature > -1 or self._humidity > -1
        ):
            # Less than a second since last read, which is too soon according
            # to the datasheet
            return

        self._send_init_signal()
        pulses = self._capture_pulses()
        buffer = self._convert_pulses_to_buffer(pulses)
        self._verify_checksum(buffer)

        self._humidity = buffer[0] + buffer[1] / 10
        self._temperature = buffer[2] + buffer[3] / 10
        self._last_measure = utime.ticks_us()

    @property
    def humidity(self):
        self.measure()
        return self._humidity

    @property
    def temperature(self):
        self.measure()
        return self._temperature

    def _send_init_signal(self):
        self._pin.init(Pin.OUT, Pin.PULL_DOWN)
        self._pin.value(1)
        utime.sleep_ms(50)
        self._pin.value(0)
        utime.sleep_ms(18)

    @micropython.native
    def _capture_pulses(self):
        pin = self._pin
        pin.init(Pin.IN, Pin.PULL_UP)

        val = 1
        idx = 0
        transitions = bytearray(EXPECTED_PULSES)
        unchanged = 0
        timestamp = utime.ticks_us()

        while unchanged < MAX_UNCHANGED:
            if val != pin.value():
                if idx >= EXPECTED_PULSES:
                    raise InvalidPulseCount(
                        "Got more than {} pulses".format(EXPECTED_PULSES)
                    )
                now = utime.ticks_us()
                transitions[idx] = now - timestamp
                timestamp = now
                idx += 1

                val = 1 - val
                unchanged = 0
            else:
                unchanged += 1
        pin.init(Pin.OUT, Pin.PULL_DOWN)
        if idx != EXPECTED_PULSES:
            raise InvalidPulseCount(
                "Expected {} but got {} pulses".format(EXPECTED_PULSES, idx)
            )
        return transitions[4:]

    def _convert_pulses_to_buffer(self, pulses):
        """Convert a list of 80 pulses into a 5 byte buffer

        The resulting 5 bytes in the buffer will be:
            0: Integral relative humidity data
            1: Decimal relative humidity data
            2: Integral temperature data
            3: Decimal temperature data
            4: Checksum
        """
        # Convert the pulses to 40 bits
        binary = 0
        for idx in range(0, len(pulses), 2):
            binary = binary << 1 | int(pulses[idx] > HIGH_LEVEL)

        # Split into 5 bytes
        buffer = array.array("B")
        for shift in range(4, -1, -1):
            buffer.append(binary >> shift * 8 & 0xFF)
        return buffer

    def _verify_checksum(self, buffer):
        # Calculate checksum
        checksum = 0
        for buf in buffer[0:4]:
            checksum += buf
        if checksum & 0xFF != buffer[4]:
            raise InvalidChecksum()

错误原因 #

MicroPython官方固件已内置dht模块，执行from dht import ...时，系统优先加载内置模块而非自定义的dht.py。内置dht模块未定义InvalidChecksum异常类，因此触发导入错误。

解决步骤 #

1. 重命名自定义模块：将dht.py重命名为dht_custom.py（或其他不与内置模块重名的名称）。
2. 修改主代码导入语句：

from machine import Pin
import time
from dht_custom import DHT11, InvalidChecksum  # 替换为新模块名

sensor = DHT11(Pin(16, Pin.OUT, Pin.PULL_DOWN))

while True:
    try:
        temp = sensor.temperature
        humidity = sensor.humidity
        print("Temperature: {}°C   Humidity: {:.0f}% ".format(temp, humidity))
    except InvalidChecksum:
        print("数据校验失败，正在重新读取...")
    except Exception as e:
        print(f"读取错误：{str(e)}")
    time.sleep(2)

3. 添加异常捕获：在主循环中加入异常处理，可有效处理传感器读取时的校验错误或其他异常，提升程序稳定性。

补充说明 #

若不想重命名文件，也可将自定义dht.py放入子目录，通过相对路径导入，但重命名是最直接简便的解决方案。

内容的提问来源于stack exchange，提问作者captain2001