examples/luftdaten.py

   1 #!/usr/bin/env python
   2
   3 import requests
   4 import ST7735
   5 import time
   6 from bme280 import BME280
   7 from pms5003 import PMS5003, ReadTimeoutError
   8 from subprocess import PIPE, Popen, check_output
   9 from PIL import Image, ImageDraw, ImageFont
  10
  11 try:
  12     from smbus2 import SMBus
  13 except ImportError:
  14     from smbus import SMBus
  15
  16 print("""luftdaten.py - Reads temperature, pressure, humidity,
  17 PM2.5, and PM10 from Enviro plus and sends data to Luftdaten,
  18 the citizen science air quality project.
  19
  20 Note: you'll need to register with Luftdaten at:
  21 https://meine.luftdaten.info/ and enter your Raspberry Pi
  22 serial number that's displayed on the Enviro plus LCD along
  23 with the other details before the data appears on the
  24 Luftdaten map.
  25
  26 Press Ctrl+C to exit!
  27
  28 """)
  29
  30 bus = SMBus(1)
  31
  32 # Create BME280 instance
  33 bme280 = BME280(i2c_dev=bus)
  34
  35 # Create LCD instance
  36 disp = ST7735.ST7735(
  37     port=0,
  38     cs=1,
  39     dc=9,
  40     backlight=12,
  41     rotation=270,
  42     spi_speed_hz=10000000
  43 )
  44
  45 # Initialize display
  46 disp.begin()
  47
  48 # Create PMS5003 instance
  49 pms5003 = PMS5003()
  50
  51
  52 # Read values from BME280 and PMS5003 and return as dict
  53 def read_values():
  54     values = {}
  55     cpu_temp = get_cpu_temperature()
  56     raw_temp = bme280.get_temperature()
  57     comp_temp = raw_temp - ((cpu_temp - raw_temp) / comp_factor)
  58     values["temperature"] = "{:.2f}".format(comp_temp)
  59     values["pressure"] = "{:.2f}".format(bme280.get_pressure() * 100)
  60     values["humidity"] = "{:.2f}".format(bme280.get_humidity())
  61     try:
  62         pm_values = pms5003.read()
  63         values["P2"] = str(pm_values.pm_ug_per_m3(2.5))
  64         values["P1"] = str(pm_values.pm_ug_per_m3(10))
  65     except ReadTimeoutError:
  66         pms5003.reset()
  67         pm_values = pms5003.read()
  68         values["P2"] = str(pm_values.pm_ug_per_m3(2.5))
  69         values["P1"] = str(pm_values.pm_ug_per_m3(10))
  70     return values
  71
  72
  73 # Get CPU temperature to use for compensation
  74 def get_cpu_temperature():
  75     process = Popen(['vcgencmd', 'measure_temp'], stdout=PIPE)
  76     output, _error = process.communicate()
  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 time_since_update = 0
 170 update_time = time.time()
 171
 172 # Main loop to read data, display, and send to Luftdaten
 173 while True:
 174     try:
 175         time_since_update = time.time() - update_time
 176         values = read_values()
 177         print(values)
 178         if time_since_update > 145:
 179             resp = send_to_luftdaten(values, id)
 180             update_time = time.time()
 181             print("Response: {}\n".format("ok" if resp else "failed"))
 182         display_status()
 183     except Exception as e:
 184         print(e)