Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def doit(self):
shape = (4000, 4000)
no_data_value = NDV
best_pixel_data = dict()
# TODO
if Satellite.LS8.value in self.satellites:
bands = Ls8Arg25Bands
else:
bands = Ls57Arg25Bands
for band in bands:
best_pixel_data[band] = empty_array(shape=shape, dtype=numpy.int16, ndv=no_data_value)
best_pixel_satellite = empty_array(shape=shape, dtype=numpy.int16, ndv=NDV)
# best_pixel_epoch = empty_array(shape=shape, dtype=numpy.int32, ndv=NDV)
best_pixel_date = empty_array(shape=shape, dtype=numpy.int32, ndv=NDV)
current_satellite = empty_array(shape=shape, dtype=numpy.int16, ndv=NDV)
# current_epoch = empty_array(shape=shape, dtype=numpy.int32, ndv=NDV)
current_date = empty_array(shape=shape, dtype=numpy.int32, ndv=NDV)
metadata = None
bands = Ls57Arg25Bands
for band in bands:
best_pixel_data[band] = empty_array(shape=shape, dtype=numpy.int16, ndv=no_data_value)
best_pixel_satellite = empty_array(shape=shape, dtype=numpy.int16, ndv=NDV)
# best_pixel_epoch = empty_array(shape=shape, dtype=numpy.int32, ndv=NDV)
best_pixel_date = empty_array(shape=shape, dtype=numpy.int32, ndv=NDV)
current_satellite = empty_array(shape=shape, dtype=numpy.int16, ndv=NDV)
# current_epoch = empty_array(shape=shape, dtype=numpy.int32, ndv=NDV)
current_date = empty_array(shape=shape, dtype=numpy.int32, ndv=NDV)
metadata = None
SATELLITE_DATA_VALUES = {Satellite.LS5: 5, Satellite.LS7: 7, Satellite.LS8: 8}
for tile in self.get_tiles(sort=SortType.DESC):
# Get ARG25 dataset
dataset = tile.datasets[DatasetType.ARG25]
_log.info("Processing ARG tile [%s]", dataset.path)
if not metadata:
metadata = get_dataset_metadata(dataset)
band_data = None
if self.apply_pq_filter:
band_data = get_dataset_data_with_pq(dataset, tile.datasets[DatasetType.PQ25])
else:
band_data = get_dataset_data(dataset)
choices=range(110, 155 + 1), required=True, metavar="110 ... 155")
self.parser.add_argument("--x-max", help="X index of tiles", action="store", dest="x_max", type=int,
choices=range(110, 155 + 1), required=True, metavar="110 ... 155")
self.parser.add_argument("--y-min", help="Y index of tiles", action="store", dest="y_min", type=int,
choices=range(-45, -10 + 1), required=True, metavar="-45 ... -10")
self.parser.add_argument("--y-max", help="Y index of tiles", action="store", dest="y_max", type=int,
choices=range(-45, -10 + 1), required=True, metavar="-45 ... -10")
self.parser.add_argument("--acq-min", help="Acquisition Date", action="store", dest="acq_min", type=str,
default="1980")
self.parser.add_argument("--acq-max", help="Acquisition Date", action="store", dest="acq_max", type=str,
default="2020")
self.parser.add_argument("--satellite", help="The satellite(s) to include", action="store", dest="satellite",
type=satellite_arg, nargs="+", choices=Satellite,
default=[Satellite.LS5, Satellite.LS7],
metavar=" ".join([s.name for s in Satellite]))
self.parser.add_argument("--mask-pqa-apply", help="Apply PQA mask", action="store_true", dest="mask_pqa_apply",
default=False)
self.parser.add_argument("--mask-pqa-mask", help="The PQA mask to apply", action="store", dest="mask_pqa_mask",
type=pqa_mask_arg, nargs="+", choices=PqaMask, default=[PqaMask.PQ_MASK_CLEAR],
metavar=" ".join([s.name for s in PqaMask]))
self.parser.add_argument("--mask-wofs-apply", help="Apply WOFS mask", action="store_true",
dest="mask_wofs_apply",
default=False)
self.parser.add_argument("--mask-wofs-mask", help="The WOFS mask to apply", action="store",
dest="mask_wofs_mask",
type=wofs_mask_arg, nargs="+", choices=WofsMask, default=[WofsMask.WET],
metavar=" ".join([s.name for s in WofsMask]))
parser.set_defaults(log_level=logging.INFO)
parser.add_argument("--lat", help="Latitude value of pixel", action="store", dest="latitude", type=float, required=True)
parser.add_argument("--lon", help="Longitude value of pixel", action="store", dest="longitude", type=float, required=True)
parser.add_argument("--acq-min", help="Acquisition Date", action="store", dest="acq_min", type=str)
parser.add_argument("--acq-max", help="Acquisition Date", action="store", dest="acq_max", type=str)
# parser.add_argument("--process-min", help="Process Date", action="store", dest="process_min", type=str)
# parser.add_argument("--process-max", help="Process Date", action="store", dest="process_max", type=str)
#
# parser.add_argument("--ingest-min", help="Ingest Date", action="store", dest="ingest_min", type=str)
# parser.add_argument("--ingest-max", help="Ingest Date", action="store", dest="ingest_max", type=str)
parser.add_argument("--satellite", help="The satellite(s) to include", action="store", dest="satellite",
type=satellite_arg, nargs="+", choices=Satellite, default=[Satellite.LS5, Satellite.LS7], metavar=" ".join([s.name for s in Satellite]))
parser.add_argument("--apply-pqa", help="Apply PQA mask", action="store_true", dest="apply_pqa", default=False)
parser.add_argument("--pqa-mask", help="The PQA mask to apply", action="store", dest="pqa_mask",
type=pqa_mask_arg, nargs="+", choices=PqaMask, default=[PqaMask.PQ_MASK_CLEAR], metavar=" ".join([s.name for s in PqaMask]))
parser.add_argument("--hide-no-data", help="Don't output records that are completely no data value(s)", action="store_false", dest="output_no_data", default=True)
supported_dataset_types = dataset_type_database + dataset_type_filesystem + dataset_type_derived_nbar
# For now only only one type of dataset per customer
parser.add_argument("--dataset-type", help="The type of dataset from which values will be retrieved", action="store",
dest="dataset_type",
type=dataset_type_arg,
#nargs="+",
choices=supported_dataset_types, default=DatasetType.ARG25, required=True, metavar=" ".join([s.name for s in supported_dataset_types]))
parser.set_defaults(log_level=logging.INFO)
parser.add_argument("--x", help="X grid reference", action="store", dest="x", type=int, choices=range(110, 155+1), required=True, metavar="[110 - 155]")
parser.add_argument("--y", help="Y grid reference", action="store", dest="y", type=int, choices=range(-45, -10+1), required=True, metavar="[-45 - -10]")
parser.add_argument("--acq-min", help="Acquisition Date", action="store", dest="acq_min", type=str, required=True)
parser.add_argument("--acq-max", help="Acquisition Date", action="store", dest="acq_max", type=str, required=True)
# parser.add_argument("--process-min", help="Process Date", action="store", dest="process_min", type=str)
# parser.add_argument("--process-max", help="Process Date", action="store", dest="process_max", type=str)
#
# parser.add_argument("--ingest-min", help="Ingest Date", action="store", dest="ingest_min", type=str)
# parser.add_argument("--ingest-max", help="Ingest Date", action="store", dest="ingest_max", type=str)
parser.add_argument("--satellite", help="The satellite(s) to include", action="store", dest="satellite",
type=satellite_arg, nargs="+", choices=Satellite, default=[Satellite.LS5, Satellite.LS7], metavar=" ".join([s.name for s in Satellite]))
parser.add_argument("--apply-pqa", help="Apply PQA mask", action="store_true", dest="apply_pqa", default=False)
parser.add_argument("--pqa-mask", help="The PQA mask to apply", action="store", dest="pqa_mask",
type=pqa_mask_arg, nargs="+", choices=PqaMask, default=[PqaMask.PQ_MASK_CLEAR], metavar=" ".join([s.name for s in PqaMask]))
supported_dataset_types = dataset_type_database + dataset_type_derived_nbar
parser.add_argument("--dataset-type", help="The types of dataset to retrieve", action="store",
dest="dataset_type",
type=dataset_type_arg,
nargs="+",
choices=supported_dataset_types, default=DatasetType.ARG25, metavar=" ".join([s.name for s in supported_dataset_types]))
parser.add_argument("--output-directory", help="Output directory", action="store", dest="output_directory",
type=writeable_dir, required=True)
Ls57Arg25Bands.GREEN: 0.0549,
Ls57Arg25Bands.RED: 0.1075,
Ls57Arg25Bands.NEAR_INFRARED: 0.1855,
Ls57Arg25Bands.SHORT_WAVE_INFRARED_1: -0.4357,
Ls57Arg25Bands.SHORT_WAVE_INFRARED_2: 0.8085},
TasselCapIndex.SIXTH: {
Ls57Arg25Bands.BLUE: 0.1084,
Ls57Arg25Bands.GREEN: -0.9022,
Ls57Arg25Bands.RED: 0.4120,
Ls57Arg25Bands.NEAR_INFRARED: 0.0573,
Ls57Arg25Bands.SHORT_WAVE_INFRARED_1: -0.0251,
Ls57Arg25Bands.SHORT_WAVE_INFRARED_2: 0.0238}
},
Satellite.LS8:
{
TasselCapIndex.BRIGHTNESS: {
Ls8Arg25Bands.BLUE: 0.3029,
Ls8Arg25Bands.GREEN: 0.2786,
Ls8Arg25Bands.RED: 0.4733,
Ls8Arg25Bands.NEAR_INFRARED: 0.5599,
Ls8Arg25Bands.SHORT_WAVE_INFRARED_1: 0.508,
Ls8Arg25Bands.SHORT_WAVE_INFRARED_2: 0.1872},
TasselCapIndex.GREENNESS: {
Ls8Arg25Bands.BLUE: -0.2941,
Ls8Arg25Bands.GREEN: -0.2430,
Ls8Arg25Bands.RED: -0.5424,
Ls8Arg25Bands.NEAR_INFRARED: 0.7276,
Ls8Arg25Bands.SHORT_WAVE_INFRARED_1: 0.0713,
Ls8Arg25Bands.SHORT_WAVE_INFRARED_2: -0.1608},
self.parser.add_argument("--acq-max", help="Acquisition Date", action="store", dest="acq_max", type=str,
default="2020")
self.parser.add_argument("--season", help="Seasonal acquisition range within acquisition period (e.g. --season WINTER 06 08 means WINTER from JUN-01 to AUG-30)",
action="store", dest="season", type=str, nargs=3, metavar=" ")
# parser.add_argument("--process-min", help="Process Date", action="store", dest="process_min", type=str)
# parser.add_argument("--process-max", help="Process Date", action="store", dest="process_max", type=str)
#
# parser.add_argument("--ingest-min", help="Ingest Date", action="store", dest="ingest_min", type=str)
# parser.add_argument("--ingest-max", help="Ingest Date", action="store", dest="ingest_max", type=str)
self.parser.add_argument("--satellite", help="The satellite(s) to include", action="store", dest="satellite",
type=satellite_arg, nargs="+", choices=Satellite,
default=[Satellite.LS5, Satellite.LS7],
metavar=" ".join([s.name for s in Satellite]))
self.parser.add_argument("--mask-pqa-apply", help="Apply PQA mask", action="store_true", dest="mask_pqa_apply",
default=False)
self.parser.add_argument("--mask-pqa-mask", help="The PQA mask to apply", action="store", dest="mask_pqa_mask",
type=pqa_mask_arg, nargs="+", choices=PqaMask, default=[PqaMask.PQ_MASK_CLEAR],
metavar=" ".join([s.name for s in PqaMask]))
self.parser.add_argument("--mask-wofs-apply", help="Apply WOFS mask", action="store_true",
dest="mask_wofs_apply",
default=False)
self.parser.add_argument("--mask-wofs-mask", help="The WOFS mask to apply", action="store",
dest="mask_wofs_mask",
type=wofs_mask_arg, nargs="+", choices=WofsMask, default=[WofsMask.WET],
def get_filename_file_list(satellite, dataset_type, x, y):
satellite_str = "LS"
if satellite == Satellite.LS8 and dataset_type == DatasetType.ARG25:
satellite_str += "8"
dataset_type_str = {
DatasetType.ARG25: "NBAR",
DatasetType.PQ25: "PQA",
DatasetType.FC25: "FC",
DatasetType.NDVI: "NDVI",
DatasetType.EVI: "EVI",
DatasetType.NBR: "NBR"
}[dataset_type]
return "{satellite}_{dataset}_{x:03d}_{y:04d}.files.txt".format(satellite=satellite_str, dataset=dataset_type, x=x, y=y)
return os.path.join(self.output_directory,"LS_WOFS_{longitude:03.5f}_{latitude:03.5f}_{acq_min}_{acq_max}.csv".format(latitude=self.latitude,
longitude=self.longitude,
acq_min=self.acq_min,
acq_max=self.acq_max))
satellite_str = ""
if Satellite.LS5 in self.satellites or Satellite.LS7 in self.satellites or Satellite.LS8 in self.satellites:
satellite_str += "LS"
if Satellite.LS5 in self.satellites:
satellite_str += "5"
if Satellite.LS7 in self.satellites:
satellite_str += "7"
if Satellite.LS8 in self.satellites:
satellite_str += "8"
dataset_str = ""
if dataset_type == DatasetType.ARG25:
dataset_str += "NBAR"
elif dataset_type == DatasetType.PQ25:
dataset_str += "PQA"
elif dataset_type == DatasetType.FC25:
dataset_str += "FC"
elif dataset_type == DatasetType.WATER:
dataset_str += "WOFS"
def decode_satellite_as_instrument(satellite):
instruments = {Satellite.LS5: "TM", Satellite.LS7: "ETM", Satellite.LS8: "OLI"}
if satellite in instruments:
return instruments[satellite]
return "N/A"