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, ReadTimeoutError
   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     try:
  59         pm_values = pms5003.read()
  60         values["P2"] = str(pm_values.pm_ug_per_m3(2.5))
  61         values["P1"] = str(pm_values.pm_ug_per_m3(10))
  62     except ReadTimeoutError:
  63         pms5003 = PMS5003()
  64         pm_values = pms5003.read()
  65         values["P2"] = str(pm_values.pm_ug_per_m3(2.5))
  66         values["P1"] = str(pm_values.pm_ug_per_m3(10))
  67     return values
  68
  69 # Get CPU temperature to use for compensation
  70 def get_cpu_temperature():
  71     process = Popen(['vcgencmd', 'measure_temp'], stdout=PIPE)
  72     output, _error = process.communicate()
  73     return float(output[output.index('=') + 1:output.rindex("'")])
  74
  75 # Get Raspberry Pi serial number to use as ID
  76 def get_serial_number():
  77     with open('/proc/cpuinfo','r') as f:
  78         for line in f:
  79             if line[0:6]=='Serial':
  80                 return(line.split(":")[1].strip())
  81
  82 # Check for Wi-Fi connection
  83 def check_wifi():
  84     if check_output(['hostname', '-I']):
  85         return True
  86     else:
  87         return False
  88
  89 # Display Raspberry Pi serial and Wi-Fi status on LCD
  90 def display_status():
  91     wifi_status = "connected" if check_wifi() else "disconnected"
  92     text_colour = (255, 255, 255)
  93     back_colour = (0, 170, 170) if check_wifi() else (85, 15, 15)
  94     id = get_serial_number()
  95     message = "{}\nWi-Fi: {}".format(id, wifi_status)
  96     img = Image.new('RGB', (WIDTH, HEIGHT), color=(0, 0, 0))
  97     draw = ImageDraw.Draw(img)
  98     size_x, size_y = draw.textsize(message, font)
  99     x = (WIDTH - size_x) / 2
 100     y = (HEIGHT / 2) - (size_y / 2)
 101     draw.rectangle((0, 0, 160, 80), back_colour)
 102     draw.text((x, y), message, font=font, fill=text_colour)
 103     disp.display(img)
 104
 105 def send_to_luftdaten(values, id):
 106     pm_values = dict(i for i in values.items() if i[0].startswith("P"))
 107     temp_values = dict(i for i in values.items() if not i[0].startswith("P"))
 108
 109     resp_1 = requests.post("https://api.luftdaten.info/v1/push-sensor-data/",
 110         json={
 111             "software_version": "enviro-plus 0.0.1",
 112             "sensordatavalues": [{"value_type": key, "value": val} for key, val in pm_values.items()]
 113         },
 114         headers={
 115             "X-PIN":    "1",
 116             "X-Sensor": id,
 117             "Content-Type": "application/json",
 118             "cache-control": "no-cache"
 119         }
 120     )
 121
 122     resp_2 = requests.post("https://api.luftdaten.info/v1/push-sensor-data/",
 123         json={
 124             "software_version": "enviro-plus 0.0.1",
 125             "sensordatavalues": [{"value_type": key, "value": val} for key, val in temp_values.items()]
 126         },
 127         headers={
 128             "X-PIN":    "11",
 129             "X-Sensor": id,
 130             "Content-Type": "application/json",
 131             "cache-control": "no-cache"
 132         }
 133     )
 134
 135     if resp_1.ok and resp_2.ok:
 136         return True
 137     else:
 138         return False
 139
 140 # Compensation factor for temperature
 141 comp_factor = 1.2
 142
 143 # Raspberry Pi ID to send to Luftdaten
 144 id = "raspi-" + get_serial_number()
 145
 146 # Width and height to calculate text position
 147 WIDTH = disp.width
 148 HEIGHT = disp.height
 149
 150 # Text settings
 151 font_size = 16
 152 font = ImageFont.truetype("fonts/Asap/Asap-Bold.ttf", font_size)
 153
 154 # Display Raspberry Pi serial and Wi-Fi status
 155 print("Raspberry Pi serial: {}".format(get_serial_number()))
 156 print("Wi-Fi: {}\n".format("connected" if check_wifi() else "disconnected"))
 157
 158 # Main loop to read data, display, and send to Luftdaten
 159 while True:
 160     try:
 161         values = read_values()
 162         print(values)
 163         resp = send_to_luftdaten(values, id)
 164         print("Response: {}\n".format("ok" if resp else "failed"))
 165         display_status()
 166     except Exception as e:
 167         print(e)