Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
except AssertionError:
warnings.warn(
"irreparable geometry found in vector input file"
)
continue
geom = clean_geometry_type(
feature_geom.intersection(dst_bbox),
feature_geom.geom_type
)
if geom:
# Reproject each feature to tile CRS
if vector_crs == dst_crs and validity_check:
assert geom.is_valid
else:
try:
geom = reproject_geometry(
geom,
src_crs=vector_crs,
dst_crs=dst_crs,
validity_check=validity_check
)
except ValueError:
warnings.warn("feature reprojection failed")
yield {
'properties': feature['properties'],
'geometry': mapping(geom)
}
"file": None,
"area": None
}
else:
abs_path = os.path.join(mapchete_config.config_dir, input_file)
try:
# Sentinel-2 datasets can be in directories as well
assert os.path.isfile(abs_path) or os.path.isdir(abs_path)
except AssertionError:
raise IOError("no such file", abs_path)
extension = os.path.splitext(abs_path)[1]
if extension == ".mapchete":
mapchete_process = Mapchete(MapcheteConfig(abs_path))
prepared = {
"file": mapchete_process,
"area": reproject_geometry(
mapchete_process.config.process_area(),
mapchete_process.tile_pyramid.crs,
mapchete_config.tile_pyramid.crs
)
}
elif extension in [".SAFE", ".zip", ".ZIP"]:
prepared = {
"file": _prepare_sentinel2(SentinelDataSet(input_file)),
"area": file_bbox(abs_path, mapchete_config.tile_pyramid)
}
else:
prepared = {
"file": abs_path,
"area": file_bbox(abs_path, mapchete_config.tile_pyramid)
}
dst_bounds=None,
dst_crs=None,
validity_check=None
):
assert isinstance(input_file, str)
assert isinstance(dst_bounds, tuple)
assert isinstance(dst_crs, CRS)
assert isinstance(validity_check, bool)
with fiona.open(input_file, 'r') as vector:
vector_crs = CRS(vector.crs)
# Reproject tile bounding box to source file CRS for filter:
if vector_crs == dst_crs:
dst_bbox = box(*dst_bounds)
else:
dst_bbox = reproject_geometry(
box(*dst_bounds),
src_crs=dst_crs,
dst_crs=vector_crs,
validity_check=True
)
for feature in vector.filter(bbox=dst_bbox.bounds):
feature_geom = shape(feature['geometry'])
if not feature_geom.is_valid:
try:
feature_geom = feature_geom.buffer(0)
assert feature_geom.is_valid
warnings.warn(
"fixed invalid vector input geometry"
)
except AssertionError:
warnings.warn(
def _reproject_tile_bbox(self, out_crs=None):
"""
Returns tile bounding box reprojected to source file CRS. If bounding
box overlaps with antimeridian, a MultiPolygon is returned.
"""
return reproject_geometry(
clip_geometry_to_srs_bounds(
self.tile.bbox(pixelbuffer=self.pixelbuffer),
self.tile.tile_pyramid
),
self.tile.crs,
out_crs
)