Skip to content

Instantly share code, notes, and snippets.

@ekerstein
Created August 25, 2019 18:17
Show Gist options
  • Star 2 You must be signed in to star a gist
  • Fork 0 You must be signed in to fork a gist
  • Save ekerstein/56757f993a7a8d17eabbca5a8562f5cc to your computer and use it in GitHub Desktop.
Save ekerstein/56757f993a7a8d17eabbca5a8562f5cc to your computer and use it in GitHub Desktop.
Python script to pull PurpleAir data for multiple sensors from the Thingspeak API, combine them, and export a CSV file. To run this script you'll need to adjust the "user config" section, and you may also want to adjust the Thingspeak query parameters starting on row 49.
import requests # for url requests
from statistics import median # for median of array
import json # for json reading/writing
import time # for epoch timestamp
import csv # for writing csv files
#################################################
# USER CONFIG
data_dir = '/code/data/' # keep trailing slash
# Thingspeak sensor list (two feeds per sensor)
sensors = [
{
'id': '813992',
'key': '0F7FRDWOSFGWETPH'
},
{
'id': '813994',
'key': '9VPLFBBQDCOSVJWK'
},
{
'id': '813997',
'key': '4406OWFMIRW4575W'
},
{
'id': '813998',
'key': 'Z4C30UYGXVX3QY5K'
}
]
#################################################
# GET DATA
# Make master data objects
data = {}
fields = ['created_at'] # manually add created_at field for csv header mapping
# Iterate through sensor list
for sensor in sensors:
# Set ID/key
sensor_id = sensor['id']
sensor_key = sensor['key']
url = 'https://api.thingspeak.com/channels/{}/feeds.json'.format(sensor_id)
# Thingspeak API parameters: https://www.mathworks.com/help/thingspeak/readdata.html
params = {
'api_key': sensor_key, # sensor API key
'days': '7', # days of data to get
'round': '1', # round decimal places
'average': '10' # 10 minute averages
}
r = requests.get(url, params=params)
#print(r.url) # print out full URL request with params
data_raw = r.json()
############################
# Parse sensor data
fields_sensor = {}
data_sensor = []
# Make sensor field list
for d in data_raw['channel']:
if d.startswith('field'):
field_name = data_raw['channel'][d]
# Add to field sensor mapping
fields_sensor[d] = field_name
# Add to master field list
if field_name not in fields:
fields.append(field_name)
# Parse sensor data feeds
for i, row in enumerate(data_raw['feeds']):
# Add object to data
data_sensor.append({})
for key in row:
# Add and rename
if key in fields_sensor:
new_key = fields_sensor[key]
data_sensor[i][new_key] = row[key]
# Or just add
else:
data_sensor[i][key] = row[key]
'''
# Check field mapping
print(data_raw['feeds'][0])
print(data_sensor[0])
'''
############################
# Add sensor data to master data dictionary
for row in data_sensor:
created_at = row['created_at']
# Make new timestamp object if needed
if created_at not in data:
data[created_at] = {}
# Add to timestamp object
for key in row:
# If key exists, append to value list
if key in data[created_at]:
data[created_at][key].append(row[key])
# Else make new value list
else:
data[created_at][key] = [row[key]]
#################################################
# Make CSV file
# Make csv data array with median values
data_csv = []
for timestamp in data:
row = data[timestamp].copy()
# For each key
for key in row:
# Collapse value list to median, or take first value (for date strings or None)
try:
row[key] = median(map(float, row[key]))
except:
row[key] = row[key][0]
# Append to array
data_csv.append(row)
# Write to CSV file
with open(data_dir + 'purpleair.csv', 'w', newline='') as f:
writer = csv.DictWriter(f, fieldnames=fields)
# Write header
writer.writeheader()
# Write rows
for row in data_csv:
empty = True
# Check if empty row
for key in row:
if key != 'created_at' and row[key]:
empty = False
# If not empty row
if empty == False:
writer.writerow(row)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment