Skip to content

Instantly share code, notes, and snippets.

@c4r-gists

c4r-gists/DHT22.py

Last active Feb 7, 2018
Embed
What would you like to do?
How to send temperature and humidity using DHT22(11) sensor
# -*- coding: utf-8 -*-
import sys
import time
import cloud4rpi
import rpi
import Adafruit_DHT
# Put your device token here. To get the token,
# sign up at https://cloud4rpi.io and create a device.
DEVICE_TOKEN = '__YOUR_DEVICE_TOKEN__'
# Constants
DHT_PIN = 4
DATA_SENDING_INTERVAL = 30 # secs
DIAG_SENDING_INTERVAL = 60 # secs
POLL_INTERVAL = 0.5 # secs
temp, hum = None, None
last_update = time.time() - 20
# Read from an sensor connected to GPIO
def update_data():
global last_update, hum, temp
if time.time() - last_update > 10:
hum, temp = Adafruit_DHT.read_retry(Adafruit_DHT.DHT22, DHT_PIN)
last_update = time.time()
def get_t():
update_data()
return round(temp, 2) if temp is not None else None
def get_h():
update_data()
return round(hum, 2) if hum is not None else None
def main():
# Put variable declarations here
variables = {
'DHT22 Temp': {
'type': 'numeric',
'bind': get_t
},
'DHT22 Humidity': {
'type': 'numeric',
'bind': get_h
},
'CPU Temp': {
'type': 'numeric',
'bind': rpi.cpu_temp
}
}
# Put system data declarations here
diagnostics = {
'CPU Temp': rpi.cpu_temp,
'IP Address': rpi.ip_address,
'Host Name': rpi.host_name,
'Operating System': rpi.os_name
}
device = cloud4rpi.connect(DEVICE_TOKEN)
device.declare(variables)
device.declare_diag(diagnostics)
device.publish_config()
# Adds a 1 second delay to ensure device variables are created
time.sleep(1)
try:
diag_timer = 0
data_timer = 0
while True:
if data_timer <= 0:
device.publish_data()
data_timer = DATA_SENDING_INTERVAL
if diag_timer <= 0:
device.publish_diag()
diag_timer = DIAG_SENDING_INTERVAL
diag_timer -= POLL_INTERVAL
data_timer -= POLL_INTERVAL
time.sleep(POLL_INTERVAL)
except KeyboardInterrupt:
cloud4rpi.log.info('Keyboard interrupt received. Stopping...')
except Exception as e:
error = cloud4rpi.get_error_message(e)
cloud4rpi.log.error("ERROR! %s %s", error, sys.exc_info()[0])
finally:
sys.exit(0)
if __name__ == '__main__':
main()
# Install python-dev tools
sudo apt-get update
sudo apt-get install build-essential python python-dev
# Download/clone the library and install it
git clone https://github.com/adafruit/Adafruit_Python_DHT
cd Adafruit_Python_DHT
sudo python setup.py install
# Install Cloud4RPi client library
sudo apt install python-pip -y
sudo pip install cloud4rpi
# -*- coding: utf-8 -*-
from os import uname
from socket import gethostname
import subprocess
import re
def parse_output(pattern, args):
try:
out_str = subprocess.check_output(args)
if isinstance(out_str, bytes):
out_str = out_str.decode()
except Exception:
out_str = ''
match = re.search(pattern, out_str)
return match.group(1) if match else None
def cpu_temp():
t_str = parse_output(r'temp=(\S*)\'C', ['vcgencmd', 'measure_temp'])
return float(t_str) if t_str else None
def ip_address():
return parse_output(r'(\S*)', ['hostname', '-I'])
def host_name():
return gethostname()
def os_name():
return " ".join(uname())
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment