Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def get_cell_csv_filename(self):
acq_min = format_date(self.acq_min)
acq_max = format_date(self.acq_max)
# TODO other distinguishing characteristics (e.g. dataset type(s))
return os.path.join(
self.output_directory,
"cells_{satellites}_{x_min:03d}_{x_max:03d}_{y_min:04d}_{y_max:04d}_{acq_min}_{acq_max}.csv".format(
satellites=get_satellite_string(self.satellites), x_min=self.x_min,
x_max=self.x_max, y_min=self.y_min, y_max=self.y_max,
acq_min=acq_min, acq_max=acq_max
))
def generate_raster_metadata(self):
return {
"X_INDEX": "{x:03d}".format(x=self.x),
"Y_INDEX": "{y:04d}".format(y=self.y),
"DATASET_TYPE": self.dataset_type.name,
"ACQUISITION_DATE": "{acq_min} to {acq_max}".format(acq_min=format_date(self.acq_min), acq_max=format_date(self.acq_max)),
"SATELLITES": " ".join([s.name for s in self.satellites]),
"PIXEL_QUALITY_FILTER": self.mask_pqa_apply and " ".join([mask.name for mask in self.mask_pqa_mask]) or "",
"WATER_FILTER": self.mask_wofs_apply and " ".join([mask.name for mask in self.mask_wofs_mask]) or ""
}
def get_tile_csv_filename(self):
acq_min = format_date(self.acq_min)
acq_max = format_date(self.acq_max)
# TODO other distinguishing characteristics (e.g. dataset types)
return os.path.join(
self.output_directory,
"tiles_{satellites}_{x_min:03d}_{x_max:03d}_{y_min:04d}_{y_max:04d}_{acq_min}_{acq_max}.csv".format(
satellites=get_satellite_string(self.satellites),
x_min=self.x, x_max=self.x, y_min=self.y, y_max=self.y,
acq_min=acq_min, acq_max=acq_max
))
def generate_raster_metadata(self):
return {
"X_INDEX": "{x:03d}".format(x=self.x),
"Y_INDEX": "{y:04d}".format(y=self.y),
"DATASET_TYPE": self.dataset_type.name,
"ACQUISITION_DATE": "{acq_min} to {acq_max}".format(acq_min=format_date(self.acq_min), acq_max=format_date(self.acq_max)),
"SATELLITES": " ".join([s.name for s in self.satellites]),
"PIXEL_QUALITY_FILTER": self.mask_pqa_apply and " ".join([mask.name for mask in self.mask_pqa_mask]) or "",
"WATER_FILTER": self.mask_wofs_apply and " ".join([mask.name for mask in self.mask_wofs_mask]) or ""
}
def get_tile_csv_filename(self):
acq_min = format_date(self.acq_min)
acq_max = format_date(self.acq_max)
# TODO other distinguishing characteristics (e.g. dataset types)
return os.path.join(
self.output_directory,
"tiles_{satellites}_{x_min:03d}_{x_max:03d}_{y_min:04d}_{y_max:04d}_{acq_min}_{acq_max}.csv".format(
satellites=get_satellite_string(self.satellites),
x_min=self.x, x_max=self.x, y_min=self.y, y_max=self.y,
acq_min=acq_min, acq_max=acq_max
))
def generate_raster_metadata(self):
return {
"X_INDEX": "{x:03d}".format(x=self.x),
"Y_INDEX": "{y:04d}".format(y=self.y),
"DATASET_TYPE": "{dataset_type} STACK".format(dataset_type=self.dataset_type.name),
"ACQUISITION_DATE": "{acq_min} to {acq_max}".format(acq_min=format_date(self.acq_min), acq_max=format_date(self.acq_max)),
"SEASON": self.season.name,
"SATELLITES": " ".join([s.name for s in self.satellites]),
"PIXEL_QUALITY_FILTER": self.mask_pqa_apply and " ".join([mask.name for mask in self.mask_pqa_mask]) or ""
}
def get_tile_csv_filename(self):
acq_min = format_date(self.acq_min)
acq_max = format_date(self.acq_max)
# TODO other distinguishing characteristics (e.g. dataset types)
return os.path.join(
self.output_directory,
"tiles_{satellites}_{x_min:03d}_{x_max:03d}_{y_min:04d}_{y_max:04d}_{acq_min}_{acq_max}.csv".format(
satellites=get_satellite_string(self.satellites),
x_min=self.x, x_max=self.x, y_min=self.y, y_max=self.y,
acq_min=acq_min, acq_max=acq_max
))
mask = get_mask_wofs(wofs, self.mask_wofs_mask, mask=mask)
# _log.info("mask[3500,3500] is [%s]", mask[3500, 3500])
data = get_dataset_data_masked(dataset, mask=mask, ndv=ndv)
_log.debug("data is [%s]", data)
# _log.info("data[3500,3500] is [%s]", data[band][3500, 3500])
stack_band = raster.GetRasterBand(index)
stack_band.SetDescription(os.path.basename(dataset.path))
stack_band.SetNoDataValue(ndv)
stack_band.WriteArray(data[band])
stack_band.ComputeStatistics(True)
stack_band.SetMetadata({"ACQ_DATE": format_date(tile.end_datetime), "SATELLITE": dataset.satellite.name})
stack_band.FlushCache()
del stack_band
if raster:
raster.FlushCache()
raster = None
del raster