James Duffy

Personal AQI Service

// By •
featured-image.png

Every year the California fire season feels longer and more devastating, and this year, with everything else going on, it feels especially bad. So I finally took the time to build something to keep an eye on it: a script that ingests PurpleAir sensor data, graphs it in Grafana, and alerts me when the air near my apartment goes above 50 AQI.

The script pulls sensor data from PurpleAir’s API every five minutes or so and writes it to MySQL. On top of that I set up a Grafana dashboard with a map of the sensors and a few points of interest. Grafana also handles the alerting and sends me a push notification through Pushover.net when the AQI stays above 50 for 10 minutes.

MySQL and Grafana both run in containers on my Synology NAS, with their data on external volumes.

The Python script itself runs on my Intel NUC, kicked off by a cron job every five minutes. Each run it:

  1. Pulls the JSON data from PurpleAir
  2. Filters out any sensor that is:
    • Not located “outside”
    • Not reported in the last 5 minutes
    • Not within a specific longitude and latitude range
  3. Converts the raw PM 2.5 value to US AQI
  4. Inserts everything into the database

#MySQL Table

 1CREATE TABLE `sensor_data` (
 2  `id` int NOT NULL AUTO_INCREMENT,
 3  `purpleair_id` int DEFAULT '0',
 4  `created_at` datetime DEFAULT CURRENT_TIMESTAMP,
 5  `time` int DEFAULT NULL,
 6  `name` varchar(100) DEFAULT NULL,
 7  `latitude` decimal(10,8) DEFAULT NULL,
 8  `longitude` decimal(11,8) DEFAULT NULL,
 9  `geohash` varchar(12) DEFAULT NULL,
10  `pm2_5` decimal(8,2) DEFAULT NULL,
11  `aqi` int DEFAULT '0',
12  PRIMARY KEY (`id`),
13  KEY `purpleair_id` (`purpleair_id`),
14  KEY `time` (`time`),
15  KEY `idx_sensor_data_aqi` (`aqi`)
16) ENGINE=InnoDB DEFAULT CHARSET=utf8;

#Ingestion Script

 1import requests
 2import time
 3
 4import pymysql
 5import pymysql.cursors
 6
 7from json.decoder import JSONDecodeError
 8
 9from retrying import retry
10
11# Connect to the database
12connection = pymysql.connect(
13    host='',
14    user='',
15    password='',
16    db='',
17    charset='utf8')
18
19
20def aqi_from_pm(pm):
21    if pm > 350.5:
22        return calculate_aqi(pm, 500, 401, 500, 350.5)
23    elif pm > 250.5:
24        return calculate_aqi(pm, 400, 301, 350.4, 250.5)
25    elif pm > 150.5:
26        return calculate_aqi(pm, 300, 201, 250.4, 150.5)
27    elif pm > 55.5:
28        return calculate_aqi(pm, 200, 151, 150.4, 55.5)
29    elif pm > 35.5:
30        return calculate_aqi(pm, 150, 101, 55.4, 35.5)
31    elif pm > 12.1:
32        return calculate_aqi(pm, 100, 51, 35.4, 12.1)
33    elif pm >= 0:
34        return calculate_aqi(pm, 50, 0, 12, 0)
35    else:
36        return False
37
38
39def calculate_aqi(Cp, Ih, Il, BPh, BPl):
40    a = Ih - Il
41    b = BPh - BPl
42    c = Cp - BPl
43    return round(((a/b) * c + Il))
44
45
46@retry(wait_fixed=30000, stop_max_attempt_number=3)
47def fetch_results():
48    purpleair = requests.get("https://www.purpleair.com/json")
49    return purpleair.json()['results']
50
51purpleair_results = fetch_results()
52
53sql = "INSERT INTO `sensor_data` (`purpleair_id`, `time`, `name`, `latitude`, `longitude`, `pm2_5`, `aqi`) VALUES (%s, %s, %s, %s, %s, %s, %s)"
54
55data = []
56for measurement in purpleair_results:
57    is_outside = 'DEVICE_LOCATIONTYPE' in measurement and measurement['DEVICE_LOCATIONTYPE'] == 'outside'
58    is_recent = 'AGE' in measurement and measurement['AGE'] <= 300
59    is_valid = 'ID' in measurement and 'Label' in measurement and 'Lat' in measurement and 'Lon' in measurement and 'PM2_5Value' in measurement
60
61    if is_outside and is_recent and is_valid:
62        # Include California and some of the surrounding area
63        is_longitude =  -125.22216797 < measurement['Lon'] < -113.26904297
64        is_latitude = 32.17561248 < measurement['Lat'] < 42.16340342
65
66        if is_longitude and is_latitude:
67            raw_pm = measurement['PM2_5Value']
68            pm2_5 = float(raw_pm) if '.' in raw_pm else False
69
70            if pm2_5:
71                data.append([
72                    measurement['ID'],
73                    measurement['LastSeen'],
74                    measurement['Label'].strip(),
75                    measurement['Lat'],
76                    measurement['Lon'],
77                    pm2_5,
78                    aqi_from_pm(pm2_5)
79                ])
80
81cursor = connection.cursor()
82
83for sensor in data:
84    print(sensor)
85    cursor.execute(sql, sensor)
86
87connection.commit()
88
89connection.close()

Related Posts