landsat

if __name__ == '__main__':
   p = beam.Pipeline(options=beam.utils.options.PipelineOptions())
   index_file = 'gs://gcp-public-data-landsat/index.csv.gz'
   output_file = 'output.txt'

   # center of Reunion Island
   lat =-21.1; lon = 55.50 

   # Read the index file and find the best look
   scenes = (p
      | 'read_index' >> beam.Read(beam.io.TextFileSource(index_file))
      | 'filter_scenes' >> beam.FlatMap(lambda line: filterScenes(line, lat, lon) )
      | 'least_cloudy' >> beam.CombinePerKey(clearest)
   )

   # write out info about scene
   scenes | beam.Map(lambda (yrmon, scene): scene.__dict__) | 'scene_info' >> beam.io.textio.WriteToText(output_file)

   # compute ndvi on scene
   scenes | 'compute_ndvi' >> beam.Map(lambda (yrmon, scene): ndvi.computeNdvi(scene.BASE_URL, 'gs://cloud-training-demos/landsat/'))

   p.run()