First, get an access token and set it as the value of DEN_ACCESS_TOKEN in the environment before starting this IPython Notebook.
In [ ]:
import os
In [ ]:
access_token = os.environ["DEN_ACCESS_TOKEN"]
In [ ]:
API_PROTOCOL = "https"
API_LOCATION = "developer-api.nest.com"
In [ ]:
from urlparse import SplitResult, urlunsplit
from urllib import urlencode
In [ ]:
def get_api_url(path=""):
query = urlencode({"auth": access_token})
split = SplitResult(scheme=API_PROTOCOL, netloc=API_LOCATION, path=path, query=query, fragment="")
return urlunsplit(split)
In [ ]:
get_api_url()
In [ ]:
import requests
In [ ]:
r = requests.get(get_api_url())
print r.status_code
assert r.status_code == requests.codes.OK
In [ ]:
r.json()
In [ ]:
def get(path=""):
"""Get a dictionary of results for the given path"""
r = requests.get(get_api_url(path))
r.raise_for_status()
return r.json()
In [ ]:
get()
In [ ]:
get("structures")
In [ ]:
devices = get("devices")
thermostats = devices["thermostats"]
for t in thermostats:
print thermostats[t].keys()
In [ ]:
import json
In [ ]:
def get_stream(path=""):
"""Make a GET to the stream API and return the response object."""
r = requests.get(get_api_url(path), headers={"Accept": "text/event-stream"}, stream=True)
r.raise_for_status()
return r
In [ ]:
r = get_stream()
In [ ]:
r.status_code
In [ ]:
r.headers
In [ ]:
from time import sleep
In [ ]:
from contextlib import closing
In [ ]:
import logging
Need to reload the logging module because IPython Notebook has already loaded it.
In [ ]:
reload(logging)
In [ ]:
logging.basicConfig(filename="den.log", level=logging.DEBUG, format="%(asctime)s %(levelname)s %(message)s")
In [ ]:
logging.info("Cool!")
In [ ]:
def _is_event(line):
"""Is the given line an event line?"""
return line.startswith("event:")
In [ ]:
def _is_data(line):
"""Is the given line a data line?"""
return line.startswith("data:")
In [ ]:
def _process_event(line):
"""Process the given event line."""
_, event = line.split(":", 1)
event = event.strip()
return None if event == "keep-alive" else event
In [ ]:
def _process_data(line):
"""Process the given data line."""
_, data_str = line.split(":", 1)
return json.loads(data_str.strip())
In [ ]:
def _process(line):
"""Process the given line."""
value = None
if _is_event(line):
value = _process_event(line)
elif _is_data(line):
value = _process_data(line)
return value
In [ ]:
def stream():
"""Stream results from the API."""
with closing(get_stream()) as stream:
while True:
for l in stream.iter_lines():
if l:
value = _process(l)
if value:
logging.info(value)
sleep(5)
In [ ]:
from requests.exceptions import StreamConsumedError
In [ ]:
try:
stream()
except StreamConsumedError as e:
logging.critical("Stream finished: %s" % e)
except:
logging.critical("Unexpected error!")