Skip to content

Instantly share code, notes, and snippets.

@maxmalynowsky
Last active May 5, 2022 16:15
Show Gist options
  • Star 1 You must be signed in to star a gist
  • Fork 0 You must be signed in to fork a gist
  • Save maxmalynowsky/4b9744d292e5a2ffe8d0cae1ce948455 to your computer and use it in GitHub Desktop.
Save maxmalynowsky/4b9744d292e5a2ffe8d0cae1ce948455 to your computer and use it in GitHub Desktop.
import subprocess
import zipfile
import logging
import shutil
from multiprocessing import Pool
from pathlib import Path
from urllib import request
MAX_ZOOM = 10
REMOTE_URL = 'https://data.fieldmaps.io'
cwd = Path(__file__).parent
tmp = cwd / 'tmp'
outputs = cwd / 'outputs'
logging.basicConfig(level=logging.INFO,
format='%(asctime)s - %(name)s - %(message)s',
datefmt='%Y-%m-%d %H:%M:%S')
logger = logging.getLogger(__name__)
def admx_polygons(l):
zip_path = tmp / f'adm{l}_polygons.gpkg.zip'
gpkg_path = tmp / f'adm{l}_polygons.gpkg'
geojsonl_path = tmp / f'adm{l}_polygons.geojsonl'
tile_path = outputs / f'adm{l}_polygons'
r = request.build_opener()
r.addheaders = [('User-agent', 'Mozilla/5.0')]
request.install_opener(r)
if l > 0:
url = f"{REMOTE_URL}/edge-matched/humanitarian/intl/adm{l}_polygons.gpkg.zip"
else:
url = f"{REMOTE_URL}/adm0/intl/adm0_polygons.gpkg.zip"
request.urlretrieve(url, zip_path)
logger.info(f'downloaded: adm{l}_polygons')
with zipfile.ZipFile(zip_path, 'r') as z:
z.extractall(tmp)
zip_path.unlink(missing_ok=True)
logger.info(f'extracted: adm{l}_polygons')
subprocess.run(['ogr2ogr', geojsonl_path, gpkg_path],
stderr=subprocess.DEVNULL)
logger.info(f'converted: adm{l}_polygons')
gpkg_path.unlink(missing_ok=True)
options = [
'tippecanoe',
f'--layer=adm{l}_polygons',
f'--maximum-zoom={MAX_ZOOM}',
'--simplify-only-low-zooms',
'--detect-shared-borders',
'--read-parallel',
'--no-tile-size-limit',
'--no-tile-compression',
'--force',
f'--output-to-directory={tile_path}',
geojsonl_path,
]
for x in range(0, 5):
if l >= x:
options.append(f'--include=adm{x}_id')
options.append(f'--include=adm{x}_name')
subprocess.run(options, stderr=subprocess.DEVNULL)
geojsonl_path.unlink(missing_ok=True)
logger.info(f'tiled: adm{l}_polygons')
if __name__ == '__main__':
shutil.rmtree(tmp, ignore_errors=True)
shutil.rmtree(outputs, ignore_errors=True)
tmp.mkdir(parents=True, exist_ok=True)
outputs.mkdir(parents=True, exist_ok=True)
results = []
pool = Pool()
for l in range(0, 5):
result = pool.apply_async(admx_polygons, args=[l])
results.append(result)
pool.close()
pool.join()
shutil.rmtree(tmp, ignore_errors=True)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment