Python Landsat AWS Elasticsearch Loader

landsat-aws-es-loader.py

# load https://aws.amazon.com/public-datasets/landsat/ scene_list into ES index

import csv
from datetime import datetime
import json

from elasticsearch import Elasticsearch

index_name = 'landsat-aws'
type_name = 'FeatureCollection'

es = Elasticsearch()

def gen_polygon(minx, miny, maxx, maxy):
    minx = float(minx)
    miny = float(miny)
    maxx = float(maxx)
    maxy = float(maxy)
    return [[
        [minx, miny],
        [maxx, miny],
        [maxx, maxy],
        [minx, maxy],
        [minx, miny]
    ]]

def gen_acqdate(value):
    try:
        t = datetime.strptime(value, '%Y-%m-%d %H:%M:%S.%f').isoformat()
    except ValueError:
        t = datetime.strptime(value, '%Y-%m-%d %H:%M:%S').isoformat()
    return t

# delete index if exists
if es.indices.exists(index_name):
    es.indices.delete(index_name)

# index settings
settings = {
    'mappings': {
        'FeatureCollection': {
            'properties': {
                'geometry': {
                    'type': 'geo_shape'
                }
            }
        }
    }
}

# create index
es.indices.create(index=index_name, body=settings)

with open('scene_list') as ff:
    c = csv.DictReader(ff)

    for row in c:
        geometry = gen_polygon(
            row['min_lon'],
            row['min_lat'],
            row['max_lon'],
            row['max_lat']
        )
        product = {
            'ID': row['productId'],
            'type': 'Feature',
            'geometry': {
                'type': 'Polygon',
                'coordinates': geometry
            },
            'properties': {
                'productId': row['productId'],
                'entityId': row['entityId'],
                'acquisitionDate': gen_acqdate(row['acquisitionDate']),
                'cloudCover': float(row['cloudCover']),
                'processingLevel': row['processingLevel'],
                'path': int(row['path']),
                'row': int(row['row']),
                'download_url': row['download_url']
            }
        }
 
        res = es.index(index=index_name, doc_type=type_name, id=row['productId'], body=product)