Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_products_are_valid(dea_index: Index):
for product in dea_index.products.get_all():
validation_messages = [str(m) for m in eodatasets3.validate.validate_product(product.definition)]
assert validation_messages == [], f"{product.name} has validation issues: {validation_messages}"
def _verify_checksum(log, metadata_path, dry_run=True):
dataset_path, all_files = path_utils.get_dataset_paths(metadata_path)
checksum_file = _expected_checksum_path(dataset_path)
if not checksum_file.exists():
# Ingested data doesn't currently have them, so it's only a warning.
log.warning("checksum.missing", checksum_file=checksum_file)
return None
ch = verify.PackageChecksum()
ch.read(checksum_file)
if not dry_run:
for file, successful in ch.iteratively_verify():
if successful:
log.debug("checksum.pass", file=file)
else:
log.error("checksum.failure", file=file)
return False
log.debug("copy.verify", file_count=len(all_files))
return True
def _convert_old_odc_dataset_to_new(ds: Dataset) -> DatasetDoc:
product = ProductDoc(name=ds.type.name)
properties = StacPropertyView(ds.metadata_doc['properties'])
return DatasetDoc(
id=ds.id,
product=product,
crs=ds.crs.crs_str,
properties=properties
def _convert_old_odc_dataset_to_new(ds: Dataset) -> DatasetDoc:
product = ProductDoc(name=ds.type.name)
properties = StacPropertyView(ds.metadata_doc['properties'])
return DatasetDoc(
id=ds.id,
product=product,
crs=ds.crs.crs_str,
properties=properties
def _convert_old_odc_dataset_to_new(ds: Dataset) -> DatasetDoc:
product = ProductDoc(name=ds.type.name)
properties = StacPropertyView(ds.metadata_doc['properties'])
return DatasetDoc(
id=ds.id,
product=product,
crs=ds.crs.crs_str,
properties=properties
log.info('loaded and transformed')
dtypes = set(str(v.dtype) for v in output_data.data_vars.values())
if 'int8' in dtypes:
log.info('Found dtype=int8 in output data, converting to uint8 for geotiffs')
output_data = output_data.astype('uint8', copy=False)
if 'crs' not in output_data.attrs:
output_data.attrs['crs'] = crs
source_doc = _convert_old_odc_dataset_to_new(task.dataset)
# Ensure output path exists
output_location = Path(task.settings.output.location)
output_location.mkdir(parents=True, exist_ok=True)
with DatasetAssembler(output_location, naming_conventions="dea") as p:
p.add_source_dataset(source_doc, auto_inherit_properties=True)
# Copy in metadata and properties
for k, v in task.settings.output.metadata.items():
setattr(p, k, v)
for k, v in task.settings.output.properties.items():
p.properties[k] = v
p.processed = datetime.utcnow()
p.note_software_version(
'd2dtransformer',
"https://github.com/GeoscienceAustralia/digitalearthau",
"0.1.0"
)