landsat_area

landsat_area

def clearest_look(allscenes, lat, lon, i, j):
   name = 'pt_{}x{}'.format(i,j)
   return (allscenes
     | 'covers_{}'.format(name) >> beam.FlatMap(lambda scene: filterByLocation(scene, lat+0.25*i, lon-0.4*j))
     | 'bymonth_{}'.format(name) >> beam.FlatMap(lambda scene: sceneByMonth(scene) )
     | 'clearest_{}'.format(name) >> beam.CombinePerKey(clearest)
   )

def run():
   # etc.
   
   lat =-21.1; lon = 55.50     # center of Reunion Island

   # Read the index file and find all scenes
   allscenes = (p
      | 'read_index' >> beam.Read(beam.io.TextFileSource(index_file))
      | 'to_scene' >> beam.Map(lambda line:  SceneInfo(line))
   )

   # find all the Landsat scenes that cover a set of points i,j in Reunion
   all_looks = [clearest_look(allscenes, lat, lon, i, j) for i in xrange(-1,2) for j in xrange(-1,2)]

   # remove duplicate scenes from the PCollections
   scenes = (all_looks
      | 'all_looks' >> beam.Flatten()
      | 'all_looks_by_month' >> beam.CombinePerKey(lambda x:x)
      | 'set_looks_by_month' >> beam.Map(lambda (a,x) : (a, set([item for sublist in x for item in sublist])))
      | 'looks_by_month' >> beam.FlatMap(lambda (k,vset): flatten_kvset(k,vset))
   )
   
   # etc.