sliceosm

A script to download OpenStreetMap extracts from SliceOSM.

#!/usr/bin/env -S uv run --script
# /// script
# requires-python = ">=3.12"
# dependencies = [
#   "typer==0.16.0",
#   "httpx==0.28.1",
#   "tqdm==4.67.1",
#   "geojson==3.2.0",
# ]
# ///
# SPDX-FileCopyrightText: 2025 kurt.town
# SPDX-License-Identifier: MIT

from pathlib import Path
import sys
from time import sleep
import geojson
import httpx
from typing import Annotated, Literal, TypedDict
from tqdm import tqdm
import typer

class Status(TypedDict):
    Timestamp: str
    CellsTotal: int
    CellsProg: int
    NodesTotal: int
    NodesProg: int
    ElemsTotal: int
    ElemsProg: int
    SizeBytes: int
    Elapsed: float
    Complete: bool

def bbox_request(bbox: list[float]) -> str:
    min_lon, min_lat, max_lon, max_lat = bbox[0], bbox[1], bbox[2], bbox[3]

    return {
        "Name": "none",
        "RegionType": "geojson",
        "RegionData": geojson.Polygon(
            coordinates=[[
                [min_lon, min_lat],
                [min_lon, max_lat],
                [max_lon, max_lat],
                [max_lon, min_lat],
                [min_lon, min_lat],
            ]])
    }

def geojson_request(gj: geojson.Polygon | geojson.MultiPolygon) -> str:
    return {
        "Name": "none",
        "RegionType": "geojson",
        "RegionData": gj,
    }

SHORTCUTS = {
    "istanbul": bbox_request([27.881927,40.688544,30.016022,41.651423]),
    #"mock": bbox_request([27, 27, 27.0001, 27.0001]),
    "bozcaada": bbox_request([25.940266,39.770995,26.108494,39.875477])
}

def update_pbar(pbar: tqdm | None, pbar_at: Literal["cell"] | Literal["node"] | Literal["elem"] | None, status: Status) -> tuple[tqdm, Literal["cell"] | Literal["node"] | Literal["elem"]]:
    if pbar is None and pbar_at is None:
        pbar = tqdm(
            total=status['CellsTotal'],
            desc="Processing cells",
            unit="cell",
            initial=status['CellsProg'],
        )
        return pbar, "cell"
    elif pbar_at == "cell":
        if pbar.n < status['CellsProg']:
            pbar.update(status['CellsProg'] - pbar.n)
        if status['CellsProg'] == status['CellsTotal']:
            pbar.close()
            pbar = tqdm(
                total=status['NodesTotal'],
                desc="Processing nodes",
                unit="node",
                initial=status['NodesProg'],
            )
            pbar_at = "node"
        return pbar, pbar_at
    elif pbar_at == "node":
        if pbar.n < status['NodesProg']:
            pbar.update(status['NodesProg'] - pbar.n)
        if status['NodesProg'] == status['NodesTotal']:
            pbar.close()
            pbar = tqdm(
                total=status['ElemsTotal'],
                desc="Processing elements",
                unit="elem",
                initial=status['ElemsProg'],
            )
            pbar_at = "elem"
        return pbar, pbar_at
    elif pbar_at == "elem":
        if pbar.n < status['ElemsProg']:
            pbar.update(status['ElemsProg'] - pbar.n)
        if status['ElemsProg'] == status['ElemsTotal']:
            pbar.close()
            pbar = None
            pbar_at = None
        return pbar, pbar_at

def main(
        shortcut: Annotated[str, typer.Option("-s", "--shortcut", help=f"Name of the area, one of {', '.join(SHORTCUTS.keys())}")] = None,
        bbox: Annotated[str, typer.Option("-b", "--bbox", help="A bbox in the format 'min_lon,min_lat,max_lon,max_lat'.")] = None,
        geojson_file: Annotated[Path, typer.Option("-g", "--geojson", exists=True, help="A geojson file containing a Polygon or MultiPolygon.")] = None,
        output: Annotated[Path, typer.Option("-o", "--output", writable=True, resolve_path=True, file_okay=True, dir_okay=False, help="File path for the downloaded extract.")] = Path("slice.osm.pbf"),
) -> None:
    """
    Download OpenStreetMap extracts from SliceOSM (https://slice.openstreetmap.us/).
    """
    req = None

    if not any([shortcut, bbox, geojson_file]):
        raise typer.BadParameter("You must provide at least one of --shortcut, --bbox, or --geojson.")
    if shortcut and shortcut not in SHORTCUTS:
        raise typer.BadParameter(f"Shortcut '{shortcut}' not found. Available shortcuts: {', '.join(SHORTCUTS.keys())}.")
    if bbox and geojson:
        raise typer.BadParameter("You cannot provide both --bbox and --geojson.")
    if output.exists():
        raise typer.BadParameter(f"Output file '{output}' already exists. Please choose a different file name or remove the existing file.")

    if shortcut:
        req = SHORTCUTS[shortcut]
    elif bbox:
        bbox_coords = [float(coord) for coord in bbox.split(",")]
        if len(bbox_coords) != 4:
            raise typer.BadParameter("bbox must be in the format 'min_lon,min_lat,max_lon,max_lat'.")
        req = bbox_request(bbox_coords)
    elif geojson_file:
        with open(geojson_file, "r") as f:
            gj = geojson.load(f)
            if not isinstance(gj, (geojson.Polygon, geojson.MultiPolygon)):
                raise typer.BadParameter("geojson must be a Polygon or MultiPolygon.")
            req = geojson_request(gj)

    resp = httpx.post(
        "https://slice.openstreetmap.us/api/",
        json=req,
    )

    uuid = resp.text

    print(f"Request UUID: {uuid}", file=sys.stderr)

    status: Status = httpx.get(
        f"https://slice.openstreetmap.us/api/{uuid}",
    ).json()

    pbar, pbar_at = update_pbar(None, None, status)

    while not status["Complete"]:
        sleep(1)
        status = httpx.get(
            f"https://slice.openstreetmap.us/api/{uuid}",
        ).json()
        pbar, pbar_at = update_pbar(pbar, pbar_at, status)

    if pbar is not None:
        pbar.close()

    content_length = status["SizeBytes"]

    with open(output, "wb") as f:
        with httpx.stream("GET", f'https://slice.openstreetmap.us/api/{uuid}.osm.pbf') as response:
            num_bytes_downloaded = 0
            pbar = tqdm(
                total=round(content_length / (1000000), 2),
                desc=f"Downloading {output.name}",
                unit="MB",
                initial=round(num_bytes_downloaded / (1000000), 2),
                bar_format='{l_bar}{bar}| {n:.2f}/{total:.2f} {unit} [{elapsed}<{remaining}]',
            )
            for chunk in response.iter_bytes():
                f.write(chunk)
                num_bytes_downloaded += len(chunk)
                pbar.update(num_bytes_downloaded / (1000000) - pbar.n)
            pbar.close()


if __name__ == "__main__":
    typer.run(main)