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_mapchete_init():
"""Raise TypeError if not MapcheteConfig object is passed."""
with pytest.raises(TypeError):
mapchete.Mapchete("wrong_type")
except:
print "FAILED: read data size"
"""processing"""
tilenum = 0
out_dir = os.path.join(scriptdir, "testdata/tmp")
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)
assert set(['Mapchete', 'raster_file', 'vector_file']).issubset(
set(available_input_formats()))
assert set(['GTiff', 'PNG', 'PNG_hillshade']).issubset(
set(available_output_formats()))
ext_to_driver = _file_ext_to_driver()
assert isinstance(ext_to_driver, dict)
assert set(['mapchete', 'tif', 'jp2', 'png', 'vrt']).issubset(
set(ext_to_driver))
for extension, driver in ext_to_driver.iteritems():
assert len(driver) == 1
"""config and base module"""
# Load source process from python file and initialize.
mapchete_file = os.path.join(scriptdir, "example.mapchete")
config = MapcheteConfig(mapchete_file)
process = Mapchete(config)
dummy1_abspath = os.path.join(scriptdir, "testdata/dummy1.tif")
dummy2_abspath = os.path.join(scriptdir, "testdata/dummy2.tif")
# Validate configuration constructor
## basic run through
try:
config = process.config
print "OK: basic configuraiton constructor run through"
except:
print "FAILED: basic configuraiton constructor run through"
raise
try:
# Check configuration at zoom level 5
zoom5 = config.at_zoom(5)
raise ValueError(
"%s zoom %s: mosaic values do not fit" % (
cleantopo_process, zoom))
except:
raise
finally:
try:
os.remove(temp_vrt)
shutil.rmtree(out_dir)
except:
pass
print "OK: tile properties from %s tiles" % tilenum
"""multiprocessing"""
mapchete_file = os.path.join(scriptdir, "testdata/cleantopo_tl.mapchete")
process = Mapchete(MapcheteConfig(mapchete_file))
assert dumps(process)
assert dumps(process.config)
assert dumps(process.config.output)
for tile in process.get_process_tiles():
assert dumps(tile)
out_dir = os.path.join(scriptdir, "testdata/tmp")
f = partial(worker, process)
try:
for zoom in reversed(process.config.zoom_levels):
pool = Pool()
try:
for raw_output in pool.imap_unordered(
f, process.get_process_tiles(zoom), chunksize=8):
process.write(raw_output)
except KeyboardInterrupt:
pool.terminate()
pool.close()
pool.join()
print "OK: multiprocessing"
except:
raise
print "FAILED: multiprocessing"
finally:
try:
shutil.rmtree(out_dir)
except:
pass
"""Vector data IO"""
mapchete_file = os.path.join(scriptdir, "testdata/geojson.mapchete")
process = Mapchete(MapcheteConfig(mapchete_file))
out_dir = os.path.join(scriptdir, "testdata/tmp")
try:
f = partial(worker, process)
pool = Pool()
try:
for output in pool.imap_unordered(
f, process.get_process_tiles(4), chunksize=1):
assert isinstance(output, BufferedTile)
if output.data:
for feature in output.data:
assert "properties" in feature
assert shape(feature["geometry"]).is_valid
else:
assert isinstance(output.data, list)
process.write(output)
except KeyboardInterrupt:
else:
print "OK: basic configuration parsing"
## read zoom level from config file
mapchete_file = os.path.join(scriptdir, "testdata/zoom.mapchete")
config = Mapchete(MapcheteConfig(mapchete_file)).config
try:
assert 5 in config.zoom_levels
print "OK: read zoom level from config file"
except:
print "FAILED: read zoom level from config file"
print mapchete_file
raise
## read min/max zoom levels from config file
mapchete_file = os.path.join(scriptdir, "testdata/minmax_zoom.mapchete")
config = Mapchete(MapcheteConfig(mapchete_file)).config
try:
for zoom in [7, 8, 9, 10]:
assert zoom in config.zoom_levels
print "OK: read min/max zoom levels from config file"
except:
print "FAILED: read min/max zoom levels from config file"
raise
## zoom levels override
mapchete_file = os.path.join(scriptdir, "testdata/minmax_zoom.mapchete")
config = Mapchete(MapcheteConfig(mapchete_file, zoom=[1, 4])).config
try:
for zoom in [1, 2, 3, 4]:
assert zoom in config.zoom_levels
print "OK: zoom levels override"
except:
print "FAILED: zoom levels override"
if not input_file in mapchete_config.prepared_input_files:
if not input_file:
prepared = {
"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,