<s> '''<STR_LIT>''' <EOL> import os <EOL> import sys <EOL> import logging <EOL> import re <EOL> import numpy <EOL> from datetime import datetime , time <EOL> from osgeo import gdal <EOL> from agdc . stacker import Stacker <EOL> from EOtools . utils import log_multiline <EOL> from EOtools . stats import temporal_stats <EOL> SCALE_FACTOR = <NUM_LIT> <EOL> console_handler = logging . StreamHandler ( sys . stdout ) <EOL> console_handler . setLevel ( logging . INFO ) <EOL> console_formatter = logging . Formatter ( '<STR_LIT>' ) <EOL> console_handler . setFormatter ( console_formatter ) <EOL> logger = logging . getLogger ( __name__ ) <EOL> if not logger . level : <EOL> logger . setLevel ( logging . DEBUG ) <EOL> logger . addHandler ( console_handler ) <EOL> class NDVIStacker ( Stacker ) : <EOL> """<STR_LIT>""" <EOL> def derive_datasets ( self , input_dataset_dict , stack_output_info , tile_type_info ) : <EOL> assert type ( input_dataset_dict ) == dict , '<STR_LIT>' <EOL> log_multiline ( logger . debug , input_dataset_dict , '<STR_LIT>' , '<STR_LIT:\t>' ) <EOL> output_dataset_dict = { } <EOL> nbar_dataset_info = input_dataset_dict [ '<STR_LIT>' ] <EOL> nbar_dataset_path = nbar_dataset_info [ '<STR_LIT>' ] <EOL> pqa_mask = self . get_pqa_mask ( input_dataset_dict [ '<STR_LIT>' ] [ '<STR_LIT>' ] ) <EOL> nbar_dataset = gdal . Open ( nbar_dataset_path ) <EOL> assert nbar_dataset , '<STR_LIT>' % nbar_dataset <EOL> logger . debug ( '<STR_LIT>' , nbar_dataset_path ) <EOL> no_data_value = - <NUM_LIT> <EOL> output_stack_path = os . path . join ( self . output_dir , '<STR_LIT>' ) <EOL> output_tile_path = os . path . join ( self . output_dir , re . sub ( '<STR_LIT>' , <EOL> '<STR_LIT>' % ( tile_type_info [ '<STR_LIT>' ] ) , <EOL> os . path . basename ( nbar_dataset_path ) <EOL> ) <EOL> ) <EOL> output_dataset_info = dict ( nbar_dataset_info ) <EOL> output_dataset_info [ '<STR_LIT>' ] = output_tile_path <EOL> output_dataset_info [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> output_dataset_info [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> output_dataset_info [ '<STR_LIT>' ] = <NUM_LIT:1> <EOL> near_ir_band_data = nbar_dataset . GetRasterBand ( <NUM_LIT:4> ) . ReadAsArray ( ) <EOL> visible_band_data = nbar_dataset . GetRasterBand ( <NUM_LIT:3> ) . ReadAsArray ( ) <EOL> logger . debug ( '<STR_LIT>' , near_ir_band_data ) <EOL> logger . debug ( '<STR_LIT>' , visible_band_data ) <EOL> logger . debug ( '<STR_LIT>' , SCALE_FACTOR ) <EOL> data_array = numpy . true_divide ( near_ir_band_data - visible_band_data , ( near_ir_band_data + visible_band_data ) ) * SCALE_FACTOR <EOL> self . apply_pqa_mask ( data_array , pqa_mask , no_data_value ) <EOL> gdal_driver = gdal . GetDriverByName ( tile_type_info [ '<STR_LIT>' ] ) <EOL> output_dataset = gdal_driver . Create ( output_tile_path , <EOL> nbar_dataset . RasterXSize , nbar_dataset . RasterYSize , <EOL> <NUM_LIT:1> , nbar_dataset . GetRasterBand ( <NUM_LIT:1> ) . DataType , <EOL> tile_type_info [ '<STR_LIT>' ] . split ( '<STR_LIT:U+002C>' ) ) <EOL> assert output_dataset , '<STR_LIT>' % output_dataset <EOL> output_dataset . SetGeoTransform ( nbar_dataset . GetGeoTransform ( ) ) <EOL> output_dataset . SetProjection ( nbar_dataset . GetProjection ( ) ) <EOL> output_band = output_dataset . GetRasterBand ( <NUM_LIT:1> ) <EOL> output_band . WriteArray ( data_array ) <EOL> output_band . SetNoDataValue ( no_data_value ) <EOL> output_band . FlushCache ( ) <EOL> output_dataset_metadata = nbar_dataset . GetMetadata ( ) <EOL> if output_dataset_metadata : <EOL> output_dataset . SetMetadata ( output_dataset_metadata ) <EOL> log_multiline ( logger . debug , output_dataset_metadata , '<STR_LIT>' , '<STR_LIT:\t>' ) <EOL> output_dataset . FlushCache ( ) <EOL> logger . info ( '<STR_LIT>' , output_tile_path ) <EOL> output_dataset_dict [ output_stack_path ] = output_dataset_info <EOL> return output_dataset_dict <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> def date2datetime ( input_date , time_offset = time . min ) : <EOL> if not input_date : <EOL> return None <EOL> return datetime . combine ( input_date , time_offset ) <EOL> ndvi_stacker = NDVIStacker ( ) <EOL> if ndvi_stacker . debug : <EOL> console_handler . setLevel ( logging . DEBUG ) <EOL> assert ( ndvi_stacker . x_index and ndvi_stacker . y_index ) , '<STR_LIT>' <EOL> assert ndvi_stacker . output_dir , '<STR_LIT>' <EOL> stack_info_dict = ndvi_stacker . stack_derived ( x_index = ndvi_stacker . x_index , <EOL> y_index = ndvi_stacker . y_index , <EOL> stack_output_dir = ndvi_stacker . output_dir , <EOL> start_datetime = date2datetime ( ndvi_stacker . start_date , time . min ) , <EOL> end_datetime = date2datetime ( ndvi_stacker . end_date , time . max ) , <EOL> satellite = ndvi_stacker . satellite , <EOL> sensor = ndvi_stacker . sensor ) <EOL> log_multiline ( logger . debug , stack_info_dict , '<STR_LIT>' , '<STR_LIT:\t>' ) <EOL> logger . info ( '<STR_LIT>' , len ( stack_info_dict ) , ndvi_stacker . output_dir ) <EOL> logger . info ( '<STR_LIT>' ) <EOL> for vrt_stack_path in stack_info_dict : <EOL> stats_dataset_path = vrt_stack_path . replace ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> temporal_stats_numexpr_module . main ( vrt_stack_path , stats_dataset_path , <EOL> noData = stack_info_dict [ vrt_stack_path ] [ <NUM_LIT:0> ] [ '<STR_LIT>' ] , <EOL> provenance = True ) <EOL> logger . info ( '<STR_LIT>' , stats_dataset_path ) </s>
