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()

Did this page help you?