🗺️ OpenStreetMap Countries GeoJSON — updated daily!
from anyio import Path

NAME = 'osm-countries-geojson'
VERSION = '1.0'
WEBSITE = 'https://github.com/Zaczero/osm-countries-geojson'
USER_AGENT = f'{NAME}/{VERSION} (+{WEBSITE})'

OVERPASS_API_URL = 'https://overpass.monicz.dev/api/interpreter'
COUNTRIES_GEOJSON_URL = 'https://raw.githubusercontent.com/nvkelso/natural-earth-vector/master/geojson/ne_110m_admin_0_sovereignty.geojson'

GEOJSON_DIR = Path('geojson')

GEOJSON_QUALITIES = (
    0.1,
    0.01,
    0.001,
    0.0001,
    0.00001,
)

BEST_GEOJSON_QUALITY = min(GEOJSON_QUALITIES)

config.py
Zaczero/osm-countries-geojson
import gzip
from decimal import Decimal

import anyio
import brotli
import orjson
from shapely.geometry import mapping
from tqdm import tqdm

from config import GEOJSON_DIR, GEOJSON_QUALITIES
from natural_earth import validate_countries
from osm_countries_gen import get_osm_countries


async def main():
    # ensure output directory exists
    await GEOJSON_DIR.mkdir(exist_ok=True)

    countries, data_timestamp = await get_osm_countries()

    await validate_countries(countries)

    for q in tqdm(GEOJSON_QUALITIES, desc='Writing GeoJSON'):
        q_str = f'{Decimal(str(q)):f}'.replace('.', '-')
        path = GEOJSON_DIR / f'osm-countries-{q_str}.geojson'

        features = []

        for country in countries:
            features.append({
                'type': 'Feature',
                'properties': {
                    'tags': country.tags,
                    'timestamp': data_timestamp,
                    'representative_point': mapping(country.representative_point),
                },
                'geometry': mapping(country.geometry[q]),
            })

        data = {
            'type': 'FeatureCollection',
            'name': path.stem,
            'crs': {
                'type': 'name',
                'properties': {
                    'name': 'urn:ogc:def:crs:OGC:1.3:CRS84',
                },
            },
            'features': features,
        }

        buffer = orjson.dumps(data)

        # uncompressed
        await path.write_bytes(buffer)

        # gzip compressed
        gz_path = path.with_suffix('.geojson.gz')
        await gz_path.write_bytes(gzip.compress(buffer, compresslevel=9))

        # brotli compressed
        br_path = path.with_suffix('.geojson.br')
        await br_path.write_bytes(brotli.compress(buffer, mode=brotli.MODE_TEXT, quality=11))


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

main.py
Zaczero/osm-countries-geojson

from typing import NamedTuple

from shapely.geometry import MultiPolygon, Point, Polygon


class OSMCountry(NamedTuple):
    tags: dict[str, str]
    geometry: dict[float, Polygon | MultiPolygon]
    representative_point: Point

models/osm_country.py
Zaczero/osm-countries-geojson
from datetime import timedelta
from typing import Sequence

from shapely.geometry import shape
from tqdm import tqdm

from config import BEST_GEOJSON_QUALITY, COUNTRIES_GEOJSON_URL
from models.osm_country import OSMCountry
from utils import get_http_client, retry_exponential


@retry_exponential(timedelta(minutes=30))
async def _get_countries() -> Sequence[dict]:
    async with get_http_client() as http:
        r = await http.get(COUNTRIES_GEOJSON_URL)
        r.raise_for_status()
    return r.json()['features']


async def validate_countries(countries: Sequence[OSMCountry]) -> None:
    ne_countries = await _get_countries()
    ne_countries = tuple(c for c in ne_countries if not any(
        n in c['properties']['NAME'] for n in (
            # ignore some entries
            'Antarctica',
            'Sahara',
        )))

    # validate country geometries by checking if the representative point is inside the geometry
    for ne_country in tqdm(ne_countries, desc='Validating'):
        ne_geom = shape(ne_country['geometry'])
        ne_point = ne_geom.representative_point()

        if not any(c.geometry[BEST_GEOJSON_QUALITY].contains(ne_point) for c in countries):
            raise ValueError(f'Country geometry not found: {ne_country["properties"]["NAME"]!r}')

natural_earth.py
Zaczero/osm-countries-geojson
from collections import defaultdict
from itertools import chain, cycle, islice, pairwise
from typing import Sequence

import networkx as nx
import numpy as np
from shapely.geometry import MultiPolygon, Polygon
from shapely.ops import unary_union
from tqdm import tqdm

from config import BEST_GEOJSON_QUALITY, GEOJSON_QUALITIES
from models.osm_country import OSMCountry
from overpass import query_overpass

_QUERY = (
    'rel[boundary=administrative][admin_level=2]["ISO3166-1"][name];'
    'out geom qt;'
)


def _connect_segments(segments: Sequence[tuple[tuple]]) -> Sequence[Sequence[tuple]]:
    # count occurrences of each node
    node_count = defaultdict(int)
    for node in chain.from_iterable(segments):
        node_count[node] += 1

    # validate that all segments are closed (i.e., start and end at intersections)
    for s in segments:
        node_id_start = s[0]
        node_id_end = s[-1]
        if node_count[node_id_start] < 2 or node_count[node_id_end] < 2:
            raise ValueError(f'Segments must be closed (node/{node_id_start}, node/{node_id_end})')

    # node = intersection, node_count > 1
    # edge = segment between intersections
    G = nx.DiGraph()

    # build the graph
    for segment in segments:
        subsegment_start = None
        subsegment = []
        for node in segment:
            # intersection node
            if node_count[node] > 1:
                if subsegment_start:
                    if len(subsegment) == 0:
                        G.add_edge(subsegment_start, node)
                        G.add_edge(node, subsegment_start)
                    elif len(subsegment) == 1:
                        first = subsegment[0]
                        G.add_edge(subsegment_start, first)
                        G.add_edge(first, node)
                        G.add_edge(node, first)
                        G.add_edge(first, subsegment_start)
                    else:
                        first = subsegment[0]
                        last = subsegment[-1]
                        G.add_edge(subsegment_start, first)
                        G.add_edge(first, last, subsegment=subsegment)
                        G.add_edge(last, node)
                        G.add_edge(node, last)
                        G.add_edge(last, first, subsegment=subsegment[::-1])
                        G.add_edge(first, subsegment_start)
                subsegment = []
                subsegment_start = node
            # normal node
            elif subsegment_start:
                subsegment.append(node)

    # set to store connected segments (closed loops)
    connected = set()

    # function to normalize a closed loop
    def connected_normalize(segment: list) -> tuple:
        min_idx = np.argmin(segment, axis=0)[0]

        # normalize starting point
        aligned = tuple(chain(
            islice(segment, min_idx, len(segment) - 1),
            islice(segment, min_idx + 1)))

        # normalize orientation
        if segment[-1] < segment[1]:
            aligned = aligned[::-1]

        return aligned

    for c in nx.simple_cycles(G):
        c = tuple(islice(cycle(c), len(c) + 1))  # close the cycle

        merged_unordered: list[list] = []

        for u, v in pairwise(c):
            if subsegment := G[u][v].get('subsegment'):
                merged_unordered.append(subsegment)
            else:
                merged_unordered.append([u, v])

        if len(merged_unordered) < 2:
            # this realistically will mean broken data: a single node, a small loop, etc.
            raise Exception(f'Single-segment cycle ({c!r})')

        first = merged_unordered[0]
        second = merged_unordered[1]

        # proper orientation of the first segment
        if first[0] in (second[0], second[-1]):
            merged = first[::-1]
        else:
            merged = first

        for segment in merged_unordered[1:]:
            if merged[-1] == segment[0]:
                merged.extend(islice(segment, 1, None))
            elif merged[-1] == segment[-1]:
                merged.extend(islice(reversed(segment), 1, None))
            else:
                print('⚠️ Invalid cycle')
                break
        else:
            if len(merged) >= 4 and merged[1] != merged[-2]:
                connected.add(connected_normalize(merged))

    return tuple(connected)


async def get_osm_countries() -> tuple[Sequence[OSMCountry], float]:
    print('Querying Overpass API')
    countries, data_timestamp = await query_overpass(_QUERY, timeout=300, must_return=True)
    result = []

    for country in tqdm(countries, desc='Processing geometry'):
        outer_segments = []
        inner_segments = []

        for member in country.get('members', []):
            if member['type'] != 'way':
                continue

            if member['role'] == 'outer':
                outer_segments.append(tuple((g['lon'], g['lat']) for g in member['geometry']))
            elif member['role'] == 'inner':
                inner_segments.append(tuple((g['lon'], g['lat']) for g in member['geometry']))

        try:
            outer_polys = tuple(Polygon(s) for s in _connect_segments(outer_segments))
            inner_polys = tuple(Polygon(s) for s in _connect_segments(inner_segments))
            geometry: dict[float, Polygon | MultiPolygon] = {}

            for q in GEOJSON_QUALITIES:
                outer_simple = (p.simplify(q) for p in outer_polys)
                outer_simple = tuple(p for p in outer_simple if p.is_valid)
                inner_simple = (p.simplify(q) for p in inner_polys)
                inner_simple = tuple(p for p in inner_simple if p.is_valid)

                if not outer_simple:
                    raise Exception('No outer polygons')

                outer_union = unary_union(outer_simple)
                inner_union = unary_union(inner_simple)
                geometry[q] = outer_union.difference(inner_union)
        except Exception as e:
            country_name = country['tags'].get('name', '??')
            raise Exception(f'Error processing {country_name}') from e

        representative_point = geometry[BEST_GEOJSON_QUALITY].representative_point()

        result.append(OSMCountry(
            tags=country['tags'],
            geometry=geometry,
            representative_point=representative_point,
        ))

    return tuple(result), data_timestamp

osm_countries_gen.py
Zaczero/osm-countries-geojson
from datetime import UTC, datetime, timedelta
from typing import Sequence

from config import OVERPASS_API_URL
from utils import get_http_client, retry_exponential


@retry_exponential(timedelta(minutes=30))
async def query_overpass(query: str, *, timeout: int, must_return: bool = True) -> tuple[Sequence[dict], float]:
    join = '' if query.startswith('[') else ';'
    query = f'[out:json][timeout:{timeout}]{join}{query}'

    async with get_http_client() as http:
        r = await http.post(
            OVERPASS_API_URL,
            data={'data': query},
            timeout=timeout * 2)
        r.raise_for_status()

    data = r.json()
    data_timestamp = datetime \
        .strptime(data['osm3s']['timestamp_osm_base'], '%Y-%m-%dT%H:%M:%SZ') \
        .replace(tzinfo=UTC) \
        .timestamp()

    if must_return and not data['elements']:
        raise Exception('No elements returned')

    return data['elements'], data_timestamp

overpass.py
Zaczero/osm-countries-geojson
import functools
import time
import traceback
from contextlib import contextmanager
from datetime import timedelta
from math import inf
from typing import Generator

import anyio
import httpx

from config import USER_AGENT


def retry_exponential(timeout: timedelta | None, *, start: float = 1):
    if timeout is None:
        timeout_seconds = inf
    else:
        timeout_seconds = timeout.total_seconds()

    def decorator(func):
        @functools.wraps(func)
        async def wrapper(*args, **kwargs):
            ts = time.perf_counter()
            sleep = start

            while True:
                try:
                    return await func(*args, **kwargs)
                except Exception:
                    print(f'[⛔] {func.__name__} failed')
                    traceback.print_exc()
                    if (time.perf_counter() + sleep) - ts > timeout_seconds:
                        raise
                    await anyio.sleep(sleep)
                    sleep = min(sleep * 2, 1800)  # max 30 minutes

        return wrapper
    return decorator


def get_http_client(base_url: str = '') -> httpx.AsyncClient:
    return httpx.AsyncClient(
        base_url=base_url,
        headers={'User-Agent': USER_AGENT},
        timeout=httpx.Timeout(60, connect=15),
    )

utils.py
Zaczero/osm-countries-geojson