Emilio Mayorga, University of Washington, Seattle. 2018-5-17 (minor updates to documentation on 2018-8-19). Demo put together using as a starting point instructions from Azavea from October 2017. See also the related, previous notebook, https://github.com/WikiWatershed/model-my-watershed/blob/develop/doc/MMW_API_watershed_demo.ipynb
The Model My Watershed API allows you to delineate watersheds and analyze geo-data for watersheds and arbitrary areas. You can read more about the work at WikiWatershed or use the web app.
MMW users can discover their API keys through the user interface, and test the MMW geoprocessing API on either the live or staging apps. An Account page with the API key is available from either app (live or staging). To see it, go to the app, log in, and click on "Account" in the dropdown that appears when you click on your username in the top right. Your key is different between staging and production. For testing with the live (production) API and key, go to https://modelmywatershed.org/api/docs/
The API can be tested from the command line using curl. This example uses the production API to test the watershed endpoint:
curl -H "Content-Type: application/json" -H "Authorization: Token YOUR_API_KEY" -X POST
-d '{ "location": [39.67185,-75.76743] }' https://modelmywatershed.org/api/watershed/
In [1]:
import json
import requests
from requests.adapters import HTTPAdapter
from requests.packages.urllib3.util.retry import Retry
In [2]:
def requests_retry_session(
retries=3,
backoff_factor=0.3,
status_forcelist=(500, 502, 504),
session=None,
):
session = session or requests.Session()
retry = Retry(
total=retries,
read=retries,
connect=retries,
backoff_factor=backoff_factor,
status_forcelist=status_forcelist,
)
adapter = HTTPAdapter(max_retries=retry)
session.mount('http://', adapter)
session.mount('https://', adapter)
return session
MMW production API endpoint base url.
In [3]:
api_url = "https://modelmywatershed.org/api/"
The job is not completed instantly and the results are not returned directly by the API request that initiated the job. The user must first issue an API request to confirm that the job is complete, then fetch the results. The demo presented here performs automated retries (checks) until the server confirms the job is completed, then requests the JSON results and converts (deserializes) them into a Python dictionary.
In [4]:
def get_job_result(api_url, s, jobrequest):
url_tmplt = api_url + "jobs/{job}/"
get_url = url_tmplt.format
result = ''
while not result:
get_req = requests_retry_session(session=s).get(get_url(job=jobrequest['job']))
result = json.loads(get_req.content)['result']
return result
In [5]:
s = requests.Session()
In [6]:
APIToken = '<YOUR API TOKEN STRING>' # ENTER YOUR OWN API TOKEN
s.headers.update({
'Authorization': APIToken,
'Content-Type': 'application/json'
})
Parameters passed to the "analyze" API requests.
In [7]:
from shapely.geometry import box, MultiPolygon
In [8]:
width = 0.0004 # Looks like using a width smaller than 0.0002 causes a problem with the API?
In [9]:
# GOOS: (-88.5552, 40.4374) elev 240.93. Agriculture Site—Goose Creek (Corn field) Site (GOOS) at IML CZO
# SJER: (-119.7314, 37.1088) elev 403.86. San Joaquin Experimental Reserve Site (SJER) at South Sierra CZO
lon, lat = -119.7314, 37.1088
In [10]:
bbox = box(lon-0.5*width, lat-0.5*width, lon+0.5*width, lat+0.5*width)
In [11]:
payload = MultiPolygon([bbox]).__geo_interface__
json_payload = json.dumps(payload)
In [12]:
payload
Out[12]:
In [13]:
# convenience function, to simplify the request calls, below
def analyze_api_request(api_name, s, api_url, json_payload):
post_url = "{}analyze/{}/".format(api_url, api_name)
post_req = requests_retry_session(session=s).post(post_url, data=json_payload)
jobrequest_json = json.loads(post_req.content)
# Fetch and examine job result
result = get_job_result(api_url, s, jobrequest_json)
return result
In [14]:
result = analyze_api_request('land', s, api_url, json_payload)
Everything below is just exploration of the results. Examine the content of the results (as JSON, and Python dictionaries)
In [15]:
type(result), result.keys()
Out[15]:
result is a dictionary with one item, survey. This item in turn is a dictionary with 3 items: displayName, name, categories. The first two are just labels. The data are in the categories item.
In [16]:
result['survey'].keys()
Out[16]:
In [ ]:
categories = result['survey']['categories']
In [17]:
len(categories), categories[1]
Out[17]:
In [18]:
land_categories_nonzero = [d for d in categories if d['coverage'] > 0]
In [19]:
land_categories_nonzero
Out[19]:
In [20]:
result = analyze_api_request('terrain', s, api_url, json_payload)
result is a dictionary with one item, survey. This item in turn is a dictionary with 3 items: displayName, name, categories. The first two are just labels. The data are in the categories item.
In [22]:
categories = result['survey']['categories']
In [23]:
len(categories), categories
Out[23]:
In [24]:
[d for d in categories if d['type'] == 'average']
Out[24]:
In [25]:
result = analyze_api_request('climate', s, api_url, json_payload)
result is a dictionary with one item, survey. This item in turn is a dictionary with 3 items: displayName, name, categories. The first two are just labels. The data are in the categories item.
In [26]:
categories = result['survey']['categories']
In [27]:
len(categories), categories[:2]
Out[27]:
In [28]:
ppt = [d['ppt'] for d in categories]
tmean = [d['tmean'] for d in categories]
In [43]:
# ppt is in cm, right?
sum(ppt)
Out[43]:
In [32]:
import calendar
import numpy as np
In [31]:
calendar.mdays
Out[31]:
In [44]:
# Annual tmean needs to be weighted by the number of days per month
sum(np.asarray(tmean) * np.asarray(calendar.mdays[1:]))/365
Out[44]:
In [ ]: