Skip to content

Instantly share code, notes, and snippets.

Embed
What would you like to do?
Fetch Netatmo Weather Station measurements and store in InfluxDB
#!/usr/bin/env python
import os
import sys
import json
import time
import requests
# Get your client ID and secret by creating an App at https://dev.netatmo.com/
NETATMO_CLIENT_ID = ""
NETATMO_CLIENT_SECRET = ""
NETATMO_USERNAME = ""
NETATMO_PASSWORD = ""
# Ensure that the hostname and database name are correct for your InfluxDB
INFLUX_HOST = 'localhost'
INFLUX_PORT = '8086'
INFLUX_DATABASE = 'netatmo'
INFLUXDB_WRITE_URL = 'http://{}:{}/write?precision=s&db={}'.format(INFLUX_HOST, INFLUX_PORT, INFLUX_DATABASE)
print('Will write measurements to InfluxDB at endpoint {}'.format(INFLUXDB_WRITE_URL))
data = dict(grant_type='password', client_id=NETATMO_CLIENT_ID,
client_secret=NETATMO_CLIENT_SECRET, username=NETATMO_USERNAME,
password=NETATMO_PASSWORD, scope='read_station')
resp = requests.post('https://api.netatmo.com/oauth2/token', data=data)
if resp.status_code == 200:
token = resp.json()
token['expiry'] = int(time.time()) + token['expires_in']
while True:
# Check if token needs refresh
if token['expiry'] - int(time.time()) < 600:
data = dict(grant_type='refresh_token', refresh_token=token['refresh_token'], client_id=NETATMO_CLIENT_ID, client_secret=NETATMO_CLIENT_SECRET)
resp = requests.post('https://api.netatmo.com/oauth2/token', data=data)
if resp.status_code == 200:
token = resp.json()
token['expiry'] = int(time.time()) + token['expires_in']
# Fetch measurements
resp = requests.get('https://api.netatmo.com/api/getstationsdata?access_token=' + token['access_token'])
if resp.status_code == 200:
data = resp.json()
payload = ""
for device in data['body']['devices']:
timestamp = device['dashboard_data']['time_utc']
# Normalize station name and module names to ensure they become valid InfluxDB label values
stationname = device['station_name'].replace(' ', '_').encode('utf-8')
modulename = device['module_name'].replace(' ', '_').encode('utf-8')
for datatype in device['data_type']:
payload += '{0},station_name={1},module_name={2} value={3} {4}\n'.format(datatype, stationname, modulename, device['dashboard_data'][datatype], timestamp)
for module in device['modules']:
modulename = module['module_name'].replace(' ', '_').encode('utf-8')
for datatype in module['data_type']:
if datatype == 'Wind':
for subtype in ['WindStrength', 'WindAngle', 'GustStrength', 'GustAngle']:
payload += '{0},station_name={1},module_name={2} value={3} {4}\n'.format(subtype, stationname, modulename, module['dashboard_data'][subtype], timestamp)
else:
payload += '{0},station_name={1},module_name={2} value={3} {4}\n'.format(datatype, stationname, modulename, module['dashboard_data'][datatype], timestamp)
# Debug output to ensure payload contains data and has valid InfluxDB format
print('Writing the following data points to InfluxDB:')
print(payload)
# Write to InfluxDB
resp = requests.post(INFLUXDB_WRITE_URL, data=payload)
# New data arrives every 10 min so the sleep period is a bit shorter to ensure you get it all
time.sleep(480)
@sbrooke
Copy link

sbrooke commented Mar 15, 2020

I'm working on setting up a docker container to run this and your script has been invaluable. I'm also running into the same issue with offline modules. I'm trying to learn python, but pretty early in the process. Any suggestions for doing the error handling you mentioned above?

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment