DHT Library for ESP

[PLUB]

Hi,
I am using Micropython on an ESP32 and want to understand how add-on libraries work.
I started with the HD44780 library, and now I'm trying the DHT11 / 22 library. I found several but they are either for Raspberry, or for Cpython, or others but not for ESP. In short, I can't do the conversion with the hardware commands.
Do you know where can I find one?
Thank you in advance

[eydam-prototyping]

There is a DHT driver already integrated in MicroPython:
http://docs.micropython.org/en/latest/e ... dht-driver

Does this answer your question?

[PLUB]

Hi,
Yes I know there is a pilot but what I want is to understand how it works. Is there a DHT.py for ESP?
Regards.

[rpr]

Do you want to use dht on an esp32?

http://docs.micropython.org/en/latest/e ... dht-driver

OR

Do you want to know the details of how the dht module works?

https://github.com/micropython/micropyt ... rivers/dht

[PLUB]

Hi,
thank you for your reply. The Github https://github.com/micropython/micropyt ... rivers/dht matches my request but it's written in C. I'm looking for something like https://github.com/garyns/ pigpio-dht / blob / master / lib / pigpio_dht / dhtxx.py, adapted to ESP32.
Regards

[rpr]

Now I understand. It looks like you want a driver written completely in micropython. Maybe one exists but I'm guessing that probably not since the C built in driver already exists. Maybe someone has written it in python as purely an academic exercise.

[Roberthh]

@PLUB: Most likely it won't work. Reading the DHT11/22, the code must tell the difference between pulses ~20 and ~60 µs long. The micropython code uses C for that purpose. A similar code for pycom devices uses the RMT unit for that task. The esp32 is notorious for skews in timing. So using a python code with polling will not result in the required resolution due to interrupts that may happen. Using interrupts like in the code you cited will also not work, due to the very long and varying interrupt latency of the micropython + esp32 combo.

If you want to try anyhow. I have encloses the code used for Pycom devices once. The critical call is pycom.pulses_get(), which returns a list of (level, duration) pairs, with duration give in µs. That would have to be re-modeled by a call, polling the pin and collecting the times.

Code: Select all

import time
import pycom
from machine import Pin

class DHTResult:
    'DHT sensor result returned by DHT.read() method'

    ERR_NO_ERROR = 0
    ERR_MISSING_DATA = 1
    ERR_CRC = 2

    error_code = ERR_NO_ERROR
    temperature = -1
    humidity = -1

    def __init__(self, error_code, temperature, humidity):
        self.error_code = error_code
        self.temperature = temperature
        self.humidity = humidity

    def is_valid(self):
        return self.error_code == DHTResult.ERR_NO_ERROR


class DHT:
    'DHT sensor (dht11, dht21,dht22) reader class for Pycom'

    __dhttype = 0

    def __init__(self, pin, sensor=0):
        self.__pin = pin
        self.__dhttype = sensor
        self.__pin(1)
        time.sleep(1.0)

    def read(self):
        # pull down to low
        self.__send_and_sleep(0, 0.019)
        data = pycom.pulses_get(self.__pin, 100)
        self.__pin.init(Pin.OPEN_DRAIN)
        self.__pin(1)
        bits = []
        for a,b in data:
                if a ==1 and 18 <= b <= 28:
                        bits.append(0)
                if a ==1 and 65 <= b <= 75:
                        bits.append(1)
        if len(bits) != 40:
            return DTHResult(DTHResult.ERR_MISSING_DATA, 0, 0)
        # we have the bits, calculate bytes
        the_bytes = self.__bits_to_bytes(bits)
        # calculate checksum and check
        checksum = self.__calculate_checksum(the_bytes)
        if the_bytes[4] != checksum:
            return DTHResult(DTHResult.ERR_CRC, 0, 0)
        # ok, we have valid data, return it
        [int_rh, dec_rh, int_t, dec_t, csum] = the_bytes
        if self.__dhttype==0:           #dht11
            rh = int_rh                 #dht11 20% ~ 90%
            t = int_t   #dht11 0..50°C
        else:                   #dht21,dht22
            rh = ((int_rh * 256) + dec_rh)/10
            t = (((int_t & 0x7F) * 256) + dec_t)/10
            if (int_t & 0x80) != 0:
                t = -t
        return DTHResult(DTHResult.ERR_NO_ERROR, t, rh)


    def __send_and_sleep(self, output, mysleep):
        self.__pin(output)
        time.sleep(mysleep)

    def __bits_to_bytes(self, bits):
        the_bytes = []
        byte = 0

        for i in range(0, len(bits)):
            byte <<= 1
            if (bits[i]):
                byte |= 1
            if ((i % 8) == 7):
                the_bytes.append(byte)
                byte = 0
        return the_bytes

    def __calculate_checksum(self, the_bytes):
        return the_bytes[0] + the_bytes[1] + the_bytes[2] + the_bytes[3] & 255

[PLUB]

Hi,
Thank you for your answers.
I'll see if the RMT module can help me move forward.
On the other hand if someone has an idea to build a routine equivalent to pycom.pulses_get (), I am interested.
Regards

[Roberthh]

Just to prove that I'm wrong, here is the variant of my first code using polling. Seems to work fine. It assumes a clock frequency of 160 MHz. At other frequencies the duration of 0 and 1 bits have to be aligned.

Code: Select all

import time
from machine import Pin, disable_irq, enable_irq

class DTHResult:
    'DHT sensor result returned by DHT.read() method'

    ERR_NO_ERROR = 0
    ERR_MISSING_DATA = 1
    ERR_CRC = 2

    error_code = ERR_NO_ERROR
    temperature = -1
    humidity = -1

    def __init__(self, error_code, temperature, humidity, dew_point):
        self.error_code = error_code
        self.temperature = temperature
        self.humidity = humidity
        self.dew_point = dew_point

    def is_valid(self):
        return self.error_code == DTHResult.ERR_NO_ERROR


class DTH:
    'DHT sensor (dht11, dht21,dht22) reader class for Pycom'

    __dhttype = 0

    def __init__(self, pin, sensor=1):
        self.__pin_obj = Pin(pin, mode=Pin.OPEN_DRAIN)
        self.__pin = pin
        self.__dhttype = sensor
        self.__pin_obj(1)
        time.sleep(1.0)
        self.buffer = bytearray(1024)

    def dew_point(self, temp, humid):
        import math
        """
        Compute the dew point temperature for the current Temperature
        and Humidity measured pair
        """
        h = ((math.log(humid, 10) - 2) / 0.4343 +
                (17.62 * temp) / (243.12 + temp))
        dew_p = 243.12 * h / (17.62 - h)
        return dew_p

    @micropython.native
    def pulses_get(self, pin):
        # start the reading
        pin(0)
        time.sleep_ms(19)

        state = disable_irq()
        pin(1)
        for _ in range(1024):
            self.buffer[_] = pin()
        enable_irq(state)

    def read(self):
        self.pulses_get(self.__pin_obj)
        state = self.buffer[0]
        pos = 0
        bits = []
        for _ in range(1024):
            if state != self.buffer[_]:
                bl = _ - pos
                if state == 1 and 2 <= bl < 7:
                        bits.append(0)
                if state == 1 and 8 <= bl <= 15:
                        bits.append(1)
                pos = _
                state = self.buffer[_]

        if len(bits) != 40:
            return DTHResult(DTHResult.ERR_MISSING_DATA, 0, 0)
        # we have the bits, calculate bytes
        the_bytes = self.__bits_to_bytes(bits)
        # calculate checksum and check
        checksum = self.__calculate_checksum(the_bytes)
        if the_bytes[4] != checksum:
            return DTHResult(DTHResult.ERR_CRC, 0, 0)
        # ok, we have valid data, return it
        [int_rh, dec_rh, int_t, dec_t, csum] = the_bytes
        if self.__dhttype == 0:         # dht11
            rh = int_rh                 # dht11 20% ~ 90%
            t = int_t   # dht11 0..50°C
        else:           # dht21,dht22
            rh = ((int_rh * 256) + dec_rh)/10
            t = (((int_t & 0x7F) * 256) + dec_t)/10
            if (int_t & 0x80) != 0:
                t = -t
        dew_point = self.dew_point(t, rh)
        return DTHResult(DTHResult.ERR_NO_ERROR, t, rh, dew_point)

    def __bits_to_bytes(self, bits):
        the_bytes = []
        byte = 0

        for i in range(0, len(bits)):
            byte <<= 1
            if (bits[i]):
                byte |= 1
            if ((i % 8) == 7):
                the_bytes.append(byte)
                byte = 0
        return the_bytes

    def __calculate_checksum(self, the_bytes):
        return (the_bytes[0] + the_bytes[1] + the_bytes[2] + the_bytes[3]) & 255

th = DTH(4, 1)

def run():
    time.sleep(0.5)
    result = th.read()
    if result.is_valid():
        print("Humidity: %.1f %%" % result.humidity)
        print("Temperature: %.1f C" % result.temperature)
        print("Dew Point: %.1f C" % result.dew_point)

run()