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