Example
Appendix: Getting Results in Python
The following Python 3 program demonstrates how synchronously retrieve results from a project.
Each resource type is implemented as a simple class, and in main
is the code needed to navigate
and iterate through the results system.
The program requires that you have the environment variables $GO_API_URL
and $GO_API_TOKEN
set. To run the program, just supply the project and version ids as command line arguments:
$ exapmple.py project_id version_id
If you have further questions about accessing results, please contact the Orbital Insight Client
Success Team ([email protected]).
#!/usr/bin/env python3
# Copyright 2019 Orbital Insight Inc., all rights reserved.
# Contains confidential and trade secret information.
# Government Users: Commercial Computer Software - Use governed by
# terms of Orbital Insight commercial license agreement.
import os
import sys
import requests
from typing import Dict, List
token = os.getenv('GO_API_TOKEN')
base_url = 'https://go-services.orbitalinsight.com/api/v2/go'
headers = {"Authorization": "Bearer {0}".format(token)}
def _get_data(path: str) -> dict:
""" Given an endpoint, do a GET and return the result as a dict
"""
url = '{0}{1}?limit=-1'.format(base_url, path)
response = requests.get(url, headers=headers)
if response.status_code != 200:
raise RuntimeError(base_url + path + '\n' + 'HTTP GET failed: {0} {1}'.format(
response.status_code, response.reason))
return response.json()['data']
class GeoInfo(object):
def __init__(self, data: dict):
self._data = data
def dump(self, depth: int):
print('{0}GEOINFO: {1}'.format(' '*depth, self))
class TimeSeries(object):
def __init__(self, data: dict):
self._data = data
def dump(self, timeseries_type: str, depth: int):
if timeseries_type == 'raw.count':
s = '{0} points'.format(len(self._data))
elif timeseries_type == 'scene.subaoi.fillforward.normalized.count':
s = '{0} points'.format(len(self._data))
elif timeseries_type == 'change_detection.valid_geoms':
s = '{0}/{1}'.format(self._data['y_value']['change_class'],
self._data['y_value']['change_type'])
else:
raise RuntimeError('timeseries type not supported: ' + timeseries_type)
print('{0}TIMESERIES {1}: {2}'.format(' '*depth, timeseries_type, s))
class Metric(object):
def __init__(self, data: dict):
self._data = data
def dump(self, depth: int):
import pdb ; pdb.set_trace()
print('{0}METRIC: {1}'.format(' '*depth, self))
class AOI(object):
def __init__(self, data: dict):
self._data = data
self.id = self._data['id'] # type: str
class Result(object):
def __init__(self, project: "Project", data: dict):
self._project = project
self._data = data
self.division_id = self._data['division_id'] # type: str
self.id = self._data['result_id'] # type: str
self.timeseries_types = self._data['timeseries'] # type: List[str]
if not self.timeseries_types:
self.timeseries_types = list()
def get_timeseries(self, timeseries_type: str) -> List[TimeSeries]:
""" Return the Timeseries objects pointed to by this Result
"""
p = "/projects/{0}/versions/{1}/results/{2}/timeseries/{3}".format(
self._project.project_id, self._project.version_id, self.id, timeseries_type)
data = _get_data(p)
timeseries = [TimeSeries(item) for item in data]
return timeseries
def get_geoinfo(self) -> List[GeoInfo]:
""" Return the GeoInfo objects pointed to by this Result
"""
data = _get_data("/projects/{0}/versions/{1}/results/{2}/geoinfo".format(
self._project.project_id, self._project.version_id, self.id))
geoinfo = [GeoInfo(item) for item in data]
return geoinfo
def get_metrics(self) -> List[Metric]:
""" Return the Metric objects pointed to by this Result
"""
p = "/projects/{0}/versions/{1}/results/{2}/metrics".format(
self._project.project_id, self._project.version_id, self.id)
data = _get_data(p)
metrics = [Metric(item) for item in data]
return metrics
def dump(self, depth):
""" For this Result, dump whatever types of data it has, e.g. GeoInfo or Timeseries
information
"""
print('{0}Result: {1}'.format(' '*depth, self.id))
geoinfo_list = self.get_geoinfo()
for geoinfo in geoinfo_list:
geoinfo.dump(depth+1)
metric_list = self.get_metrics()
for metric in metric_list:
metric.dump(depth+1)
for timeseries_type in self.timeseries_types:
timeseries_list = self.get_timeseries(timeseries_type)
for timeseries in timeseries_list:
timeseries.dump(timeseries_type, depth+1)
class Division(object):
def __init__(self, data: dict):
self._data = data
self.id = self._data['division_id'] # type: str
self.subdivision_ids = self._data['children'] # type: List[str]
class Project(object):
def __init__(self, project_id, version_id):
self.project_id = project_id # type: str
self.version_id = version_id # type: str
def get_results(self) -> List[Result]:
data = _get_data("/projects/{0}/versions/{1}/results".format(
self.project_id, self.version_id))
results = [Result(self, item) for item in data]
return results
def get_divisions(self) -> List[Division]:
data = _get_data("/projects/{0}/versions/{1}/divisions".format(
self.project_id, self.version_id))
divisions = [Division(item) for item in data]
return divisions
def get_aois(self) -> List[AOI]:
data = _get_data("/projects/{0}/versions/{1}/aois".format(
self.project_id, self.version_id))
aois = [AOI(item) for item in data]
return aois
def make_results_map(results: List[Result]) -> Dict[str, List[Result]]:
""" Each Result is associated with one or more division ids. Construct a dict which maps a
division id to all results associated with it.
"""
results_map = dict() # type: Dict[str, List[Result]]
for result in results:
key = result.division_id
if key not in results_map:
results_map[key] = list() # type: List[Result]
results_map[key].append(result)
return results_map
def visit_division(division: Division,
division_map: Dict[str, Division],
results_map: Dict[str, List[Result]],
depth: int) -> None:
""" Find and dump the Results associated with this Division, and then visit its children
(subdivisions)
"""
print('{0}Division: {1}'.format(' '*depth, division.id))
# dump the Results for this Division
for result in results_map[division.id]:
result.dump(depth)
# visit the child Divisions of this Division
for subdivision_id in division.subdivision_ids:
subdivision = division_map[subdivision_id]
visit_division(subdivision, division_map, results_map, depth+1)
def main():
project_id = sys.argv[1]
version_id = sys.argv[2]
project = Project(project_id, version_id)
# get Results data for all the project
results = project.get_results()
results_map = make_results_map(results)
# get all AOIs for the project
aois = project.get_aois()
# get all divisions for the project
divisions = project.get_divisions()
# Division.children stores ids but we need to get the actual divisions, so make a quick
# lookup table first
division_map = dict() # type: Dict[str, Division]
for division in divisions:
assert division.id not in division_map
division_map[division.id] = division
# iterate over all the AOIs
for aoi in aois:
print('AOI:', aoi.id)
# visit the Divisions associated with this AOI
for division in divisions:
if division.id == aoi.id:
visit_division(division, division_map, results_map, 1)
main()
Updated over 3 years ago