Created
August 25, 2019 18:17
-
-
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.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
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