Raspberry Pi Pico – DHT11

Sample Code DHT11 – dht.py #

import utime
import rp2
from rp2 import PIO, asm_pio
from machine import Pin

   
@asm_pio(set_init=(PIO.OUT_HIGH),autopush=True, push_thresh=8)
def DHT11_PIO():
    # clock set at 500Khz  Cycle is 2us
    # drive output low for at least 20ms
    set(y,1)                    # 0
    pull()                      # 1
    mov(x,osr)                  # 2
    set(pindirs,1)              # 3 set pin to output
    set(pins,0)                 # 4 set pin low
    label ('waitx')
    jmp(x_dec,'waitx')          # 5 decrement x reg every 32 cycles
    set(pindirs,0)              # 6 set pin to input 
    # STATE A. Wait for high at least 80us. max should be  very short
    set(x,31)                   # 7 
    label('loopA')
    jmp(pin,'got_B')            # 8
    jmp(x_dec,'loopA')          # 9
    label('Error')
    in_(y,1)                    # 10
    jmp('Error')                # 11  Infinity loop error

    # STATE B. Get HIGH pulse. max should be 40us
    label('got_B')
    set(x,31)                   # 12
    label('loop_B')
    jmp(x_dec,'check_B')        # 13
    jmp('Error')                # 14
    label('check_B') 
    jmp(pin,'loop_B')           # 15
 
    # STATE C. Get LOW pulse. max should be 80us
    set(x,31)                   # 16
    label('loop_C')
    jmp(pin,'got_D')            # 17     
    jmp(x_dec,'loop_C')         # 18
    jmp('Error')                # 19
    
    # STATE D. Get HIGH pulse. max should be 80us
    label('got_D')
    set(x,31)                   # 20
    label('loop_D')
    jmp(x_dec,'check_D')        # 21
    jmp('Error')                # 22
    label('check_D')
    jmp(pin,'loop_D')           # 23
    
    # STATE E. Get Low pulse delay. should be around 50us
    set(x,31)                   # 24
    label('loop_E')
    jmp(pin,'got_F')            # 25
    jmp(x_dec,'loop_E')         # 26
    jmp('Error')                # 27
   
    # STATE F.
    # wait 40 us
    label('got_F')              
    nop() [20]                  # 28
    in_(pins,1)                 # 29
    # now wait for low pulse
    set(x,31)                   # 30
    jmp('loop_D')               # 31    



class DHT11:
    
    def __init__(self,dataPin, powerPin=None,dht11=False,smID=1):
        self.dataPin = dataPin
        self.powerPin = powerPin
        self.dht11 = dht11
        self.smID = smID
        self.dataPin.init(Pin.IN, Pin.PULL_UP)
        if self.powerPin is not None:
            self.powerPin.init(Pin.OUT)
            self.powerPin.value(0)
        self.sm= rp2.StateMachine(self.smID)
  
    def read_array(self):
        if self.powerPin is not None:
            self.powerPin.value(1)
        utime.sleep_ms(100)
        #start state machine
        self.sm.init(DHT11_PIO,freq=500000,
                     set_base=self.dataPin,
                     in_base=self.dataPin,
                     jmp_pin=self.dataPin)
        if self.dht11:
            self.sm.put(10000)
        else:
            self.sm.put(1000)
        self.sm.active(1)
        value = []
        for i in range(5):
            value.append(self.sm.get())
        self.sm.active(0)
        return value
 
    def read(self):
        value = self.read_array()
        sumV = 0
        for i in range(4):
            sumV += value[i]
        if (sumV & 0xff) == value[4]:
            if self.dht11:
                humidity=value[0] & 0x7f
                temperature=value[2] 
            else:                
                humidity=((value[0]<<8)  + value[1])/10.0
                temperature=(((value[2] &0x7f) << 8)  + value[3]) /10.0 
            if (value[2] & 0x80) == 0x80:
                temperature = -temperature            
            return temperature, humidity
        else:
            return None, None
        
if __name__ == "__main__":
    from machine import Pin
    from DHT11 import DHT11
    import utime
    dht_data = Pin(15,Pin.IN,Pin.PULL_UP)
    dht_sensor=DHT11(dht_data,Pin(14,Pin.OUT),dht11=False)
    while True:
        T,H = dht_sensor.read()
        if T is None:
            print(" sensor error")
        else:
            print("{:3.1f}'C  {:3.1f}%".format(T,H))
        #DHT22 not responsive if delay to short
        utime.sleep_ms(1000)

Sample Code main.py #

from machine import Pin
from time import sleep
from dht import  DHT11


pin = Pin(15,Pin.IN,Pin.PULL_UP)
dht11 = DHT11(pin,None,dht11=True)

while True:
    sleep(1)
    T,H = dht11.read()
    if T is None:
        print(" sensor error")
    else:
        print("Temperature :" + str(T) + "C   "+ "Humidity:"+ str(H) +"%")

Sample Code Oled – oled_main.py #

from machine import I2C,Pin, ADC
from time import sleep
from ssd1306 import SSD1306_I2C
from dht import  DHT11

i2c = I2C(0, sda = Pin(0), scl = Pin(1), freq = 400000)
oled = SSD1306_I2C(128, 64,i2c)


dht_data = Pin(15,Pin.IN,Pin.PULL_UP)
dht11 = DHT11(dht_data,None,dht11=True)

for x in range(65, 1, -1):
    oled.fill(0)
    oled.text("Welcome to",20,x)
    oled.text("voidlooprobotech", 0, (x+12))
    oled.show()
    sleep(0.02)
    
oled.text("DHT11 with",21,28)
oled.text("Raspberry Pi", 12,42)
oled.text("PICO board", 19,56)
oled.show()
for x in range(1, 5, 1):
    oled.invert(1)
    sleep(.5)
    oled.invert(0)
    sleep(.5)
sleep(0.5)
while True:

    T,H = dht11.read()
    if T is None:
        print(" sensor error")
    else:
        print("Temperature :" + str(T) + "C   "+ "Humidity:"+ str(H) +"%")
    oled.fill(0)
    oled.text("Welcome to",20,1)
    oled.text("voidlooprobotech", 0,12)
    oled.text("Temperature:" + str (T) + "C",0,28)
    oled.text("Humidity:" + str (H) + "%", 12,42)
    oled.text("Subscribe Please", 0,56)
    oled.show()
    sleep(1)

Sample Code SSD1306 Oled Driver – ssd1306.py #

# MicroPython SSD1306 OLED driver, I2C and SPI interfaces

from micropython import const
import framebuf


# register definitions
SET_CONTRAST = const(0x81)
SET_ENTIRE_ON = const(0xA4)
SET_NORM_INV = const(0xA6)
SET_DISP = const(0xAE)
SET_MEM_ADDR = const(0x20)
SET_COL_ADDR = const(0x21)
SET_PAGE_ADDR = const(0x22)
SET_DISP_START_LINE = const(0x40)
SET_SEG_REMAP = const(0xA0)
SET_MUX_RATIO = const(0xA8)
SET_COM_OUT_DIR = const(0xC0)
SET_DISP_OFFSET = const(0xD3)
SET_COM_PIN_CFG = const(0xDA)
SET_DISP_CLK_DIV = const(0xD5)
SET_PRECHARGE = const(0xD9)
SET_VCOM_DESEL = const(0xDB)
SET_CHARGE_PUMP = const(0x8D)

# Subclassing FrameBuffer provides support for graphics primitives
# http://docs.micropython.org/en/latest/pyboard/library/framebuf.html
class SSD1306(framebuf.FrameBuffer):
    def __init__(self, width, height, external_vcc):
        self.width = width
        self.height = height
        self.external_vcc = external_vcc
        self.pages = self.height // 8
        self.buffer = bytearray(self.pages * self.width)
        super().__init__(self.buffer, self.width, self.height, framebuf.MONO_VLSB)
        self.init_display()

    def init_display(self):
        for cmd in (
            SET_DISP | 0x00,  # off
            # address setting
            SET_MEM_ADDR,
            0x00,  # horizontal
            # resolution and layout
            SET_DISP_START_LINE | 0x00,
            SET_SEG_REMAP | 0x01,  # column addr 127 mapped to SEG0
            SET_MUX_RATIO,
            self.height - 1,
            SET_COM_OUT_DIR | 0x08,  # scan from COM[N] to COM0
            SET_DISP_OFFSET,
            0x00,
            SET_COM_PIN_CFG,
            0x02 if self.width > 2 * self.height else 0x12,
            # timing and driving scheme
            SET_DISP_CLK_DIV,
            0x80,
            SET_PRECHARGE,
            0x22 if self.external_vcc else 0xF1,
            SET_VCOM_DESEL,
            0x30,  # 0.83*Vcc
            # display
            SET_CONTRAST,
            0xFF,  # maximum
            SET_ENTIRE_ON,  # output follows RAM contents
            SET_NORM_INV,  # not inverted
            # charge pump
            SET_CHARGE_PUMP,
            0x10 if self.external_vcc else 0x14,
            SET_DISP | 0x01,
        ):  # on
            self.write_cmd(cmd)
        self.fill(0)
        self.show()

    def poweroff(self):
        self.write_cmd(SET_DISP | 0x00)

    def poweron(self):
        self.write_cmd(SET_DISP | 0x01)

    def contrast(self, contrast):
        self.write_cmd(SET_CONTRAST)
        self.write_cmd(contrast)

    def invert(self, invert):
        self.write_cmd(SET_NORM_INV | (invert & 1))

    def show(self):
        x0 = 0
        x1 = self.width - 1
        if self.width == 64:
            # displays with width of 64 pixels are shifted by 32
            x0 += 32
            x1 += 32
        self.write_cmd(SET_COL_ADDR)
        self.write_cmd(x0)
        self.write_cmd(x1)
        self.write_cmd(SET_PAGE_ADDR)
        self.write_cmd(0)
        self.write_cmd(self.pages - 1)
        self.write_data(self.buffer)


class SSD1306_I2C(SSD1306):
    def __init__(self, width, height, i2c, addr=0x3C, external_vcc=False):
        self.i2c = i2c
        self.addr = addr
        self.temp = bytearray(2)
        self.write_list = [b"\x40", None]  # Co=0, D/C#=1
        super().__init__(width, height, external_vcc)

    def write_cmd(self, cmd):
        self.temp[0] = 0x80  # Co=1, D/C#=0
        self.temp[1] = cmd
        self.i2c.writeto(self.addr, self.temp)

    def write_data(self, buf):
        self.write_list[1] = buf
        self.i2c.writevto(self.addr, self.write_list)


class SSD1306_SPI(SSD1306):
    def __init__(self, width, height, spi, dc, res, cs, external_vcc=False):
        self.rate = 10 * 1024 * 1024
        dc.init(dc.OUT, value=0)
        res.init(res.OUT, value=0)
        cs.init(cs.OUT, value=1)
        self.spi = spi
        self.dc = dc
        self.res = res
        self.cs = cs
        import time

        self.res(1)
        time.sleep_ms(1)
        self.res(0)
        time.sleep_ms(10)
        self.res(1)
        super().__init__(width, height, external_vcc)

    def write_cmd(self, cmd):
        self.spi.init(baudrate=self.rate, polarity=0, phase=0)
        self.cs(1)
        self.dc(0)
        self.cs(0)
        self.spi.write(bytearray([cmd]))
        self.cs(1)

    def write_data(self, buf):
        self.spi.init(baudrate=self.rate, polarity=0, phase=0)
        self.cs(1)
        self.dc(1)
        self.cs(0)
        self.spi.write(buf)
        self.cs(1)