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:

$ 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 = ''
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'],
            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._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._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,, 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,
        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,
        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
        print('{0}Result: {1}'.format('  '*depth,

        geoinfo_list = self.get_geoinfo()
        for geoinfo in geoinfo_list:

        metric_list = self.get_metrics()
        for metric in metric_list:

        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._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]

    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
    print('{0}Division: {1}'.format('  '*depth,

    # dump the Results for this Division
    for result in results_map[]:

    # 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 not in division_map
        division_map[] = division

    # iterate over all the AOIs
    for aoi in aois:

        # visit the Divisions associated with this AOI
        for division in divisions:
            if ==
                visit_division(division, division_map, results_map, 1)


Did this page help you?