examples/luftdaten.py

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