examples/luftdaten.py

   1 #!/usr/bin/env python
   2
   3 import time
   4 import json
   5 import requests
   6 import ST7735
   7 from bme280 import BME280
   8 from pms5003 import PMS5003
   9 from subprocess import PIPE, Popen, check_output
  10 from PIL import Image, ImageDraw, ImageFont
  11
  12 try:
  13     from smbus2 import SMBus
  14 except ImportError:
  15     from smbus import SMBus
  16
  17 print("""luftdaten.py - Reads temperature, pressure, humidity, PM2.5, and PM10 from
  18 Enviro plus and sends data to Luftdaten, the citizen science air quality project.
  19
  20 Note: you'll need to register with Luftdaten at: https://meine.luftdaten.info/ and
  21 enter your Raspberry Pi serial number that's displayed on the Enviro plus LCD
  22 along with the other details before the data appears on the Luftdaten map.
  23
  24 Press Ctrl+C to exit!
  25
  26 """)
  27
  28 bus = SMBus(1)
  29
  30 # Create BME280 instance
  31 bme280 = BME280(i2c_dev=bus)
  32
  33 # Create PMS5003 instance
  34 pms5003 = PMS5003()
  35
  36 # Create LCD instance
  37 disp = ST7735.ST7735(
  38     port=0,
  39     cs=1,
  40     dc=9,
  41     backlight=12,
  42     rotation=270,
  43     spi_speed_hz=10000000
  44 )
  45
  46 # Initialize display
  47 disp.begin()
  48
  49 # Read values from BME280 and PMS5003 and return as dict
  50 def read_values():
  51     values = {}
  52     cpu_temp = get_cpu_temperature()
  53     raw_temp = bme280.get_temperature()
  54     comp_temp = raw_temp - ((cpu_temp - raw_temp) / comp_factor)
  55     values["temperature"] = "{:.2f}".format(comp_temp)
  56     values["pressure"] = "{:.2f}".format(bme280.get_pressure() * 100)
  57     values["humidity"] = "{:.2f}".format(bme280.get_humidity())
  58     pm_values = pms5003.read()
  59     values["P2"] = str(pm_values.pm_ug_per_m3(2.5))
  60     values["P1"] = str(pm_values.pm_ug_per_m3(10))
  61     return values
  62
  63 # Get CPU temperature to use for compensation
  64 def get_cpu_temperature():
  65     process = Popen(['vcgencmd', 'measure_temp'], stdout=PIPE)
  66     output, _error = process.communicate()
  67     return float(output[output.index('=') + 1:output.rindex("'")])
  68
  69 # Get Raspberry Pi serial number to use as ID
  70 def get_serial_number():
  71     with open('/proc/cpuinfo','r') as f:
  72         for line in f:
  73             if line[0:6]=='Serial':
  74                 return(line.split(":")[1].strip())
  75
  76 # Check for Wi-Fi connection
  77 def check_wifi():
  78     if check_output(['hostname', '-I']):
  79         return True
  80     else:
  81         return False
  82
  83 # Display Raspberry Pi serial and Wi-Fi status on LCD
  84 def display_status():
  85     wifi_status = "connected" if check_wifi() else "disconnected"
  86     text_colour = (255, 255, 255)
  87     back_colour = (0, 170, 170) if check_wifi() else (85, 15, 15)
  88     id = get_serial_number()
  89     message = "{}\nWi-Fi: {}".format(id, wifi_status)
  90     img = Image.new('RGB', (WIDTH, HEIGHT), color=(0, 0, 0))
  91     draw = ImageDraw.Draw(img)
  92     size_x, size_y = draw.textsize(message, font)
  93     x = (WIDTH - size_x) / 2
  94     y = (HEIGHT / 2) - (size_y / 2)
  95     draw.rectangle((0, 0, 160, 80), back_colour)
  96     draw.text((x, y), message, font=font, fill=text_colour)
  97     disp.display(img)
  98
  99 def send_to_luftdaten(values, id):
 100     pm_values = dict(i for i in values.items() if i[0].startswith("P"))
 101     temp_values = dict(i for i in values.items() if not i[0].startswith("P"))
 102
 103     resp_1 = requests.post("https://api.luftdaten.info/v1/push-sensor-data/",
 104         json={
 105             "software_version": "enviro-plus 0.0.1",
 106             "sensordatavalues": [{"value_type": key, "value": val} for key, val in pm_values.items()]
 107         },
 108         headers={
 109             "X-PIN":    "1",
 110             "X-Sensor": id,
 111             "Content-Type": "application/json",
 112             "cache-control": "no-cache"
 113         }
 114     )
 115
 116     resp_2 = requests.post("https://api.luftdaten.info/v1/push-sensor-data/",
 117         json={
 118             "software_version": "enviro-plus 0.0.1",
 119             "sensordatavalues": [{"value_type": key, "value": val} for key, val in temp_values.items()]
 120         },
 121         headers={
 122             "X-PIN":    "11",
 123             "X-Sensor": id,
 124             "Content-Type": "application/json",
 125             "cache-control": "no-cache"
 126         }
 127     )
 128
 129     if resp_1.ok and resp_2.ok:
 130         return True
 131     else:
 132         return False
 133
 134 # Compensation factor for temperature
 135 comp_factor = 1.2
 136
 137 # Raspberry Pi ID to send to Luftdaten
 138 id = "raspi-" + get_serial_number()
 139
 140 # Width and height to calculate text position
 141 WIDTH = disp.width
 142 HEIGHT = disp.height
 143
 144 # Text settings
 145 font_size = 16
 146 font = ImageFont.truetype("fonts/Asap/Asap-Bold.ttf", font_size)
 147
 148 # Display Raspberry Pi serial and Wi-Fi status
 149 print("Raspberry Pi serial: {}".format(get_serial_number()))
 150 print("Wi-Fi: {}\n".format("connected" if check_wifi() else "disconnected"))
 151
 152 # Main loop to read data, display, and send to Luftdaten
 153 while True:
 154     try:
 155         values = read_values()
 156         print(values)
 157         resp = send_to_luftdaten(values, id)
 158         print("Response: {}\n".format("ok" if resp else "failed"))
 159         display_status()
 160     except Exception as e:
 161         print(e)