Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
for cleantopo_process in [
"testdata/cleantopo_tl.mapchete", "testdata/cleantopo_br.mapchete"
]:
try:
shutil.rmtree(out_dir)
pass
except:
pass
mapchete_file = os.path.join(scriptdir, cleantopo_process)
process = Mapchete(MapcheteConfig(mapchete_file))
for zoom in range(6):
tiles = []
for tile in process.get_process_tiles(zoom):
output = process.execute(tile)
tiles.append(output)
assert isinstance(output, BufferedTile)
assert isinstance(output.data, ma.MaskedArray)
assert output.data.shape == output.shape
assert not ma.all(output.data.mask)
process.write(output)
tilenum += 1
mosaic, mosaic_affine = create_mosaic(tiles)
try:
temp_vrt = os.path.join(out_dir, str(zoom)+".vrt")
gdalbuildvrt = "gdalbuildvrt %s %s/%s/*/*.tif > /dev/null" % (
temp_vrt, out_dir, zoom)
os.system(gdalbuildvrt)
with rasterio.open(temp_vrt, "r") as testfile:
for file_item, mosaic_item in zip(
testfile.meta["transform"], mosaic_affine
):
try:
process_tile : ``BufferedTile``
must be member of process ``TilePyramid``
"""
data = prepare_array(
data,
masked=True,
nodata=self.output_params["nodata"],
dtype=self.profile(process_tile)["dtype"]
)
if data.mask.all():
logger.debug("data empty, nothing to write")
else:
# Convert from process_tile to output_tiles and write
for tile in self.pyramid.intersecting(process_tile):
out_tile = BufferedTile(tile, self.pixelbuffer)
write_window = from_bounds(
*out_tile.bounds,
transform=self.rio_file.transform,
height=self.rio_file.height,
width=self.rio_file.width
).round_lengths(pixel_precision=0).round_offsets(pixel_precision=0)
if _window_in_out_file(write_window, self.rio_file):
logger.debug("write data to window: %s", write_window)
self.rio_file.write(
extract_from_array(
in_raster=data,
in_affine=process_tile.affine,
out_tile=out_tile
) if process_tile != out_tile else data,
window=write_window,
)
Parameters
----------
zoom : integer
zoom level
row : integer
tile matrix row
col : integer
tile matrix column
Returns
-------
buffered tile : ``BufferedTile``
"""
tile = self.tile_pyramid.tile(zoom, row, col)
return BufferedTile(tile, pixelbuffer=self.pixelbuffer)
def __init__(self, tile, pixelbuffer=0):
"""Initialize."""
if isinstance(tile, BufferedTile):
tile = TilePyramid(
tile.tp.grid,
tile_size=tile.tp.tile_size,
metatiling=tile.tp.metatiling
).tile(*tile.id)
Tile.__init__(self, tile.tile_pyramid, tile.zoom, tile.row, tile.col)
self._tile = tile
self.pixelbuffer = pixelbuffer
must be member of process ``TilePyramid``
"""
rgba = self._prepare_array_for_png(data)
data = ma.masked_where(rgba == self.output_params["nodata"], rgba)
if data.mask.all():
logger.debug("data empty, nothing to write")
else:
# in case of S3 output, create an boto3 resource
bucket_resource = get_boto3_bucket(self._bucket) if self._bucket else None
# Convert from process_tile to output_tiles and write
for tile in self.pyramid.intersecting(process_tile):
out_path = self.get_path(tile)
self.prepare_path(tile)
out_tile = BufferedTile(tile, self.pixelbuffer)
write_raster_window(
in_tile=process_tile,
in_data=data,
out_profile=self.profile(out_tile),
out_tile=out_tile,
out_path=out_path,
bucket_resource=bucket_resource
)
----------
tile : tuple or BufferedTile
pyramid : BufferedTilePyramid
pyramid tile is being generated from if tile is tuple
Returns
-------
BufferedTile
Raises
------
TypeError if tile type is invalid.
"""
if isinstance(tile, tuple):
return pyramid.tile(*tile)
elif isinstance(tile, BufferedTile):
return tile
else:
raise TypeError("tile must be tuple or BufferedTile: %s" % tile)
def get_parent(self):
"""
Get tile parent (intersecting tile in previous zoom level).
Returns
-------
parent : ``BufferedTile``
"""
return BufferedTile(self._tile.get_parent(), self.pixelbuffer)
Parameters
----------
output_tile : BufferedTile or tile index tuple
Member of the output tile pyramid (not necessarily the process
pyramid, if output has a different metatiling setting)
Returns
-------
data : NumPy array or features
process output
"""
if self.config.mode not in ["readonly", "continue", "overwrite"]:
raise ValueError("process mode must be readonly, continue or overwrite")
if isinstance(output_tile, tuple):
output_tile = self.config.output_pyramid.tile(*output_tile)
elif isinstance(output_tile, BufferedTile):
pass
else:
raise TypeError("output_tile must be tuple or BufferedTile")
return self.config.output.read(output_tile)