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