perfect-postcode/pipeline/download/map_assets.py

import argparse
import sys
import urllib.request
from concurrent.futures import ThreadPoolExecutor, as_completed
from pathlib import Path

from pipeline.transform.transform_poi import NAPTAN_EMOJIS, _CATEGORIES

GLYPHS_BASE = "https://protomaps.github.io/basemaps-assets/fonts"
SPRITES_BASE = "https://protomaps.github.io/basemaps-assets/sprites/v4"
TWEMOJI_BASE = "https://cdn.jsdelivr.net/gh/twitter/twemoji@14.0.2/assets/72x72"
POI_ICON_BASE = "https://geolytix.github.io/MapIcons"

# Font stacks used by @protomaps/basemaps with lang='en'
FONT_STACKS = ["Noto Sans Regular", "Noto Sans Italic", "Noto Sans Medium"]

# Fallback emoji not in any category
_FALLBACK_EMOJIS = ["📍"]

POI_ICON_PATHS = [
    "asda/asda_express_24px.svg",
    "asda/asda_green_basket_24px.svg",
    "asda/asda_green_trolley_24px.svg",
    "asda/asda_living_24px.svg",
    "asda/asda_pfs_24px.svg",
    "asda/asda_primary.svg",
    "asda/asda_superstore_green_trolley_24px.svg",
    "brands/aldi_24px.svg",
    "brands/amazon_fresh_alt_24px.svg",
    "brands/booths_24px.svg",
    "brands/budgens_24px.svg",
    "brands/centra_24px.svg",
    "brands/cook.svg",
    "brands/coop_24px.svg",
    "brands/costco_24px.svg",
    "brands/dunnes_stores_24px.svg",
    "brands/farmfoods_updated_24px.svg",
    "brands/heron_24px.svg",
    "brands/iceland_24px.svg",
    "brands/iceland_food_warehouse_24px.svg",
    "brands/lidl_24px.svg",
    "brands/little_waitrose_24px.svg",
    "brands/makro_24px.svg",
    "brands/mns_24px.svg",
    "brands/mns_food_24px.svg",
    "brands/mns_high_street_24px.svg",
    "brands/mns_hospital_24px.svg",
    "brands/mns_moto_24px.svg",
    "brands/mns_outlet_24px.svg",
    "brands/morrisons_24px.svg",
    "brands/morrisons_daily_24px.svg",
    "brands/sainsburys_24px.svg",
    "brands/sainsburys_local_24px.svg",
    "brands/spar_24px.svg",
    "brands/tesco_24px.svg",
    "brands/tesco_express_24px.svg",
    "brands/tesco_extra_24px.svg",
    "brands/waitrose_24px.svg",
    "brands/wholefoods_24px.svg",
    "logos/planet_organic_24px.svg",
    "public_transport/london_tube.svg",
]


def collect_twemoji_codes() -> list[str]:
    """Derive twemoji hex codes from transform_poi categories.

    Matches the frontend's emojiToTwemojiUrl() which does
    emoji.codePointAt(0).toString(16).
    """
    emojis: set[str] = set()

    for _group, _name, emoji, _osm_keys in _CATEGORIES:
        emojis.add(emoji)

    for emoji in NAPTAN_EMOJIS.values():
        emojis.add(emoji)

    for emoji in _FALLBACK_EMOJIS:
        emojis.add(emoji)

    # First codepoint hex, matching frontend logic
    return sorted({f"{ord(e[0]):x}" for e in emojis})


def download_file(url: str, dest: Path) -> tuple[bool, str]:
    """Download a single file. Returns (success, url)."""
    dest.parent.mkdir(parents=True, exist_ok=True)
    try:
        urllib.request.urlretrieve(url, dest)
        return True, url
    except urllib.error.HTTPError as e:
        print(f"  {e.code} {url}", file=sys.stderr)
        return False, url
    except Exception as e:
        print(f"  ERROR {url}: {e}", file=sys.stderr)
        return False, url


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument(
        "--output",
        type=Path,
        required=True,
        help="Output directory",
    )
    args = parser.parse_args()
    out: Path = args.output

    twemoji_codes = collect_twemoji_codes()

    # Build download list
    tasks: list[tuple[str, Path]] = []

    # Font glyphs: 256 range files per font stack
    for font in FONT_STACKS:
        font_encoded = font.replace(" ", "%20")
        font_dir = out / "fonts" / font
        for start in range(0, 65536, 256):
            end = start + 255
            name = f"{start}-{end}.pbf"
            url = f"{GLYPHS_BASE}/{font_encoded}/{name}"
            tasks.append((url, font_dir / name))

    # Sprite sheets (light/dark, 1x and 2x)
    sprites_dir = out / "sprites"
    for theme in ("light", "dark"):
        for suffix in ("json", "png"):
            url = f"{SPRITES_BASE}/{theme}.{suffix}"
            tasks.append((url, sprites_dir / f"{theme}.{suffix}"))
            url_2x = f"{SPRITES_BASE}/{theme}@2x.{suffix}"
            tasks.append((url_2x, sprites_dir / f"{theme}@2x.{suffix}"))

    # Twemoji PNGs
    twemoji_dir = out / "twemoji"
    for code in twemoji_codes:
        url = f"{TWEMOJI_BASE}/{code}.png"
        tasks.append((url, twemoji_dir / f"{code}.png"))

    # Branded POI icons are served from this local bundle at runtime.
    poi_icons_dir = out / "poi-icons"
    for icon_path in POI_ICON_PATHS:
        url = f"{POI_ICON_BASE}/{icon_path}"
        tasks.append((url, poi_icons_dir / icon_path))

    # Skip already-downloaded files
    remaining = [(url, dest) for url, dest in tasks]

    print(f"Downloading {len(remaining)} assets")

    ok = 0
    fail = 0
    with ThreadPoolExecutor(max_workers=20) as pool:
        futures = {
            pool.submit(download_file, url, dest): url for url, dest in remaining
        }
        for future in as_completed(futures):
            success, url = future.result()
            if success:
                ok += 1
            else:
                fail += 1

    print(f"Done: {ok} downloaded, {fail} failed")


if __name__ == "__main__":
    main()