Core API Reference

yoromaps

Top-level module with convenience imports.

yoromaps

Yoro Maps — Offline maps, routing, and POI, worldwide.

Companion to the yoro geocoding package.

Usage::

import yoromaps

# Build a .yoromaps file for Mali
yoromaps.build("ML", "mali.yoromaps")

# Route between Yoro codes
conn = yoromaps.open_db("mali.yoromaps")
result = yoromaps.route(conn, start_lat=12.6, start_lon=-8.0, end_lat=14.5, end_lon=-4.0)
print(f"{result.distance_km} km, {result.duration_min} min")

# Route from Yoro codes
legs = yoromaps.route_from_codes(conn, ["ML-ABC", "ML-XYZ"])

# As something other than a car. One of yoromaps.MODES; the mode changes
# the minutes *and* the roads the line runs along, since it is a
# permission as well as a speed.
legs = yoromaps.route_from_codes(conn, ["ML-ABC", "ML-XYZ"], mode="motorcycle")

RouteResult dataclass

Result of a routing query.

Source code in src/yoromaps/routing.py

@dataclass
class RouteResult:
    """Result of a routing query."""

    distance_km: float
    duration_min: float
    nodes: list[int]
    geometry: GeoJson
    steps: list[dict[str, object]]
    found: bool = True
    #: How far the points asked about are from the road the journey really
    #: starts and ends on. Not diagnostics: past `ACCESS_LEG_M` this is a walk
    #: somebody has to make, it is drawn at the ends of the line, and an
    #: answer that hid it would be a route starting somewhere nobody is.
    from_snap_m: float = 0.0
    to_snap_m: float = 0.0
    #: What that walk leads to, and what it comes back from — the name of the
    #: first and last road of the path. Empty when the road has no name, which
    #: is most of them outside the cities.
    from_road: str = ""
    to_road: str = ""
    #: How this was travelled. Stated because it changed the answer — both
    #: the minutes and the roads the line runs along.
    mode: str = "car"

build(area_code, output, pbf_path=None, include_tiles=False, zoom_min=6, zoom_max=12, include_pois=True, progress=None)

Build a .yoromaps file for an area.

An area is a country that Geofabrik serves whole, or one region of a country it does not — France's extract is 4 830 MB and Alsace's is 124 MB, and only one of those is something to ask somebody to download.

The bundle records the country, not the area. Codes inside Alsace are French codes, addressed against France's box: an address does not change because somebody downloaded less of the map around it.

Parameters:

Name	Type	Description	Default
area_code	str	An area code — "ML" for a whole country, "FR:alsace" for one region of one.	required
output	str | Path	Output .yoromaps file path.	required
pbf_path	str | Path | None	Path to an existing PBF file. If None, downloads from Geofabrik.	None
include_tiles	bool	Also download map tiles (slow, ~200+ MB).	False
zoom_min	int	Min tile zoom level (if include_tiles).	6
zoom_max	int	Max tile zoom level (if include_tiles).	12
include_pois	bool	Also extract the shops, schools and hospitals OSM knows about. On by default — a bundle without them routes to places the user has no way to find.	True
progress	ProgressCallback | None	Optional callable(message, current, total).	None

Returns:

Type	Description
Path	Path to the created .yoromaps file.

Source code in src/yoromaps/download.py

def build(
    area_code: str,
    output: str | Path,
    pbf_path: str | Path | None = None,
    include_tiles: bool = False,
    zoom_min: int = 6,
    zoom_max: int = 12,
    include_pois: bool = True,
    progress: ProgressCallback | None = None,
) -> Path:
    """Build a .yoromaps file for an area.

    An area is a country that Geofabrik serves whole, or one region of a
    country it does not — France's extract is 4 830 MB and Alsace's is 124 MB,
    and only one of those is something to ask somebody to download.

    The bundle records the *country*, not the area. Codes inside Alsace are
    French codes, addressed against France's box: an address does not change
    because somebody downloaded less of the map around it.

    Args:
        area_code: An area code — "ML" for a whole country, "FR:alsace" for
            one region of one.
        output: Output .yoromaps file path.
        pbf_path: Path to an existing PBF file. If None, downloads from Geofabrik.
        include_tiles: Also download map tiles (slow, ~200+ MB).
        zoom_min: Min tile zoom level (if include_tiles).
        zoom_max: Max tile zoom level (if include_tiles).
        include_pois: Also extract the shops, schools and hospitals OSM knows
            about. On by default — a bundle without them routes to places the
            user has no way to find.
        progress: Optional callable(message, current, total).

    Returns:
        Path to the created .yoromaps file.
    """
    target = _area_or_refuse(area_code)
    output = Path(output)

    if pbf_path is None:
        tmp_dir = tempfile.mkdtemp(prefix="yoromaps_")
        downloaded = download_pbf(target.code, tmp_dir, progress=progress)
        if downloaded is None:
            # download_pbf only returns None when asked to skip an unchanged
            # file, which a fresh build never does. Refuse rather than carry a
            # None into the extraction and fail there with no explanation.
            raise RuntimeError(f"No PBF was downloaded for {target.code}")
        source_pbf = downloaded
    else:
        source_pbf = Path(pbf_path)

    conn = open_db(output, create=True)
    # `country` is what addresses the POIs, so it stays the country even when
    # the bundle covers one region of it. `area` says which piece this is.
    set_metadata(conn, "country", target.country)
    set_metadata(conn, "country_name", COUNTRIES[target.country].name)
    set_metadata(conn, "area", target.code)
    set_metadata(conn, "area_name", target.name)
    set_metadata(conn, "bbox", ",".join(str(v) for v in target.bbox))

    _build_graph(conn, source_pbf, progress)
    if include_pois:
        _build_pois(conn, source_pbf, progress)
    if include_tiles:
        _build_tiles(conn, target.bbox, zoom_min, zoom_max, progress)

    conn.close()
    return output

update(db_path, pbf_cache_dir=None, progress=None)

Update an existing .yoromaps file with the latest OSM data.

Downloads the PBF only if Geofabrik has a newer version.

Rebuilds the road graph and re-extracts the OSM POIs from the same file. POIs from any other source — the ones contributed locally — are left untouched, since only rows tagged as OSM-sourced are replaced.

Parameters:

Name	Type	Description	Default
db_path	str | Path	Path to the existing .yoromaps file.	required
pbf_cache_dir	str | Path | None	Directory to cache PBF files. Defaults to same dir as db.	None
progress	ProgressCallback | None	Optional progress callback.	None

Returns:

Type	Description
UpdateStats	Dict with updated (bool), nodes, edges, previous_date, new_date.

Source code in src/yoromaps/download.py

def update(
    db_path: str | Path,
    pbf_cache_dir: str | Path | None = None,
    progress: ProgressCallback | None = None,
) -> UpdateStats:
    """Update an existing .yoromaps file with the latest OSM data.

    Downloads the PBF only if Geofabrik has a newer version.

    Rebuilds the road graph and re-extracts the OSM POIs from the same file.
    POIs from any other source — the ones contributed locally — are left
    untouched, since only rows tagged as OSM-sourced are replaced.

    Args:
        db_path: Path to the existing .yoromaps file.
        pbf_cache_dir: Directory to cache PBF files. Defaults to same dir as db.
        progress: Optional progress callback.

    Returns:
        Dict with ``updated`` (bool), ``nodes``, ``edges``, ``previous_date``, ``new_date``.
    """
    db_path = Path(db_path)
    conn = open_db(db_path)

    # The area first: a bundle covering one region must refresh from that
    # region's extract, not from the whole country it belongs to — which is
    # the difference between 124 MB and 4 830 MB for a bundle of Alsace.
    # Bundles built before areas existed carry only the country, and for them
    # the country is the area.
    area_code = get_metadata(conn, "area") or get_metadata(conn, "country")
    if not area_code:
        conn.close()
        raise ValueError(f"No area metadata in {db_path}. Is this a valid .yoromaps file?")

    last_update = get_metadata(conn, "osm_update_date")
    if pbf_cache_dir is None:
        pbf_cache_dir = db_path.parent

    if progress:
        progress(f"Checking for updates ({area_code})...", 0, 3)

    # Download PBF (only if newer)
    pbf_path = download_pbf(
        area_code,
        pbf_cache_dir,
        progress=progress,
        if_newer_than=last_update,
    )

    if pbf_path is None:
        unchanged = _unchanged_stats(conn, last_update)
        conn.close()
        return unchanged

    stats = _replace_graph(conn, pbf_path, progress)
    poi_stats = extract_pois(str(pbf_path), conn, progress=progress)

    now = format_datetime(datetime.now(timezone.utc), usegmt=True)
    set_metadata(conn, "graph_nodes", str(stats["nodes"]))
    set_metadata(conn, "graph_edges", str(stats["edges"]))
    set_metadata(conn, "poi_count", str(poi_stats["pois"]))
    set_metadata(conn, "osm_update_date", now)

    if progress:
        progress("Done", 3, 3)

    conn.close()

    return {
        "updated": True,
        "nodes": stats["nodes"],
        "edges": stats["edges"],
        "pois": poi_stats["pois"],
        "previous_date": last_update,
        "new_date": now,
    }

open_db(path, create=False)

Open a .yoromaps database. Creates schema if create is True.

Existing databases are upgraded transparently: the schema is idempotent (CREATE ... IF NOT EXISTS), so files built by older versions gain new tables and indexes on first open, and columns added to a table that already existed are applied by _add_late_columns.

Source code in src/yoromaps/db.py

def open_db(path: str | Path, create: bool = False) -> sqlite3.Connection:
    """Open a .yoromaps database. Creates schema if *create* is True.

    Existing databases are upgraded transparently: the schema is idempotent
    (``CREATE ... IF NOT EXISTS``), so files built by older versions gain new
    tables and indexes on first open, and columns added to a table that
    already existed are applied by `_add_late_columns`.
    """
    path = Path(path)
    exists = path.exists()

    if not exists and not create:
        raise FileNotFoundError(f"Database not found: {path}")

    conn = sqlite3.connect(str(path))
    conn.row_factory = sqlite3.Row
    conn.execute("PRAGMA journal_mode=WAL")
    conn.execute("PRAGMA synchronous=NORMAL")
    conn.execute("PRAGMA foreign_keys=ON")

    conn.executescript(SCHEMA_SQL)
    _add_late_columns(conn)
    stored = get_metadata(conn, "schema_version")
    if stored is None or int(stored) < SCHEMA_VERSION:
        conn.execute(
            "INSERT OR REPLACE INTO metadata VALUES ('schema_version', ?)",
            (str(SCHEMA_VERSION),),
        )
    conn.commit()

    return conn

db_config(path)

Return a Django DATABASES dict entry for a .yoromaps file.

Source code in src/yoromaps/db.py

def db_config(path: str) -> dict[str, str]:
    """Return a Django DATABASES dict entry for a .yoromaps file."""
    return {
        "ENGINE": "django.db.backends.sqlite3",
        "NAME": str(path),
    }

route(conn, start_lat, start_lon, end_lat, end_lon, mode='car')

Find the quickest route between two GPS coordinates.

Loads the graph into memory on each call. For multiple routes, use Graph.from_db() directly.

Source code in src/yoromaps/routing.py

def route(
    conn: sqlite3.Connection,
    start_lat: float,
    start_lon: float,
    end_lat: float,
    end_lon: float,
    mode: str = "car",
) -> RouteResult:
    """Find the quickest route between two GPS coordinates.

    Loads the graph into memory on each call. For multiple routes,
    use ``Graph.from_db()`` directly.
    """
    graph = Graph.from_db(conn)
    return graph.route(start_lat, start_lon, end_lat, end_lon, mode)

route_from_codes(conn, codes, graph=None, mode='car')

Route through multiple Yoro codes in order.

Returns a list of RouteResult, one per leg. Pass a preloaded graph (e.g. from :func:get_graph) to avoid reloading it from the database — the same graph answers for every mode, so one load serves them all.

Source code in src/yoromaps/routing.py

def route_from_codes(
    conn: sqlite3.Connection,
    codes: list[str],
    graph: Graph | None = None,
    mode: str = "car",
) -> list[RouteResult]:
    """Route through multiple Yoro codes in order.

    Returns a list of RouteResult, one per leg. Pass a preloaded *graph*
    (e.g. from :func:`get_graph`) to avoid reloading it from the database —
    the same graph answers for every mode, so one load serves them all.
    """
    import yoro

    if graph is None:
        graph = Graph.from_db(conn)
    decoded = [yoro.decode(c) for c in codes]
    points = [(d["lat"], d["lon"]) for d in decoded]
    return [
        graph.route(points[i][0], points[i][1], points[i + 1][0], points[i + 1][1], mode)
        for i in range(len(points) - 1)
    ]

get_graph(db_path)

Load the road graph for a .yoromaps file, cached per process.

The graph is loaded once and reused across calls (and across HTTP requests in Django). The cache entry is invalidated when the file's mtime changes, e.g. after yoromaps update.

Source code in src/yoromaps/routing.py

def get_graph(db_path: str | Path) -> Graph:
    """Load the road graph for a .yoromaps file, cached per process.

    The graph is loaded once and reused across calls (and across HTTP
    requests in Django). The cache entry is invalidated when the file's
    mtime changes, e.g. after ``yoromaps update``.
    """
    path = str(Path(db_path).resolve())
    mtime = os.path.getmtime(path)

    cached = _graph_cache.get(path)
    if cached and cached[0] == mtime:
        return cached[1]

    with _graph_lock:
        cached = _graph_cache.get(path)
        if cached and cached[0] == mtime:
            return cached[1]

        conn = sqlite3.connect(path)
        try:
            graph = Graph.from_db(conn)
        finally:
            conn.close()
        _graph_cache[path] = (mtime, graph)
        return graph

add_poi(conn, lat, lon, name, category='other', source='local', osm_id=None, details=None)

Insert a POI; its Yoro code is computed automatically.

details is one argument rather than four because a caller usually knows none of them or all of them, and four empty strings at every call site would say nothing four times.

Source code in src/yoromaps/poi.py

def add_poi(
    conn: sqlite3.Connection,
    lat: float,
    lon: float,
    name: str,
    category: str = "other",
    source: str = "local",
    osm_id: int | None = None,
    details: PoiDetails | None = None,
) -> Poi:
    """Insert a POI; its Yoro code is computed automatically.

    `details` is one argument rather than four because a caller usually knows
    none of them or all of them, and four empty strings at every call site
    would say nothing four times.
    """
    code = _encode_for_db(conn, lat, lon)
    said: PoiDetails = details or {}
    phone = said.get("phone", "")
    website = said.get("website", "")
    opening_hours = said.get("opening_hours", "")
    address = said.get("address", "")
    cur = conn.execute(
        "INSERT INTO pois (lat, lon, yoro_code, name, category, source, osm_id, "
        "phone, website, opening_hours, address) "
        "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
        (lat, lon, code, name, category, source, osm_id, phone, website, opening_hours, address),
    )
    conn.commit()
    if cur.lastrowid is None:
        raise RuntimeError("SQLite did not report a row id for the inserted POI")
    return {
        "id": cur.lastrowid,
        "lat": lat,
        "lon": lon,
        "yoro_code": code,
        "name": name,
        "category": category,
        "source": source,
        "osm_id": osm_id,
        "distance_m": None,
        "phone": phone,
        "website": website,
        "opening_hours": opening_hours,
        "address": address,
    }

pois_near(conn, lat, lon, radius_m=1000, category=None, limit=100)

POIs within radius_m metres, sorted by distance.

Source code in src/yoromaps/poi.py

def pois_near(
    conn: sqlite3.Connection,
    lat: float,
    lon: float,
    radius_m: float = 1000,
    category: str | None = None,
    limit: int = 100,
) -> list[Poi]:
    """POIs within *radius_m* metres, sorted by distance."""
    from yoromaps.extract import haversine

    dlat = radius_m / 111_000
    dlon = radius_m / (111_000 * max(0.1, math.cos(math.radians(lat))))

    sql = "SELECT * FROM pois WHERE lat BETWEEN ? AND ? AND lon BETWEEN ? AND ?"
    params: list[object] = [lat - dlat, lat + dlat, lon - dlon, lon + dlon]
    if category:
        sql += " AND category = ?"
        params.append(category)

    hits: list[tuple[float, sqlite3.Row]] = []
    for row in conn.execute(sql, params):
        d = haversine(lat, lon, row["lat"], row["lon"])
        if d <= radius_m:
            hits.append((d, row))
    hits.sort(key=lambda t: t[0])
    return [_poi_dict(row, distance_m=d) for d, row in hits[:limit]]

pois_in_cell(conn, code, category=None)

All POIs inside the cell of a Yoro code (any precision).

Raises ValueError for invalid codes.

Source code in src/yoromaps/poi.py

def pois_in_cell(
    conn: sqlite3.Connection,
    code: str,
    category: str | None = None,
) -> list[Poi]:
    """All POIs inside the cell of a Yoro code (any precision).

    Raises ValueError for invalid codes.
    """
    b = yoro.decode(code)["bounds"]
    sql = "SELECT * FROM pois WHERE lat >= ? AND lat < ? AND lon >= ? AND lon < ?"
    params: list[object] = [b["lat_min"], b["lat_max"], b["lon_min"], b["lon_max"]]
    if category:
        sql += " AND category = ?"
        params.append(category)
    return [_poi_dict(r) for r in conn.execute(sql, params)]

search_pois(conn, query=None, category=None, limit=50)

Search POIs by name substring and/or category.

Source code in src/yoromaps/poi.py

def search_pois(
    conn: sqlite3.Connection,
    query: str | None = None,
    category: str | None = None,
    limit: int = 50,
) -> list[Poi]:
    """Search POIs by name substring and/or category."""
    sql = "SELECT * FROM pois WHERE 1=1"
    params: list[object] = []
    if query:
        sql += " AND name LIKE ?"
        params.append(f"%{query}%")
    if category:
        sql += " AND category = ?"
        params.append(category)
    sql += " ORDER BY name LIMIT ?"
    params.append(limit)
    return [_poi_dict(r) for r in conn.execute(sql, params)]

delete_poi(conn, poi_id)

Delete a POI by id. Returns True if a row was removed.

Source code in src/yoromaps/poi.py

def delete_poi(conn: sqlite3.Connection, poi_id: int) -> bool:
    """Delete a POI by id. Returns True if a row was removed."""
    cur = conn.execute("DELETE FROM pois WHERE id = ?", (poi_id,))
    conn.commit()
    return cur.rowcount > 0

extract_pois(pbf_path, conn, progress=None, source=OSM_SOURCE, precision=DEFAULT_PRECISION)

Extract POIs from a PBF into the database, replacing previous ones.

Rows carrying source are deleted first, so a rebuild refreshes OSM data without touching POIs from anywhere else. The delete and every insert share one transaction: a failure mid-extraction leaves the previous set in place.

Rows are written in batches of POI_INSERT_BATCH rather than accumulated, so peak memory does not grow with the size of the country.

Nameless features are skipped — a pin with no label is noise on a map whose whole purpose is telling someone where to go.

Parameters:

Name	Type	Description	Default
pbf_path	str | Path	Path to the .osm.pbf file.	required
conn	Connection	Open connection to a .yoromaps database.	required
progress	ProgressCallback | None	Optional callable(message, current, total).	None
source	str	Value written to pois.source, and the rows replaced.	OSM_SOURCE
precision	int	Yoro precision of the stored codes. The default addresses each POI individually; lower it only if you want neighbours to share a code on purpose.	DEFAULT_PRECISION

Returns:

Type	Description
PoiExtractStats	Counts of what was written and walked past, plus unmapped — the
PoiExtractStats	most frequent tag values that matched nothing, as
PoiExtractStats	[(("amenity", "parking"), 1234), ...].

Raises:

Type	Description
ValueError	If source is empty — it selects the rows to replace, and an empty value would delete nothing while inserting unattributable rows.
ImportError	If the extract extra is not installed.

Source code in src/yoromaps/osm_poi.py

def extract_pois(
    pbf_path: str | Path,
    conn: sqlite3.Connection,
    progress: ProgressCallback | None = None,
    source: str = OSM_SOURCE,
    precision: int = DEFAULT_PRECISION,
) -> PoiExtractStats:
    """Extract POIs from a PBF into the database, replacing previous ones.

    Rows carrying *source* are deleted first, so a rebuild refreshes OSM data
    without touching POIs from anywhere else. The delete and every insert share
    one transaction: a failure mid-extraction leaves the previous set in place.

    Rows are written in batches of POI_INSERT_BATCH rather than accumulated,
    so peak memory does not grow with the size of the country.

    Nameless features are skipped — a pin with no label is noise on a map whose
    whole purpose is telling someone where to go.

    Args:
        pbf_path: Path to the `.osm.pbf` file.
        conn: Open connection to a `.yoromaps` database.
        progress: Optional callable(message, current, total).
        source: Value written to `pois.source`, and the rows replaced.
        precision: Yoro precision of the stored codes. The default addresses
            each POI individually; lower it only if you want neighbours to
            share a code on purpose.

    Returns:
        Counts of what was written and walked past, plus ``unmapped`` — the
        most frequent tag values that matched nothing, as
        ``[(("amenity", "parking"), 1234), ...]``.

    Raises:
        ValueError: If *source* is empty — it selects the rows to replace, and
            an empty value would delete nothing while inserting unattributable
            rows.
        ImportError: If the `extract` extra is not installed.
    """
    if not source:
        raise ValueError("source must name where these POIs come from")

    _check_osmium()

    country = get_metadata(conn, "country") or ""
    collector = _PoiCollector(conn, country, source, precision)

    if progress:
        progress("Scanning POIs...", 0, 2)

    try:
        conn.execute("DELETE FROM pois WHERE source = ?", (source,))
        # `locations=True` fills in way node coordinates, which is what lets a
        # hospital mapped as a building become a point.
        _poi_handler(collector).apply_file(str(pbf_path), locations=True)
        collector.flush()
    except BaseException:
        conn.rollback()
        raise
    conn.commit()

    if progress:
        progress(f"Stored {collector.written} POIs", 2, 2)

    return {
        "pois": collector.written,
        "nodes": collector.counts["nodes"],
        "ways": collector.counts["ways"],
        "skipped_unnamed": collector.counts["skipped_unnamed"],
        "unmapped": collector.unmapped.most_common(UNMAPPED_REPORTED),
    }

categorize(tags)

The category of an OSM feature, or None when it is not a POI.

categorize({"amenity": "pharmacy"}) 'pharmacy' categorize({"shop": "bakery"}) 'shop' categorize({"highway": "residential"}) is None True

Source code in src/yoromaps/osm_poi.py

def categorize(tags: dict[str, str]) -> str | None:
    """The category of an OSM feature, or None when it is not a POI.

    >>> categorize({"amenity": "pharmacy"})
    'pharmacy'
    >>> categorize({"shop": "bakery"})
    'shop'
    >>> categorize({"highway": "residential"}) is None
    True
    """
    for key, values in VALUE_CATEGORIES.items():
        category = values.get(tags.get(key, ""))
        if category:
            return category

    for key, category in KEY_CATEGORIES.items():
        value = tags.get(key)
        if value and value not in NEGATIVE_VALUES:
            return category

    return None

yoromaps.routing

A* routing engine on the .yoromaps road graph. The search minimises time, and the speed of every edge comes from profiles at query time — so one loaded graph answers for a car, a moto, a bicycle and a pedestrian.

Graph.routes offers several ways of getting there rather than one: the quickest, then the best remaining answers once the streets already used are made dearer, keeping only those that share less than MAX_OVERLAP of their length with one already offered. Fewer come back when there are genuinely fewer — a town with one bridge has one way out.

yoromaps.routing

A* routing engine on the .yoromaps road graph.

Loads the graph adjacency list into memory for fast routing. Typical memory usage: ~100-200 MB for a country like Mali or Togo.

One graph, every mode. The edge carries what kind of road it is, and the speed comes from profiles at query time — so the same loaded graph answers for a car, a moto, a bicycle and a pedestrian without being loaded four times, and correcting a speed does not mean republishing a bundle.

The search minimises time, not distance. It used to minimise distance and report the time of whatever it found, which is a different answer wearing the same label: the shortest way through a city is rarely the quickest, and a bicycle and a car sent down the same road for the same reason were only ever going to disagree about how long it took.

ALTERNATIVE_PENALTY = 3.0 module-attribute

MAX_OVERLAP = 0.7 module-attribute

MAX_ALTERNATIVE_ATTEMPTS = 6 module-attribute

RouteResult dataclass

Result of a routing query.

Source code in src/yoromaps/routing.py

@dataclass
class RouteResult:
    """Result of a routing query."""

    distance_km: float
    duration_min: float
    nodes: list[int]
    geometry: GeoJson
    steps: list[dict[str, object]]
    found: bool = True
    #: How far the points asked about are from the road the journey really
    #: starts and ends on. Not diagnostics: past `ACCESS_LEG_M` this is a walk
    #: somebody has to make, it is drawn at the ends of the line, and an
    #: answer that hid it would be a route starting somewhere nobody is.
    from_snap_m: float = 0.0
    to_snap_m: float = 0.0
    #: What that walk leads to, and what it comes back from — the name of the
    #: first and last road of the path. Empty when the road has no name, which
    #: is most of them outside the cities.
    from_road: str = ""
    to_road: str = ""
    #: How this was travelled. Stated because it changed the answer — both
    #: the minutes and the roads the line runs along.
    mode: str = "car"

Graph

In-memory road graph loaded from a .yoromaps database.

Load once, route many times::

graph = Graph.from_db(conn)
r1 = graph.route(6.13, 1.22, 9.55, 1.18)
r2 = graph.route(6.13, 1.22, 8.98, 1.13)

Source code in src/yoromaps/routing.py

class Graph:
    """In-memory road graph loaded from a .yoromaps database.

    Load once, route many times::

        graph = Graph.from_db(conn)
        r1 = graph.route(6.13, 1.22, 9.55, 1.18)
        r2 = graph.route(6.13, 1.22, 8.98, 1.13)
    """

    def __init__(self) -> None:
        self.coords: dict[int, tuple[float, float]] = {}  # node_id → (lat, lon)
        # node_id → [(to, dist_m, kind, name)]
        #
        # The kind, not a duration: what an edge *is* does not change, what
        # it costs depends on who is travelling it. Storing the fact and
        # computing the cost is what lets one graph serve every mode.
        self.adj: dict[int, list[tuple[int, float, EdgeKind, str]]] = {}
        self._grid: dict[tuple[int, int], list[int]] | None = None
        self._grid_extent: tuple[int, int, int, int] | None = None
        self._edge_count: int | None = None

    @classmethod
    def from_db(cls, conn: sqlite3.Connection) -> Graph:
        """Load the entire graph into memory from a .yoromaps database."""
        g = cls()

        for row in conn.execute("SELECT id, lat, lon FROM nodes"):
            g.coords[row[0]] = (row[1], row[2])

        kinds: dict[tuple[str, str, str], EdgeKind] = {}
        # `edge_tag_columns` returns one of two literals, never caller input.
        for row in conn.execute(
            "SELECT from_node, to_node, distance_m, road_type, name, "
            f"{edge_tag_columns(conn)} FROM edges"
        ):
            from_n, to_n, dist, road_type, name, surface, closed_to = row
            if from_n not in g.adj:
                g.adj[from_n] = []
            # Shared: a country has millions of edges and a few hundred kinds
            # between them. Without this the same short strings are held a
            # million times over, for tens of megabytes of nothing.
            key = (road_type or "road", surface or "", closed_to or "")
            kind = kinds.get(key)
            if kind is None:
                closed = frozenset(m for m in key[2].split(",") if m)
                kind = kinds[key] = (sys.intern(key[0]), sys.intern(key[1]), closed)
            g.adj[from_n].append((to_n, dist, kind, name or ""))

        return g

    def _nearest(self, lat: float, lon: float) -> int | None:
        """Nearest node, unbounded (always returns one if graph non-empty)."""
        nid, _ = self.nearest_with_distance(lat, lon)
        return nid

    def _build_grid(self) -> None:
        grid: dict[tuple[int, int], list[int]] = {}
        for nid, (nlat, nlon) in self.coords.items():
            key = (math.floor(nlat / GRID_CELL_DEG), math.floor(nlon / GRID_CELL_DEG))
            grid.setdefault(key, []).append(nid)
        self._grid = grid
        keys = list(grid)
        self._grid_extent = (
            min(k[0] for k in keys),
            max(k[0] for k in keys),
            min(k[1] for k in keys),
            max(k[1] for k in keys),
        )

    def _ring_cells(self, ci: int, cj: int, ring: int) -> list[tuple[int, int]]:
        """Grid cells on the perimeter of a square ring, clipped to the extent.

        Only the perimeter, not the filled square: filtering a full square made
        far-away queries cubic in the ring index instead of linear.
        """
        assert self._grid_extent is not None
        i_min, i_max, j_min, j_max = self._grid_extent

        cells: list[tuple[int, int]] = []
        lo_j = max(cj - ring, j_min)
        hi_j = min(cj + ring, j_max)

        if i_min <= ci - ring <= i_max:
            cells.extend((ci - ring, j) for j in range(lo_j, hi_j + 1))
        if ring and i_min <= ci + ring <= i_max:
            cells.extend((ci + ring, j) for j in range(lo_j, hi_j + 1))
        if not ring:
            return cells

        lo_i = max(ci - ring + 1, i_min)
        hi_i = min(ci + ring - 1, i_max)
        if j_min <= cj - ring <= j_max:
            cells.extend((i, cj - ring) for i in range(lo_i, hi_i + 1))
        if j_min <= cj + ring <= j_max:
            cells.extend((i, cj + ring) for i in range(lo_i, hi_i + 1))
        return cells

    def _closest_in_cells(
        self,
        cells: list[tuple[int, int]],
        lat: float,
        lon: float,
        best_id: int | None,
        best_sq: float,
        accept: Callable[[int], bool] | None = None,
    ) -> tuple[int | None, float]:
        """Best node among *cells*, in squared degree distance.

        `accept` skips the nodes that are no use to the caller. Testing it
        here rather than filtering afterwards is what keeps the ring search
        correct: the nearest *acceptable* node can be several rings out, and a
        search that stopped at the nearest node of any kind would never see
        it.
        """
        assert self._grid is not None
        for key in cells:
            for nid in self._grid.get(key, ()):
                if accept is not None and not accept(nid):
                    continue
                nlat, nlon = self.coords[nid]
                d = (nlat - lat) ** 2 + (nlon - lon) ** 2
                if d < best_sq:
                    best_sq, best_id = d, nid
        return best_id, best_sq

    def _ring_is_outside(self, ci: int, cj: int, ring: int) -> bool:
        """True when the ring's square lies entirely off the populated grid."""
        assert self._grid_extent is not None
        i_min, i_max, j_min, j_max = self._grid_extent
        return ci + ring < i_min or ci - ring > i_max or cj + ring < j_min or cj - ring > j_max

    def nearest_with_distance(
        self, lat: float, lon: float, accept: Callable[[int], bool] | None = None
    ) -> tuple[int | None, float]:
        """Nearest node + haversine distance in metres.

        Uses a uniform grid index (built lazily on first call) and searches
        outward ring by ring. The distance metric matches the historical
        behavior: squared degree distance for the argmin, haversine for the
        returned distance.

        `accept` narrows what counts as a node. Without it this answers "what
        is the nearest piece of road", which is the wrong question for a
        router: see `_snap`.
        """
        if not self.coords:
            return None, float("inf")
        if self._grid is None:
            self._build_grid()

        ci = math.floor(lat / GRID_CELL_DEG)
        cj = math.floor(lon / GRID_CELL_DEG)

        assert self._grid_extent is not None
        i_min, i_max, j_min, j_max = self._grid_extent
        # Upper bound on rings: enough to reach the farthest populated cell.
        max_ring = max(abs(ci - i_min), abs(ci - i_max), abs(cj - j_min), abs(cj - j_max)) + 1

        best_id: int | None = None
        best_sq = float("inf")

        for ring in range(max_ring + 1):
            # Once a candidate is found, a node in ring r is at least
            # (r-1)*cell degrees away — stop when that exceeds the best.
            if best_id is not None and ((ring - 1) * GRID_CELL_DEG) ** 2 > best_sq:
                break
            if self._ring_is_outside(ci, cj, ring):
                continue
            cells = self._ring_cells(ci, cj, ring)
            best_id, best_sq = self._closest_in_cells(cells, lat, lon, best_id, best_sq, accept)

        if best_id is None:
            return None, float("inf")
        nlat, nlon = self.coords[best_id]
        return best_id, _haversine(lat, lon, nlat, nlon)

    def _travels(self, node_id: int, mode: str) -> bool:
        """True when at least one road leaving this node is open to *mode*.

        The same test `_relax` makes, asked one step earlier. A node whose
        every edge is closed to the mode is a dead end the search can never
        leave, so snapping to it is the same as refusing the journey.
        """
        for _to, _dist, kind, _name in self.adj.get(node_id, ()):
            speed = edge_speed(mode, kind)
            if speed is not None and speed > 0:
                return True
        return False

    def _snap(self, lat: float, lon: float, mode: str) -> tuple[int | None, float]:
        """The nearest road this mode can actually set off on.

        Not the nearest road. That is what this did, and it is why standing in
        a park made the router refuse to route: the nearest node was a
        footpath twenty metres away, no car may leave a footpath, and A* gave
        up on its first pop — with the boulevard two hundred metres off, in a
        bundle that knew about it perfectly well. Measured on Mali's own
        bundle, 47 of 400 points sampled around Bamako — better than one in
        ten — refused a car route for exactly that reason, and every one of
        them routes once the snap is asked to respect the mode.

        The cost is a longer walk to the road, which is honest and is the
        answer's to report: `_Ends` carries it, the geometry draws it, and the
        caller can say "a hundred and forty metres to the road" instead of
        "no route".

        Falling back to the unfiltered nearest keeps the old behaviour for the
        bundle where no node anywhere is open to the mode — a refusal is still
        the right answer there, and it should come from the search rather than
        from an empty snap.
        """
        node, metres = self.nearest_with_distance(lat, lon, lambda nid: self._travels(nid, mode))
        if node is not None:
            return node, metres
        return self.nearest_with_distance(lat, lon)

    def _road_at(self, node_ids: list[int], index: int) -> str:
        """The name of the road at one end of a path, or empty when it has none."""
        if len(node_ids) < 2:
            return ""
        first, second = (node_ids[0], node_ids[1]) if index == 0 else (node_ids[-2], node_ids[-1])
        for to_node, _dist, _kind, name in self.adj.get(first, ()):
            if to_node == second:
                return name or ""
        return ""

    def _island(self, node_id: int, mode: str) -> set[int] | None:
        """What this node can reach, when that is few enough to be an island.

        `None` once the fill runs past `ISLAND_LIMIT`: the node is on the
        network proper, and a search that failed from there failed for a
        reason no amount of re-snapping will fix.
        """
        seen = {node_id}
        stack = [node_id]
        # Bounded, and the bound is the answer: running out of budget is what
        # "this is not an island" means here.
        for _ in range(ISLAND_LIMIT):
            if not stack:
                return seen
            for to_node, _dist, kind, _name in self.adj.get(stack.pop(), ()):
                speed = edge_speed(mode, kind)
                if to_node in seen or speed is None or speed <= 0:
                    continue
                seen.add(to_node)
                stack.append(to_node)
        return None

    def _off_the_island(
        self, node: int, point: tuple[float, float], mode: str
    ) -> tuple[int, float] | None:
        """Snap again, ignoring the stub of road the last attempt landed on.

        `None` when that road is not a stub — the node is on the network
        proper, and a search that failed from there failed for a reason no
        amount of re-snapping will fix.
        """
        island = self._island(node, mode)
        if island is None:
            return None
        other, metres = self.nearest_with_distance(
            point[0], point[1], lambda nid: nid not in island and self._travels(nid, mode)
        )
        if other is None or metres > MAX_ESCAPE_M:
            return None
        return other, metres

    def _escape_islands(
        self, start_node: int, end_node: int, ends: _Ends, mode: str
    ) -> tuple[int, int, _Ends] | None:
        """Re-snap whichever end is stranded on an isolated stub of road.

        OpenStreetMap is full of them: a hundred metres of `track` behind a
        compound, drawn by somebody who never joined it to the street. Snap to
        one and the search has nowhere to go — and the reader is told there is
        no way from their village to Bamako, because of twelve segments nobody
        finished mapping.

        Once, and only after a search has already failed, so an ordinary
        journey pays nothing for it. `None` when neither end is on an island,
        which means the two points genuinely are not joined by a road this
        mode may take — and that is an answer, not a failure.
        """
        again = self._off_the_island(start_node, ends.start, mode)
        if again is not None:
            return again[0], end_node, replace(ends, from_snap_m=again[1])

        again = self._off_the_island(end_node, ends.end, mode)
        if again is not None:
            return start_node, again[0], replace(ends, to_snap_m=again[1])

        return None

    def _empty_result(self, ends: _Ends, mode: str) -> RouteResult:
        return RouteResult(
            distance_km=0,
            duration_min=0,
            nodes=[],
            geometry={"type": "LineString", "coordinates": []},
            steps=[],
            found=False,
            from_snap_m=ends.from_snap_m,
            to_snap_m=ends.to_snap_m,
            mode=mode,
        )

    def _same_node_result(self, node: int, ends: _Ends, mode: str) -> RouteResult:
        return RouteResult(
            distance_km=0,
            duration_min=0,
            nodes=[node],
            geometry=self._build_geometry([node], ends),
            steps=[],
            found=True,
            from_snap_m=ends.from_snap_m,
            to_snap_m=ends.to_snap_m,
            mode=mode,
        )

    def _edge_budget(self) -> int:
        """Pops A* may perform: one per queue entry, and entries never exceed
        one per edge plus the start. Caches, since it walks the whole graph."""
        if self._edge_count is None:
            self._edge_count = sum(len(edges) for edges in self.adj.values())
        return self._edge_count + 1

    def _reconstruct(self, came_from: dict[int, int], node: int) -> list[int]:
        """Walk the parent chain back to the start.

        Bounded by the number of nodes: `came_from` gains one entry per node at
        most, so the chain cannot be longer than the graph.
        """
        path = [node]
        for _ in range(len(self.coords)):
            if node not in came_from:
                break
            node = came_from[node]
            path.append(node)
        path.reverse()
        return path

    def _astar(
        self,
        start_node: int,
        end_node: int,
        mode: str,
        penalties: dict[tuple[int, int], float] | None = None,
    ) -> _Search | None:
        """Quickest path for *mode*, or None if there is none it may take.

        None covers two different things on purpose, because the caller
        answers them the same way: nothing connects the two points, and
        nothing connects them *by this means* — a pedestrian on either side of
        a motorway junction with no crossing is not on a broken map.

        `penalties` makes named streets dearer without touching the graph.
        That distinction matters: the graph is loaded once and shared by every
        request in the process, so a search that edited it to find an
        alternative would be editing the map under everybody else.

        There is deliberately no cost ceiling. One was written, on the belief
        that a penalised search was ruinously slow; measured against Mali's
        bundle with the graph already loaded, three routes across Bamako cost
        0.10 s unbounded and 1.56 s bounded — and the bound also threw away
        the third route. The twenty-two seconds that prompted it were the
        graph loading, once, on the first request of a process.
        """
        end_lat, end_lon = self.coords[end_node]
        # Divided by the fastest this mode ever goes, so the estimate is never
        # more than the time really left. An admissible heuristic is what
        # makes A* return the quickest route rather than an early one.
        top_ms = top_speed_kmh(mode) * _KMH_TO_MS

        frontier = _Frontier(
            open_set=[(0.0, start_node)],
            came_from={},
            cost_s={start_node: 0.0},
            time_s={start_node: 0.0},
            distance_m={start_node: 0.0},
            visited=set(),
        )

        # One pop per queue entry, and entries are bounded by the edge count.
        for _ in range(self._edge_budget()):
            if not frontier.open_set:
                break
            _, current = heapq.heappop(frontier.open_set)

            if current == end_node:
                return _Search(
                    path=self._reconstruct(frontier.came_from, current),
                    distance_m=frontier.distance_m[end_node],
                    duration_s=frontier.time_s[end_node],
                )

            if current in frontier.visited:
                continue
            frontier.visited.add(current)
            self._relax(current, frontier, mode, top_ms, (end_lat, end_lon), penalties)

        return None

    def _relax(
        self,
        current: int,
        frontier: _Frontier,
        mode: str,
        top_ms: float,
        goal: tuple[float, float],
        penalties: dict[tuple[int, int], float] | None = None,
    ) -> None:
        """Relax every edge leaving *current* that this mode may travel."""
        end_lat, end_lon = goal
        for to_node, dist_m, kind, _ in self.adj.get(current, []):
            if to_node in frontier.visited:
                continue
            speed = edge_speed(mode, kind)
            # Not slow — forbidden. A bicycle sent onto a trunk road to save
            # four minutes is the failure a speed alone cannot prevent.
            if speed is None or speed <= 0:
                continue
            # The penalty is on the *cost*, never on the reported time: a
            # route made dearer to find is not a route that takes longer to
            # travel, and `_Search` carries the honest seconds.
            penalty = 1.0
            if penalties:
                penalty = penalties.get(_street(current, to_node), 1.0)
            seconds = dist_m / (speed * _KMH_TO_MS)
            travelled = frontier.time_s[current] + seconds
            tentative = frontier.cost_s[current] + penalty * seconds
            if tentative >= frontier.cost_s.get(to_node, float("inf")):
                continue
            frontier.came_from[to_node] = current
            frontier.cost_s[to_node] = tentative
            frontier.time_s[to_node] = travelled
            frontier.distance_m[to_node] = frontier.distance_m[current] + dist_m
            nlat, nlon = self.coords.get(to_node, (0.0, 0.0))
            heapq.heappush(
                frontier.open_set,
                (tentative + _haversine(nlat, nlon, end_lat, end_lon) / top_ms, to_node),
            )

    def route(
        self,
        start_lat: float,
        start_lon: float,
        end_lat: float,
        end_lon: float,
        mode: str = "car",
    ) -> RouteResult:
        """Find the quickest route between two GPS coordinates using A*.

        `mode` is one of `profiles.MODES` and changes both numbers and
        geometry: a bicycle is not a slow car, it is barred from the roads a
        car is fastest on, and it comes back by another street. It changes
        where the journey is joined, too — see `_snap`.
        """
        start_node, from_snap_m = self._snap(start_lat, start_lon, mode)
        end_node, to_snap_m = self._snap(end_lat, end_lon, mode)
        ends = _Ends(
            start=(start_lat, start_lon),
            end=(end_lat, end_lon),
            from_snap_m=from_snap_m,
            to_snap_m=to_snap_m,
        )

        if not start_node or not end_node:
            return self._empty_result(ends, mode)

        if start_node == end_node:
            return self._same_node_result(start_node, ends, mode)

        search = self._astar(start_node, end_node, mode)
        if search is None:
            escaped = self._escape_islands(start_node, end_node, ends, mode)
            if escaped is None:
                return self._empty_result(ends, mode)
            start_node, end_node, ends = escaped
            search = self._astar(start_node, end_node, mode)
        if search is None:
            return self._empty_result(ends, mode)

        return self._dress(search, ends, mode)

    def routes(
        self,
        start_lat: float,
        start_lon: float,
        end_lat: float,
        end_lon: float,
        mode: str = "car",
        count: int = 3,
    ) -> list[RouteResult]:
        """The quickest route, then up to `count - 1` genuinely different ones.

        Sorted by how long they take, quickest first. Not by the order they
        were found, which is a different order and was briefly mistaken for
        this one: the second pass minimises a *penalised* cost, so the third
        candidate can be genuinely quicker than the second. Measured on Mali's
        own bundle it was — 12.2 minutes offered below 12.5 — and a list whose
        second entry is slower than its third is a list the reader has to
        check rather than read.

        The method is penalty and re-search: trace the best route, triple the
        cost of the streets it used, ask again, and keep the answer only if it
        shares less than `MAX_OVERLAP` of its length with one already offered.
        Two routes down the same road with a different turn at the end are one
        route, and offering them as two wastes the reader's attention on the
        one decision they do not have to make.

        Fewer than `count` come back when there are genuinely fewer ways —
        a town with one bridge has one route out, and inventing a second would
        be inventing a road. One is a complete answer; none means the same as
        `route()` returning `found=False`.
        """
        start_node, from_snap_m = self._snap(start_lat, start_lon, mode)
        end_node, to_snap_m = self._snap(end_lat, end_lon, mode)
        ends = _Ends(
            start=(start_lat, start_lon),
            end=(end_lat, end_lon),
            from_snap_m=from_snap_m,
            to_snap_m=to_snap_m,
        )

        if not start_node or not end_node:
            return [self._empty_result(ends, mode)]
        if start_node == end_node:
            return [self._same_node_result(start_node, ends, mode)]

        found = self._search_alternatives(start_node, end_node, mode, count)
        if not found:
            escaped = self._escape_islands(start_node, end_node, ends, mode)
            if escaped is None:
                return [self._empty_result(ends, mode)]
            start_node, end_node, ends = escaped
            found = self._search_alternatives(start_node, end_node, mode, count)
        if not found:
            return [self._empty_result(ends, mode)]
        found.sort(key=lambda search: search.duration_s)
        return [self._dress(search, ends, mode) for search in found]

    def _search_alternatives(
        self, start_node: int, end_node: int, mode: str, count: int
    ) -> list[_Search]:
        """Run A* repeatedly, making what was used dearer each time (R2)."""
        accepted: list[_Search] = []
        seen: list[_Streets] = []
        penalties: dict[tuple[int, int], float] = {}

        # Bounded: a city whose every way out is the same road never yields a
        # second route, however many times it is asked.
        for _ in range(MAX_ALTERNATIVE_ATTEMPTS):
            if len(accepted) >= count:
                break
            search = self._astar(start_node, end_node, mode, penalties or None)
            if search is None:
                break
            streets = self._streets_of(search.path)
            if not any(_overlap(streets, earlier) > MAX_OVERLAP for earlier in seen):
                accepted.append(search)
                seen.append(streets)
            # Penalised whether or not it was kept: a candidate rejected for
            # looking like the last one must not be handed back unchanged on
            # the next pass, or the loop spends its whole budget on it.
            for street in streets:
                penalties[street] = penalties.get(street, 1.0) * ALTERNATIVE_PENALTY

        return accepted

    def _streets_of(self, path: list[int]) -> _Streets:
        """Each street the path uses, and how much of it — for the overlap."""
        streets: _Streets = {}
        for i in range(len(path) - 1):
            for to_node, dist_m, _, _ in self.adj.get(path[i], []):
                if to_node == path[i + 1]:
                    streets[_street(path[i], to_node)] = dist_m
                    break
        return streets

    def _dress(self, search: _Search, ends: _Ends, mode: str) -> RouteResult:
        """One A* answer, as the caller receives it.

        The kilometres and the minutes are the road's, and only the road's.
        The walk to it is reported beside them rather than folded into them:
        how long it takes to cover a hundred metres of park on foot is not
        something a car profile can be asked, and adding a guess to a measured
        figure would make both unreadable.
        """
        return RouteResult(
            distance_km=round(search.distance_m / 1000, 1),
            duration_min=round(search.duration_s / 60, 1),
            nodes=search.path,
            geometry=self._build_geometry(search.path, ends),
            steps=self._build_steps(search.path, ends),
            found=True,
            from_snap_m=ends.from_snap_m,
            to_snap_m=ends.to_snap_m,
            from_road=self._road_at(search.path, 0),
            to_road=self._road_at(search.path, -1),
            mode=mode,
        )

    def _build_geometry(self, node_ids: list[int], ends: _Ends) -> GeoJson:
        """The line, from where the journey was asked to start to where it ends.

        The road part is the path's nodes. The two straight stubs at the ends
        are the walk between the point somebody named and the road nearest to
        it — drawn, because a line that begins two hundred metres from the
        reader's pin looks like a line drawn for somebody else. Straight and
        not routed: there is no data for the ground it crosses, and a
        confident-looking detour across a car park would be an invention. It
        gives the direction, which is what it is for.

        Under `ACCESS_LEG_M` the stub is dropped rather than drawn: it would
        be a few pixels of noise on the end of the route.
        """
        coords = []
        if ends.from_snap_m > ACCESS_LEG_M:
            coords.append([ends.start[1], ends.start[0]])
        for nid in node_ids:
            c = self.coords.get(nid)
            if c:
                coords.append([c[1], c[0]])  # [lon, lat]
        if ends.to_snap_m > ACCESS_LEG_M:
            coords.append([ends.end[1], ends.end[0]])
        return {"type": "LineString", "coordinates": coords}

    def _build_steps(self, node_ids: list[int], ends: _Ends) -> list[dict[str, object]]:
        """The journey in words, the walk to the road included.

        The first and last steps are the ones this used to leave out. Somebody
        standing in a park was told to continue on a boulevard they were not
        on; saying which road to reach, and how far it is, is the difference
        between an instruction and a riddle.
        """
        steps: list[dict[str, object]] = []
        if ends.from_snap_m > ACCESS_LEG_M:
            road = self._road_at(node_ids, 0)
            steps.append(self._access_step(f"Reach {road or 'the road'}", road, ends.from_snap_m))

        steps.extend(self._road_steps(node_ids))

        if ends.to_snap_m > ACCESS_LEG_M:
            road = self._road_at(node_ids, -1)
            steps.append(
                self._access_step(
                    f"Leave {road or 'the road'} for the destination", road, ends.to_snap_m
                )
            )
        return steps

    @staticmethod
    def _access_step(instruction: str, road: str, metres: float) -> dict[str, object]:
        """One end of the journey that is not on a road.

        Marked `access` so a caller can tell it apart from the driving: it is
        the only step whose distance nobody drives.
        """
        return {
            "instruction": instruction,
            "distance_m": round(metres),
            "name": road,
            "access": True,
        }

    def _road_steps(self, node_ids: list[int]) -> list[dict[str, object]]:
        """The path itself, one step per named stretch of road."""
        steps: list[dict[str, object]] = []
        prev_name = ""
        seg_distance = 0.0

        for i in range(len(node_ids) - 1):
            n_from = node_ids[i]
            n_to = node_ids[i + 1]

            # Find edge
            name = ""
            dist = 0.0
            for to, d, _, nm in self.adj.get(n_from, []):
                if to == n_to:
                    name = nm or "road"
                    dist = d
                    break

            if name != prev_name and prev_name:
                steps.append(
                    {
                        "instruction": f"Continue on {prev_name}",
                        "distance_m": round(seg_distance),
                        "name": prev_name,
                    }
                )
                seg_distance = 0.0

            seg_distance += dist
            prev_name = name

        if prev_name:
            steps.append(
                {
                    "instruction": f"Arrive via {prev_name}",
                    "distance_m": round(seg_distance),
                    "name": prev_name,
                }
            )
        return steps

from_db(conn) classmethod

Load the entire graph into memory from a .yoromaps database.

Source code in src/yoromaps/routing.py

@classmethod
def from_db(cls, conn: sqlite3.Connection) -> Graph:
    """Load the entire graph into memory from a .yoromaps database."""
    g = cls()

    for row in conn.execute("SELECT id, lat, lon FROM nodes"):
        g.coords[row[0]] = (row[1], row[2])

    kinds: dict[tuple[str, str, str], EdgeKind] = {}
    # `edge_tag_columns` returns one of two literals, never caller input.
    for row in conn.execute(
        "SELECT from_node, to_node, distance_m, road_type, name, "
        f"{edge_tag_columns(conn)} FROM edges"
    ):
        from_n, to_n, dist, road_type, name, surface, closed_to = row
        if from_n not in g.adj:
            g.adj[from_n] = []
        # Shared: a country has millions of edges and a few hundred kinds
        # between them. Without this the same short strings are held a
        # million times over, for tens of megabytes of nothing.
        key = (road_type or "road", surface or "", closed_to or "")
        kind = kinds.get(key)
        if kind is None:
            closed = frozenset(m for m in key[2].split(",") if m)
            kind = kinds[key] = (sys.intern(key[0]), sys.intern(key[1]), closed)
        g.adj[from_n].append((to_n, dist, kind, name or ""))

    return g

nearest_with_distance(lat, lon, accept=None)

Nearest node + haversine distance in metres.

Uses a uniform grid index (built lazily on first call) and searches outward ring by ring. The distance metric matches the historical behavior: squared degree distance for the argmin, haversine for the returned distance.

accept narrows what counts as a node. Without it this answers "what is the nearest piece of road", which is the wrong question for a router: see _snap.

Source code in src/yoromaps/routing.py

def nearest_with_distance(
    self, lat: float, lon: float, accept: Callable[[int], bool] | None = None
) -> tuple[int | None, float]:
    """Nearest node + haversine distance in metres.

    Uses a uniform grid index (built lazily on first call) and searches
    outward ring by ring. The distance metric matches the historical
    behavior: squared degree distance for the argmin, haversine for the
    returned distance.

    `accept` narrows what counts as a node. Without it this answers "what
    is the nearest piece of road", which is the wrong question for a
    router: see `_snap`.
    """
    if not self.coords:
        return None, float("inf")
    if self._grid is None:
        self._build_grid()

    ci = math.floor(lat / GRID_CELL_DEG)
    cj = math.floor(lon / GRID_CELL_DEG)

    assert self._grid_extent is not None
    i_min, i_max, j_min, j_max = self._grid_extent
    # Upper bound on rings: enough to reach the farthest populated cell.
    max_ring = max(abs(ci - i_min), abs(ci - i_max), abs(cj - j_min), abs(cj - j_max)) + 1

    best_id: int | None = None
    best_sq = float("inf")

    for ring in range(max_ring + 1):
        # Once a candidate is found, a node in ring r is at least
        # (r-1)*cell degrees away — stop when that exceeds the best.
        if best_id is not None and ((ring - 1) * GRID_CELL_DEG) ** 2 > best_sq:
            break
        if self._ring_is_outside(ci, cj, ring):
            continue
        cells = self._ring_cells(ci, cj, ring)
        best_id, best_sq = self._closest_in_cells(cells, lat, lon, best_id, best_sq, accept)

    if best_id is None:
        return None, float("inf")
    nlat, nlon = self.coords[best_id]
    return best_id, _haversine(lat, lon, nlat, nlon)

route(start_lat, start_lon, end_lat, end_lon, mode='car')

Find the quickest route between two GPS coordinates using A*.

mode is one of profiles.MODES and changes both numbers and geometry: a bicycle is not a slow car, it is barred from the roads a car is fastest on, and it comes back by another street. It changes where the journey is joined, too — see _snap.

Source code in src/yoromaps/routing.py

def route(
    self,
    start_lat: float,
    start_lon: float,
    end_lat: float,
    end_lon: float,
    mode: str = "car",
) -> RouteResult:
    """Find the quickest route between two GPS coordinates using A*.

    `mode` is one of `profiles.MODES` and changes both numbers and
    geometry: a bicycle is not a slow car, it is barred from the roads a
    car is fastest on, and it comes back by another street. It changes
    where the journey is joined, too — see `_snap`.
    """
    start_node, from_snap_m = self._snap(start_lat, start_lon, mode)
    end_node, to_snap_m = self._snap(end_lat, end_lon, mode)
    ends = _Ends(
        start=(start_lat, start_lon),
        end=(end_lat, end_lon),
        from_snap_m=from_snap_m,
        to_snap_m=to_snap_m,
    )

    if not start_node or not end_node:
        return self._empty_result(ends, mode)

    if start_node == end_node:
        return self._same_node_result(start_node, ends, mode)

    search = self._astar(start_node, end_node, mode)
    if search is None:
        escaped = self._escape_islands(start_node, end_node, ends, mode)
        if escaped is None:
            return self._empty_result(ends, mode)
        start_node, end_node, ends = escaped
        search = self._astar(start_node, end_node, mode)
    if search is None:
        return self._empty_result(ends, mode)

    return self._dress(search, ends, mode)

routes(start_lat, start_lon, end_lat, end_lon, mode='car', count=3)

The quickest route, then up to count - 1 genuinely different ones.

Sorted by how long they take, quickest first. Not by the order they were found, which is a different order and was briefly mistaken for this one: the second pass minimises a penalised cost, so the third candidate can be genuinely quicker than the second. Measured on Mali's own bundle it was — 12.2 minutes offered below 12.5 — and a list whose second entry is slower than its third is a list the reader has to check rather than read.

The method is penalty and re-search: trace the best route, triple the cost of the streets it used, ask again, and keep the answer only if it shares less than MAX_OVERLAP of its length with one already offered. Two routes down the same road with a different turn at the end are one route, and offering them as two wastes the reader's attention on the one decision they do not have to make.

Fewer than count come back when there are genuinely fewer ways — a town with one bridge has one route out, and inventing a second would be inventing a road. One is a complete answer; none means the same as route() returning found=False.

Source code in src/yoromaps/routing.py

def routes(
    self,
    start_lat: float,
    start_lon: float,
    end_lat: float,
    end_lon: float,
    mode: str = "car",
    count: int = 3,
) -> list[RouteResult]:
    """The quickest route, then up to `count - 1` genuinely different ones.

    Sorted by how long they take, quickest first. Not by the order they
    were found, which is a different order and was briefly mistaken for
    this one: the second pass minimises a *penalised* cost, so the third
    candidate can be genuinely quicker than the second. Measured on Mali's
    own bundle it was — 12.2 minutes offered below 12.5 — and a list whose
    second entry is slower than its third is a list the reader has to
    check rather than read.

    The method is penalty and re-search: trace the best route, triple the
    cost of the streets it used, ask again, and keep the answer only if it
    shares less than `MAX_OVERLAP` of its length with one already offered.
    Two routes down the same road with a different turn at the end are one
    route, and offering them as two wastes the reader's attention on the
    one decision they do not have to make.

    Fewer than `count` come back when there are genuinely fewer ways —
    a town with one bridge has one route out, and inventing a second would
    be inventing a road. One is a complete answer; none means the same as
    `route()` returning `found=False`.
    """
    start_node, from_snap_m = self._snap(start_lat, start_lon, mode)
    end_node, to_snap_m = self._snap(end_lat, end_lon, mode)
    ends = _Ends(
        start=(start_lat, start_lon),
        end=(end_lat, end_lon),
        from_snap_m=from_snap_m,
        to_snap_m=to_snap_m,
    )

    if not start_node or not end_node:
        return [self._empty_result(ends, mode)]
    if start_node == end_node:
        return [self._same_node_result(start_node, ends, mode)]

    found = self._search_alternatives(start_node, end_node, mode, count)
    if not found:
        escaped = self._escape_islands(start_node, end_node, ends, mode)
        if escaped is None:
            return [self._empty_result(ends, mode)]
        start_node, end_node, ends = escaped
        found = self._search_alternatives(start_node, end_node, mode, count)
    if not found:
        return [self._empty_result(ends, mode)]
    found.sort(key=lambda search: search.duration_s)
    return [self._dress(search, ends, mode) for search in found]

route(conn, start_lat, start_lon, end_lat, end_lon, mode='car')

Find the quickest route between two GPS coordinates.

Loads the graph into memory on each call. For multiple routes, use Graph.from_db() directly.

Source code in src/yoromaps/routing.py

def route(
    conn: sqlite3.Connection,
    start_lat: float,
    start_lon: float,
    end_lat: float,
    end_lon: float,
    mode: str = "car",
) -> RouteResult:
    """Find the quickest route between two GPS coordinates.

    Loads the graph into memory on each call. For multiple routes,
    use ``Graph.from_db()`` directly.
    """
    graph = Graph.from_db(conn)
    return graph.route(start_lat, start_lon, end_lat, end_lon, mode)

route_from_codes(conn, codes, graph=None, mode='car')

Route through multiple Yoro codes in order.

Returns a list of RouteResult, one per leg. Pass a preloaded graph (e.g. from :func:get_graph) to avoid reloading it from the database — the same graph answers for every mode, so one load serves them all.

Source code in src/yoromaps/routing.py

def route_from_codes(
    conn: sqlite3.Connection,
    codes: list[str],
    graph: Graph | None = None,
    mode: str = "car",
) -> list[RouteResult]:
    """Route through multiple Yoro codes in order.

    Returns a list of RouteResult, one per leg. Pass a preloaded *graph*
    (e.g. from :func:`get_graph`) to avoid reloading it from the database —
    the same graph answers for every mode, so one load serves them all.
    """
    import yoro

    if graph is None:
        graph = Graph.from_db(conn)
    decoded = [yoro.decode(c) for c in codes]
    points = [(d["lat"], d["lon"]) for d in decoded]
    return [
        graph.route(points[i][0], points[i][1], points[i + 1][0], points[i + 1][1], mode)
        for i in range(len(points) - 1)
    ]

routes_between_codes(conn, from_code, to_code, graph=None, mode='car', count=3)

Several ways from one Yoro code to another, quickest first.

One leg only, unlike route_from_codes. Alternatives through a list of waypoints multiply out — three ways to each of four stops is eighty-one journeys — and nobody asked to choose between eighty-one.

Source code in src/yoromaps/routing.py

def routes_between_codes(
    conn: sqlite3.Connection,
    from_code: str,
    to_code: str,
    graph: Graph | None = None,
    mode: str = "car",
    count: int = 3,
) -> list[RouteResult]:
    """Several ways from one Yoro code to another, quickest first.

    One leg only, unlike `route_from_codes`. Alternatives through a list of
    waypoints multiply out — three ways to each of four stops is eighty-one
    journeys — and nobody asked to choose between eighty-one.
    """
    import yoro

    if graph is None:
        graph = Graph.from_db(conn)
    start = yoro.decode(from_code)
    end = yoro.decode(to_code)
    return graph.routes(start["lat"], start["lon"], end["lat"], end["lon"], mode, count)

get_graph(db_path)

Load the road graph for a .yoromaps file, cached per process.

The graph is loaded once and reused across calls (and across HTTP requests in Django). The cache entry is invalidated when the file's mtime changes, e.g. after yoromaps update.

Source code in src/yoromaps/routing.py

def get_graph(db_path: str | Path) -> Graph:
    """Load the road graph for a .yoromaps file, cached per process.

    The graph is loaded once and reused across calls (and across HTTP
    requests in Django). The cache entry is invalidated when the file's
    mtime changes, e.g. after ``yoromaps update``.
    """
    path = str(Path(db_path).resolve())
    mtime = os.path.getmtime(path)

    cached = _graph_cache.get(path)
    if cached and cached[0] == mtime:
        return cached[1]

    with _graph_lock:
        cached = _graph_cache.get(path)
        if cached and cached[0] == mtime:
            return cached[1]

        conn = sqlite3.connect(path)
        try:
            graph = Graph.from_db(conn)
        finally:
            conn.close()
        _graph_cache[path] = (mtime, graph)
        return graph

yoromaps.poi

Points of interest, addressed by Yoro code. Every POI is encoded in the database's country domain at insertion time, falling back to the worldwide XX domain for points outside the country's bounding box.

yoromaps.poi

POI management — local points of interest addressed by Yoro codes.

Every POI gets a Yoro code at insertion time (encoded in the database's country domain, falling back to the global XX domain when the point lies outside the country bbox). The idx_pois_yoro index makes "what is at this address" lookups instant; radius and cell queries use a lat/lon bbox prefilter.

Usage::

conn = yoromaps.open_db("mali.yoromaps")
poi = add_poi(conn, 12.639, -8.002, "Grand marche", category="market")
print(poi["yoro_code"])                      # "ML-..."
pois_near(conn, 12.64, -8.0, radius_m=500)   # sorted by distance
pois_in_cell(conn, "ML-4H7A3B")

add_poi(conn, lat, lon, name, category='other', source='local', osm_id=None, details=None)

Insert a POI; its Yoro code is computed automatically.

details is one argument rather than four because a caller usually knows none of them or all of them, and four empty strings at every call site would say nothing four times.

Source code in src/yoromaps/poi.py

def add_poi(
    conn: sqlite3.Connection,
    lat: float,
    lon: float,
    name: str,
    category: str = "other",
    source: str = "local",
    osm_id: int | None = None,
    details: PoiDetails | None = None,
) -> Poi:
    """Insert a POI; its Yoro code is computed automatically.

    `details` is one argument rather than four because a caller usually knows
    none of them or all of them, and four empty strings at every call site
    would say nothing four times.
    """
    code = _encode_for_db(conn, lat, lon)
    said: PoiDetails = details or {}
    phone = said.get("phone", "")
    website = said.get("website", "")
    opening_hours = said.get("opening_hours", "")
    address = said.get("address", "")
    cur = conn.execute(
        "INSERT INTO pois (lat, lon, yoro_code, name, category, source, osm_id, "
        "phone, website, opening_hours, address) "
        "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
        (lat, lon, code, name, category, source, osm_id, phone, website, opening_hours, address),
    )
    conn.commit()
    if cur.lastrowid is None:
        raise RuntimeError("SQLite did not report a row id for the inserted POI")
    return {
        "id": cur.lastrowid,
        "lat": lat,
        "lon": lon,
        "yoro_code": code,
        "name": name,
        "category": category,
        "source": source,
        "osm_id": osm_id,
        "distance_m": None,
        "phone": phone,
        "website": website,
        "opening_hours": opening_hours,
        "address": address,
    }

pois_near(conn, lat, lon, radius_m=1000, category=None, limit=100)

POIs within radius_m metres, sorted by distance.

Source code in src/yoromaps/poi.py

def pois_near(
    conn: sqlite3.Connection,
    lat: float,
    lon: float,
    radius_m: float = 1000,
    category: str | None = None,
    limit: int = 100,
) -> list[Poi]:
    """POIs within *radius_m* metres, sorted by distance."""
    from yoromaps.extract import haversine

    dlat = radius_m / 111_000
    dlon = radius_m / (111_000 * max(0.1, math.cos(math.radians(lat))))

    sql = "SELECT * FROM pois WHERE lat BETWEEN ? AND ? AND lon BETWEEN ? AND ?"
    params: list[object] = [lat - dlat, lat + dlat, lon - dlon, lon + dlon]
    if category:
        sql += " AND category = ?"
        params.append(category)

    hits: list[tuple[float, sqlite3.Row]] = []
    for row in conn.execute(sql, params):
        d = haversine(lat, lon, row["lat"], row["lon"])
        if d <= radius_m:
            hits.append((d, row))
    hits.sort(key=lambda t: t[0])
    return [_poi_dict(row, distance_m=d) for d, row in hits[:limit]]

pois_in_cell(conn, code, category=None)

All POIs inside the cell of a Yoro code (any precision).

Raises ValueError for invalid codes.

Source code in src/yoromaps/poi.py

def pois_in_cell(
    conn: sqlite3.Connection,
    code: str,
    category: str | None = None,
) -> list[Poi]:
    """All POIs inside the cell of a Yoro code (any precision).

    Raises ValueError for invalid codes.
    """
    b = yoro.decode(code)["bounds"]
    sql = "SELECT * FROM pois WHERE lat >= ? AND lat < ? AND lon >= ? AND lon < ?"
    params: list[object] = [b["lat_min"], b["lat_max"], b["lon_min"], b["lon_max"]]
    if category:
        sql += " AND category = ?"
        params.append(category)
    return [_poi_dict(r) for r in conn.execute(sql, params)]

search_pois(conn, query=None, category=None, limit=50)

Search POIs by name substring and/or category.

Source code in src/yoromaps/poi.py

def search_pois(
    conn: sqlite3.Connection,
    query: str | None = None,
    category: str | None = None,
    limit: int = 50,
) -> list[Poi]:
    """Search POIs by name substring and/or category."""
    sql = "SELECT * FROM pois WHERE 1=1"
    params: list[object] = []
    if query:
        sql += " AND name LIKE ?"
        params.append(f"%{query}%")
    if category:
        sql += " AND category = ?"
        params.append(category)
    sql += " ORDER BY name LIMIT ?"
    params.append(limit)
    return [_poi_dict(r) for r in conn.execute(sql, params)]

delete_poi(conn, poi_id)

Delete a POI by id. Returns True if a row was removed.

Source code in src/yoromaps/poi.py

def delete_poi(conn: sqlite3.Connection, poi_id: int) -> bool:
    """Delete a POI by id. Returns True if a row was removed."""
    cur = conn.execute("DELETE FROM pois WHERE id = ?", (poi_id,))
    conn.commit()
    return cur.rowcount > 0

yoromaps.osm_poi

OpenStreetMap POI extraction. Categories are an explicit tag allowlist, not a denylist: what matches nothing is counted and reported back in stats["unmapped"] rather than guessed at. See the Points of Interest guide.

yoromaps.osm_poi

Extract points of interest from an OSM PBF into a .yoromaps database.

The road graph tells you how to get somewhere; this tells you what is there. Shops, hospitals, schools, markets and places of worship already mapped in OpenStreetMap are read out of the country extract, given their Yoro code, and written to the pois table under source='osm' — which is what makes a freshly built bundle useful offline instead of empty.

A name and a point make a dot on a map. What makes it worth opening is what the shopfront says: a number to ring, the hours it keeps, the address it goes by. Those are read too, from the tags that carry them. Most POIs have none — in Togo about one in five carries any of them — and that is the normal case, not a failure: the fields are empty, never absent.

OSM's tagging is far richer than any address book's categories, so the mapping here is an allowlist: a tag combination that is not listed is not a POI. That keeps parking bays, benches and waste baskets out of the map. What was skipped is counted and reported, so the taxonomy can grow from evidence rather than from guesses.

Categories are deliberately those of the consuming application, so a single filter spans OSM-sourced and user-contributed places alike.

DEFAULT_PRECISION = 19 module-attribute

VALUE_CATEGORIES = {'amenity': {'restaurant': 'restaurant', 'fast_food': 'restaurant', 'cafe': 'restaurant', 'bar': 'restaurant', 'pub': 'restaurant', 'food_court': 'restaurant', 'ice_cream': 'restaurant', 'marketplace': 'market', 'pharmacy': 'pharmacy', 'hospital': 'hospital', 'clinic': 'hospital', 'doctors': 'hospital', 'dentist': 'hospital', 'health_post': 'hospital', 'school': 'school', 'college': 'school', 'university': 'school', 'kindergarten': 'school', 'language_school': 'school', 'driving_school': 'school', 'library': 'school', 'place_of_worship': 'place_of_worship', 'townhall': 'government', 'courthouse': 'government', 'police': 'government', 'fire_station': 'government', 'post_office': 'government', 'embassy': 'government', 'prison': 'government', 'bus_station': 'transport', 'ferry_terminal': 'transport', 'taxi': 'transport', 'car_rental': 'transport', 'bank': 'other', 'fuel': 'other', 'bureau_de_change': 'other', 'money_transfer': 'other'}, 'tourism': {'museum': 'landmark', 'attraction': 'landmark', 'viewpoint': 'landmark', 'artwork': 'landmark', 'gallery': 'landmark', 'hotel': 'other', 'guest_house': 'other', 'hostel': 'other', 'motel': 'other'}, 'railway': {'station': 'transport', 'halt': 'transport'}, 'aeroway': {'aerodrome': 'transport', 'terminal': 'transport'}, 'public_transport': {'station': 'transport'}, 'office': {'government': 'government', 'diplomatic': 'government'}} module-attribute

KEY_CATEGORIES = {'shop': 'shop', 'craft': 'shop', 'healthcare': 'hospital', 'historic': 'landmark'} module-attribute

extract_pois(pbf_path, conn, progress=None, source=OSM_SOURCE, precision=DEFAULT_PRECISION)

Extract POIs from a PBF into the database, replacing previous ones.

Rows carrying source are deleted first, so a rebuild refreshes OSM data without touching POIs from anywhere else. The delete and every insert share one transaction: a failure mid-extraction leaves the previous set in place.

Rows are written in batches of POI_INSERT_BATCH rather than accumulated, so peak memory does not grow with the size of the country.

Nameless features are skipped — a pin with no label is noise on a map whose whole purpose is telling someone where to go.

Parameters:

Name	Type	Description	Default
pbf_path	str | Path	Path to the .osm.pbf file.	required
conn	Connection	Open connection to a .yoromaps database.	required
progress	ProgressCallback | None	Optional callable(message, current, total).	None
source	str	Value written to pois.source, and the rows replaced.	OSM_SOURCE
precision	int	Yoro precision of the stored codes. The default addresses each POI individually; lower it only if you want neighbours to share a code on purpose.	DEFAULT_PRECISION

Returns:

Type	Description
PoiExtractStats	Counts of what was written and walked past, plus unmapped — the
PoiExtractStats	most frequent tag values that matched nothing, as
PoiExtractStats	[(("amenity", "parking"), 1234), ...].

Raises:

Type	Description
ValueError	If source is empty — it selects the rows to replace, and an empty value would delete nothing while inserting unattributable rows.
ImportError	If the extract extra is not installed.

Source code in src/yoromaps/osm_poi.py

def extract_pois(
    pbf_path: str | Path,
    conn: sqlite3.Connection,
    progress: ProgressCallback | None = None,
    source: str = OSM_SOURCE,
    precision: int = DEFAULT_PRECISION,
) -> PoiExtractStats:
    """Extract POIs from a PBF into the database, replacing previous ones.

    Rows carrying *source* are deleted first, so a rebuild refreshes OSM data
    without touching POIs from anywhere else. The delete and every insert share
    one transaction: a failure mid-extraction leaves the previous set in place.

    Rows are written in batches of POI_INSERT_BATCH rather than accumulated,
    so peak memory does not grow with the size of the country.

    Nameless features are skipped — a pin with no label is noise on a map whose
    whole purpose is telling someone where to go.

    Args:
        pbf_path: Path to the `.osm.pbf` file.
        conn: Open connection to a `.yoromaps` database.
        progress: Optional callable(message, current, total).
        source: Value written to `pois.source`, and the rows replaced.
        precision: Yoro precision of the stored codes. The default addresses
            each POI individually; lower it only if you want neighbours to
            share a code on purpose.

    Returns:
        Counts of what was written and walked past, plus ``unmapped`` — the
        most frequent tag values that matched nothing, as
        ``[(("amenity", "parking"), 1234), ...]``.

    Raises:
        ValueError: If *source* is empty — it selects the rows to replace, and
            an empty value would delete nothing while inserting unattributable
            rows.
        ImportError: If the `extract` extra is not installed.
    """
    if not source:
        raise ValueError("source must name where these POIs come from")

    _check_osmium()

    country = get_metadata(conn, "country") or ""
    collector = _PoiCollector(conn, country, source, precision)

    if progress:
        progress("Scanning POIs...", 0, 2)

    try:
        conn.execute("DELETE FROM pois WHERE source = ?", (source,))
        # `locations=True` fills in way node coordinates, which is what lets a
        # hospital mapped as a building become a point.
        _poi_handler(collector).apply_file(str(pbf_path), locations=True)
        collector.flush()
    except BaseException:
        conn.rollback()
        raise
    conn.commit()

    if progress:
        progress(f"Stored {collector.written} POIs", 2, 2)

    return {
        "pois": collector.written,
        "nodes": collector.counts["nodes"],
        "ways": collector.counts["ways"],
        "skipped_unnamed": collector.counts["skipped_unnamed"],
        "unmapped": collector.unmapped.most_common(UNMAPPED_REPORTED),
    }

categorize(tags)

The category of an OSM feature, or None when it is not a POI.

categorize({"amenity": "pharmacy"}) 'pharmacy' categorize({"shop": "bakery"}) 'shop' categorize({"highway": "residential"}) is None True

Source code in src/yoromaps/osm_poi.py

def categorize(tags: dict[str, str]) -> str | None:
    """The category of an OSM feature, or None when it is not a POI.

    >>> categorize({"amenity": "pharmacy"})
    'pharmacy'
    >>> categorize({"shop": "bakery"})
    'shop'
    >>> categorize({"highway": "residential"}) is None
    True
    """
    for key, values in VALUE_CATEGORIES.items():
        category = values.get(tags.get(key, ""))
        if category:
            return category

    for key, category in KEY_CATEGORIES.items():
        value = tags.get(key)
        if value and value not in NEGATIVE_VALUES:
            return category

    return None

poi_name(tags)

The best available name, or an empty string.

Source code in src/yoromaps/osm_poi.py

def poi_name(tags: dict[str, str]) -> str:
    """The best available name, or an empty string."""
    for key in NAME_TAGS:
        name = (tags.get(key) or "").strip()
        if name:
            return name
    return ""

yoromaps.db

Database management for .yoromaps files.

yoromaps.db

Database management for .yoromaps files.

A .yoromaps file is a single SQLite database containing: - tiles: MBTiles-compatible tile storage - nodes: road graph intersections - edges: road graph segments - pois: points of interest - metadata: version, country, timestamps

SCHEMA_VERSION = 5 module-attribute

open_db(path, create=False)

Open a .yoromaps database. Creates schema if create is True.

Existing databases are upgraded transparently: the schema is idempotent (CREATE ... IF NOT EXISTS), so files built by older versions gain new tables and indexes on first open, and columns added to a table that already existed are applied by _add_late_columns.

Source code in src/yoromaps/db.py

def open_db(path: str | Path, create: bool = False) -> sqlite3.Connection:
    """Open a .yoromaps database. Creates schema if *create* is True.

    Existing databases are upgraded transparently: the schema is idempotent
    (``CREATE ... IF NOT EXISTS``), so files built by older versions gain new
    tables and indexes on first open, and columns added to a table that
    already existed are applied by `_add_late_columns`.
    """
    path = Path(path)
    exists = path.exists()

    if not exists and not create:
        raise FileNotFoundError(f"Database not found: {path}")

    conn = sqlite3.connect(str(path))
    conn.row_factory = sqlite3.Row
    conn.execute("PRAGMA journal_mode=WAL")
    conn.execute("PRAGMA synchronous=NORMAL")
    conn.execute("PRAGMA foreign_keys=ON")

    conn.executescript(SCHEMA_SQL)
    _add_late_columns(conn)
    stored = get_metadata(conn, "schema_version")
    if stored is None or int(stored) < SCHEMA_VERSION:
        conn.execute(
            "INSERT OR REPLACE INTO metadata VALUES ('schema_version', ?)",
            (str(SCHEMA_VERSION),),
        )
    conn.commit()

    return conn

db_config(path)

Return a Django DATABASES dict entry for a .yoromaps file.

Source code in src/yoromaps/db.py

def db_config(path: str) -> dict[str, str]:
    """Return a Django DATABASES dict entry for a .yoromaps file."""
    return {
        "ENGINE": "django.db.backends.sqlite3",
        "NAME": str(path),
    }

set_metadata(conn, key, value)

Source code in src/yoromaps/db.py

def set_metadata(conn: sqlite3.Connection, key: str, value: str) -> None:
    conn.execute("INSERT OR REPLACE INTO metadata VALUES (?, ?)", (key, value))
    conn.commit()

get_metadata(conn, key)

Source code in src/yoromaps/db.py

def get_metadata(conn: sqlite3.Connection, key: str) -> str | None:
    row = conn.execute("SELECT value FROM metadata WHERE key = ?", (key,)).fetchone()
    return row[0] if row else None

yoromaps.download

Download OSM data and build .yoromaps files.

yoromaps.download

Download OSM data and build/update a .yoromaps file for a country.

build(area_code, output, pbf_path=None, include_tiles=False, zoom_min=6, zoom_max=12, include_pois=True, progress=None)

Build a .yoromaps file for an area.

An area is a country that Geofabrik serves whole, or one region of a country it does not — France's extract is 4 830 MB and Alsace's is 124 MB, and only one of those is something to ask somebody to download.

The bundle records the country, not the area. Codes inside Alsace are French codes, addressed against France's box: an address does not change because somebody downloaded less of the map around it.

Parameters:

Name	Type	Description	Default
area_code	str	An area code — "ML" for a whole country, "FR:alsace" for one region of one.	required
output	str | Path	Output .yoromaps file path.	required
pbf_path	str | Path | None	Path to an existing PBF file. If None, downloads from Geofabrik.	None
include_tiles	bool	Also download map tiles (slow, ~200+ MB).	False
zoom_min	int	Min tile zoom level (if include_tiles).	6
zoom_max	int	Max tile zoom level (if include_tiles).	12
include_pois	bool	Also extract the shops, schools and hospitals OSM knows about. On by default — a bundle without them routes to places the user has no way to find.	True
progress	ProgressCallback | None	Optional callable(message, current, total).	None

Returns:

Type	Description
Path	Path to the created .yoromaps file.

Source code in src/yoromaps/download.py

def build(
    area_code: str,
    output: str | Path,
    pbf_path: str | Path | None = None,
    include_tiles: bool = False,
    zoom_min: int = 6,
    zoom_max: int = 12,
    include_pois: bool = True,
    progress: ProgressCallback | None = None,
) -> Path:
    """Build a .yoromaps file for an area.

    An area is a country that Geofabrik serves whole, or one region of a
    country it does not — France's extract is 4 830 MB and Alsace's is 124 MB,
    and only one of those is something to ask somebody to download.

    The bundle records the *country*, not the area. Codes inside Alsace are
    French codes, addressed against France's box: an address does not change
    because somebody downloaded less of the map around it.

    Args:
        area_code: An area code — "ML" for a whole country, "FR:alsace" for
            one region of one.
        output: Output .yoromaps file path.
        pbf_path: Path to an existing PBF file. If None, downloads from Geofabrik.
        include_tiles: Also download map tiles (slow, ~200+ MB).
        zoom_min: Min tile zoom level (if include_tiles).
        zoom_max: Max tile zoom level (if include_tiles).
        include_pois: Also extract the shops, schools and hospitals OSM knows
            about. On by default — a bundle without them routes to places the
            user has no way to find.
        progress: Optional callable(message, current, total).

    Returns:
        Path to the created .yoromaps file.
    """
    target = _area_or_refuse(area_code)
    output = Path(output)

    if pbf_path is None:
        tmp_dir = tempfile.mkdtemp(prefix="yoromaps_")
        downloaded = download_pbf(target.code, tmp_dir, progress=progress)
        if downloaded is None:
            # download_pbf only returns None when asked to skip an unchanged
            # file, which a fresh build never does. Refuse rather than carry a
            # None into the extraction and fail there with no explanation.
            raise RuntimeError(f"No PBF was downloaded for {target.code}")
        source_pbf = downloaded
    else:
        source_pbf = Path(pbf_path)

    conn = open_db(output, create=True)
    # `country` is what addresses the POIs, so it stays the country even when
    # the bundle covers one region of it. `area` says which piece this is.
    set_metadata(conn, "country", target.country)
    set_metadata(conn, "country_name", COUNTRIES[target.country].name)
    set_metadata(conn, "area", target.code)
    set_metadata(conn, "area_name", target.name)
    set_metadata(conn, "bbox", ",".join(str(v) for v in target.bbox))

    _build_graph(conn, source_pbf, progress)
    if include_pois:
        _build_pois(conn, source_pbf, progress)
    if include_tiles:
        _build_tiles(conn, target.bbox, zoom_min, zoom_max, progress)

    conn.close()
    return output

download_pbf(area_code, output_dir, progress=None, if_newer_than=None)

Download the latest OSM PBF for an area from Geofabrik.

Parameters:

Name	Type	Description	Default
area_code	str	An area code — "ML", or "FR:alsace" for one region.	required
output_dir	str | Path	Directory to save the PBF file.	required
progress	ProgressCallback | None	Optional progress callback.	None
if_newer_than	str | None	HTTP date string. Skips download if server file is not newer.	None

Returns:

Type	Description
Path | None	Path to the downloaded file, or None if skipped (already up to date).

Source code in src/yoromaps/download.py

def download_pbf(
    area_code: str,
    output_dir: str | Path,
    progress: ProgressCallback | None = None,
    if_newer_than: str | None = None,
) -> Path | None:
    """Download the latest OSM PBF for an area from Geofabrik.

    Args:
        area_code: An area code — "ML", or "FR:alsace" for one region.
        output_dir: Directory to save the PBF file.
        progress: Optional progress callback.
        if_newer_than: HTTP date string. Skips download if server file is not newer.

    Returns:
        Path to the downloaded file, or None if skipped (already up to date).
    """
    url = area_url(area_code)
    output_dir = Path(output_dir)
    output_dir.mkdir(parents=True, exist_ok=True)
    # A colon is legal in a filename and confusing in one; the area code is
    # spelled with a hyphen on disk and nowhere else.
    output_path = output_dir / f"{area_code.lower().replace(':', '-')}.osm.pbf"

    headers = {}
    if if_newer_than:
        headers["If-Modified-Since"] = if_newer_than

    if progress:
        progress(f"Checking {url}...", 0, 1)

    with httpx.stream("GET", url, follow_redirects=True, timeout=300, headers=headers) as resp:
        if resp.status_code == 304:
            if progress:
                progress("Already up to date", 1, 1)
            return None

        resp.raise_for_status()
        total = int(resp.headers.get("content-length", 0))
        downloaded = 0

        with open(output_path, "wb") as f:
            for chunk in resp.iter_bytes(chunk_size=65536):
                f.write(chunk)
                downloaded += len(chunk)
                if progress and total:
                    progress("Downloading PBF", downloaded, total)

    return output_path

yoromaps.tiles

MBTiles management — download and serve map tiles.

yoromaps.tiles

MBTiles management — download and serve map tiles from SQLite.

get_tile(conn, z, x, y)

Retrieve a tile from the database (XYZ scheme, converted to TMS internally).

Source code in src/yoromaps/tiles.py

def get_tile(conn: sqlite3.Connection, z: int, x: int, y: int) -> bytes | None:
    """Retrieve a tile from the database (XYZ scheme, converted to TMS internally)."""
    y_tms = (1 << z) - 1 - y
    row = conn.execute(
        "SELECT tile_data FROM tiles WHERE zoom_level=? AND tile_column=? AND tile_row=?",
        (z, x, y_tms),
    ).fetchone()
    return row[0] if row else None

tile_count(conn)

Source code in src/yoromaps/tiles.py

def tile_count(conn: sqlite3.Connection) -> int:
    row = conn.execute("SELECT COUNT(*) FROM tiles").fetchone()
    count: int = row[0]
    return count

download_tiles(conn, bbox, zoom_min=6, zoom_max=14, tile_url=DEFAULT_TILE_URL, progress=None)

Download map tiles for a bounding box into the database.

.. warning:: The default source is the public openstreetmap.org tile server, whose usage policy discourages bulk downloading — keep zoom levels low and prefer your own tile server or a commercial provider via tile_url for large areas.

Parameters:

Name	Type	Description	Default
conn	Connection	Open SQLite connection.	required
bbox	tuple[float, float, float, float]	(lon_min, lat_min, lon_max, lat_max).	required
zoom_min	int	Minimum zoom level.	6
zoom_max	int	Maximum zoom level.	14
tile_url	str	Tile URL template with {z}, {x}, {y} placeholders.	DEFAULT_TILE_URL
progress	ProgressCallback | None	Optional callable(message, current, total).	None

Returns:

Type	Description
int	Number of tiles present for the box afterwards — newly stored plus
int	already held. Tiles the server would not give up are excluded, so the
int	figure counts coverage rather than attempts.

Raises:

Type	Description
ValueError	If zoom_min exceeds zoom_max, or if not one tile could be fetched — a wrong URL template or a blocked address, which a zero return would have reported as a quiet success.

Source code in src/yoromaps/tiles.py

def download_tiles(
    conn: sqlite3.Connection,
    bbox: tuple[float, float, float, float],
    zoom_min: int = 6,
    zoom_max: int = 14,
    tile_url: str = DEFAULT_TILE_URL,
    progress: ProgressCallback | None = None,
) -> int:
    """Download map tiles for a bounding box into the database.

    .. warning::
        The default source is the public openstreetmap.org tile server,
        whose usage policy discourages bulk downloading — keep zoom levels
        low and prefer your own tile server or a commercial provider via
        *tile_url* for large areas.

    Args:
        conn: Open SQLite connection.
        bbox: (lon_min, lat_min, lon_max, lat_max).
        zoom_min: Minimum zoom level.
        zoom_max: Maximum zoom level.
        tile_url: Tile URL template with {z}, {x}, {y} placeholders.
        progress: Optional callable(message, current, total).

    Returns:
        Number of tiles present for the box afterwards — newly stored plus
        already held. Tiles the server would not give up are excluded, so the
        figure counts coverage rather than attempts.

    Raises:
        ValueError: If *zoom_min* exceeds *zoom_max*, or if not one tile could
            be fetched — a wrong URL template or a blocked address, which a
            zero return would have reported as a quiet success.
    """
    if zoom_min > zoom_max:
        raise ValueError(f"zoom_min ({zoom_min}) is above zoom_max ({zoom_max})")

    mercantile = _require_mercantile()
    _warn_on_bulk_osm(tile_url, zoom_max)

    conn.execute("INSERT OR REPLACE INTO metadata VALUES ('format', 'png')")

    by_zoom = {z: list(mercantile.tiles(*bbox, zooms=z)) for z in range(zoom_min, zoom_max + 1)}
    total = sum(len(tiles) for tiles in by_zoom.values())
    counters = {"seen": 0, "stored": 0, "kept": 0, "failed": 0}

    client = _tile_client()
    try:
        for tiles in by_zoom.values():
            _download_zoom_level(conn, client, tiles, tile_url, counters, total, progress)
    finally:
        client.close()

    if total and counters["failed"] == total:
        raise ValueError(
            f"All {total} tile requests failed. Check tile_url ({tile_url}) "
            f"and whether this address is allowed to fetch from it."
        )

    if progress:
        progress("Done", counters["seen"], total)

    return counters["stored"] + counters["kept"]

yoromaps.countries

Country registry with Geofabrik download URLs.

yoromaps.countries

Country registry — Geofabrik download paths.

Names and bounding boxes come from yoro.DOMAINS (the single source of truth for domain geometry shared by both packages); this module only adds the Geofabrik extract path for each country.

The table is generated by scripts/gen_countries.py from the index Geofabrik publishes, because there is no rule taking an ISO code to a file name: Geofabrik names its extracts after regions it decided on, and several countries have no file of their own. Those are served by the smallest extract that does contain them — the graph then carries the neighbours' roads and the bundle's bbox stays the country's, which is what Gambia has done inside the Senegal extract since the first version of this registry. It costs download size, not correctness.

COUNTRIES = {code: _country(code, path) for code, path in _GEOFABRIK_PATHS.items() if code in DOMAINS} module-attribute

GEOFABRIK_BASE = 'https://download.geofabrik.de' module-attribute

Country dataclass

Source code in src/yoromaps/countries.py

@dataclass(frozen=True, slots=True)
class Country:
    code: str
    name: str
    geofabrik_path: str
    bbox: tuple[float, float, float, float]  # lon_min, lat_min, lon_max, lat_max

geofabrik_url(country_code)

Return the Geofabrik PBF download URL for a country.

Source code in src/yoromaps/countries.py

def geofabrik_url(country_code: str) -> str:
    """Return the Geofabrik PBF download URL for a country."""
    c = COUNTRIES.get(country_code.upper())
    if not c:
        # Naming all two hundred and forty-nine would bury the one thing the
        # reader needs, which is that this code is not one of them.
        raise ValueError(
            f"Unknown country: {country_code}. "
            f"{len(COUNTRIES)} countries are known; read `yoromaps.countries.COUNTRIES`."
        )
    return f"{GEOFABRIK_BASE}/{c.geofabrik_path}-latest.osm.pbf"

yoromaps.extract

OSM PBF extraction into the road graph.

yoromaps.extract

Extract road graph from OSM PBF into a .yoromaps SQLite database.

Uses pyosmium to parse the PBF file in a single pass, extracting highway ways and their nodes into a graph suitable for routing.

ROUTABLE_HIGHWAYS = routable_highways() module-attribute

extract_graph(pbf_path, conn, progress=None)

Extract road graph from a PBF file into the database.

Reads the file twice: once for the ways that are roads and the nodes they reference, once for those nodes' coordinates. Edges are written in batches of EDGE_INSERT_BATCH so peak memory does not follow the country's size.

Parameters:

Name	Type	Description	Default
pbf_path	str | Path	Path to the .osm.pbf file.	required
conn	Connection	Open SQLite connection to a .yoromaps database.	required
progress	ProgressCallback | None	Optional callable(message, current, total) for progress reporting.	None

Returns:

Type	Description
GraphExtractStats	Counts of the nodes and road segments written.

Raises:

Type	Description
ImportError	If the extract extra is not installed.

Source code in src/yoromaps/extract.py

def extract_graph(
    pbf_path: str | Path,
    conn: sqlite3.Connection,
    progress: ProgressCallback | None = None,
) -> GraphExtractStats:
    """Extract road graph from a PBF file into the database.

    Reads the file twice: once for the ways that are roads and the nodes they
    reference, once for those nodes' coordinates. Edges are written in batches
    of EDGE_INSERT_BATCH so peak memory does not follow the country's size.

    Args:
        pbf_path: Path to the .osm.pbf file.
        conn: Open SQLite connection to a .yoromaps database.
        progress: Optional callable(message, current, total) for progress reporting.

    Returns:
        Counts of the nodes and road segments written.

    Raises:
        ImportError: If the `extract` extra is not installed.
    """
    _check_osmium()
    path = str(pbf_path)

    if progress:
        progress("Scanning ways...", 0, 2)
    collector = _way_collector()
    collector.apply_file(path)

    if progress:
        progress(
            f"Found {len(collector.ways_data)} roads, {len(collector.highway_node_ids)} nodes",
            1,
            2,
        )

    locator = _node_locator(collector.highway_node_ids)
    locator.apply_file(path, locations=True)

    if progress:
        progress("Building graph...", 2, 2)

    node_rows = [(nid, lat, lon) for nid, (lat, lon) in locator.coords.items()]
    conn.executemany("INSERT OR IGNORE INTO nodes (id, lat, lon) VALUES (?, ?, ?)", node_rows)
    edge_count = _insert_edges(conn, collector.ways_data, locator.coords)
    conn.commit()

    return {"nodes": len(node_rows), "edges": edge_count}

haversine(lat1, lon1, lat2, lon2)

Distance in meters between two GPS points.

Source code in src/yoromaps/extract.py

def haversine(lat1: float, lon1: float, lat2: float, lon2: float) -> float:
    """Distance in meters between two GPS points."""
    R = 6_371_000
    dlat = math.radians(lat2 - lat1)
    dlon = math.radians(lon2 - lon1)
    a = (
        math.sin(dlat / 2) ** 2
        + math.cos(math.radians(lat1)) * math.cos(math.radians(lat2)) * math.sin(dlon / 2) ** 2
    )
    return R * 2 * math.atan2(math.sqrt(a), math.sqrt(1 - a))

yoromaps.profiles

How fast each way of travelling goes, on each kind of road — and which roads it may not use at all. Read at query time, so a bundle gains a mode, or a corrected speed, without being rebuilt.

yoromaps.profiles

How fast each way of travelling goes, on each kind of road.

Until now the graph carried one duration per edge, computed when the bundle was built, from one speed table that described a car. Every other way of getting somewhere borrowed the car's answer: a bicycle was told it would take nine minutes to cross Bamako, and was routed down the trunk road to do it.

So the speed moved out of the file and into here, where it is read at query time. Two consequences worth stating:

• A bundle does not have to be rebuilt to gain a mode, or to correct one. The road type is already stored on every edge — the bundle always knew what kind of road it was, it just was not asked. Every number here is therefore a decision that can be revisited against a real map without republishing anything, and at least one of them already has been: see bicycle.
• A mode is not only a speed, it is a permission. None means this mode may not use that road at all: a bicycle on a motorway is not slow, it is forbidden, and a router that merely made it slow would still send somebody onto one to save four minutes.

The numbers are West African urban conditions, not European ones. A primary road in Bamako is not a primary road in Lyon: it is shared with handcarts, mopeds and pedestrians, and 70 km/h is what it does between the jams rather than an average anybody sustains. They are estimates, and they are wrong in the way every routing profile is wrong — the point is that they are wrong per mode, which is what the map has to be honest about.

A road is more than its type, and three of its tags change the answer:

• Surface. A primary in laterite is not a primary in asphalt, and outside the capitals most of them are laterite: 143,000 of Mali's 361,000 ways say unpaved. The surface is stored on the edge and caps the speed here, at query time, like everything else in this file.
• Access. access=private, motor_vehicle=no and the rest close a way to some modes and not others. Resolving OSM's tag hierarchy needs every tag of the way, so it is done once, at extraction, and the edge stores the modes it is closed to (edges.closed_to).
• One-way. It used to be baked into the graph's shape — a one-way was written as one row, so a pedestrian could not be allowed up it: there was nothing to allow. Both directions are now written, and the one against the flow is closed to the modes in ONEWAY_BOUND.

Known limit, requiring a change to the bundle rather than to this file:

• A bundle built before this file has no footways in it. Extraction only ever kept the road types a car could use, so walking routes on an old bundle follow the roads, not the paths beside them. New bundles keep them (see extract.ROUTABLE_HIGHWAYS); old ones stay correct, just coarser.

MODES = ('car', 'motorcycle', 'bicycle', 'foot') module-attribute

PROFILES = {'car': {'motorway': 110, 'motorway_link': 60, 'trunk': 90, 'trunk_link': 50, 'primary': 70, 'primary_link': 40, 'secondary': 60, 'secondary_link': 35, 'tertiary': 50, 'tertiary_link': 30, 'residential': 30, 'unclassified': 40, 'living_street': 20, 'service': 20, 'track': 15, 'path': None, 'footway': None, 'cycleway': None, 'pedestrian': None, 'steps': None}, 'motorcycle': {'motorway': 90, 'motorway_link': 55, 'trunk': 80, 'trunk_link': 45, 'primary': 60, 'primary_link': 38, 'secondary': 50, 'secondary_link': 32, 'tertiary': 45, 'tertiary_link': 28, 'residential': 28, 'unclassified': 35, 'living_street': 18, 'service': 18, 'track': 25, 'path': 15, 'cycleway': 18, 'footway': None, 'pedestrian': None, 'steps': None}, 'bicycle': {'motorway': None, 'motorway_link': None, 'trunk': 13, 'trunk_link': 11, 'primary': 14, 'primary_link': 12, 'secondary': 15, 'secondary_link': 12, 'tertiary': 15, 'tertiary_link': 12, 'residential': 14, 'unclassified': 13, 'living_street': 10, 'service': 10, 'track': 8, 'path': 8, 'footway': 6, 'cycleway': 16, 'pedestrian': 6, 'steps': 2}, 'foot': {'motorway': None, 'motorway_link': None, 'trunk': 4.2, 'trunk_link': 4.2, 'primary': 4.5, 'primary_link': 4.5, 'secondary': 4.5, 'secondary_link': 4.5, 'tertiary': 4.8, 'tertiary_link': 4.8, 'residential': 4.8, 'unclassified': 4.8, 'living_street': 4.8, 'service': 4.8, 'track': 4.2, 'path': 4.0, 'footway': 4.8, 'cycleway': 4.5, 'pedestrian': 4.8, 'steps': 1.5}} module-attribute

UNKNOWN_SPEED_KMH = {'car': 30, 'motorcycle': 28, 'bicycle': 12, 'foot': 4.5} module-attribute

UnknownMode

Bases: ValueError

Asked to route as something this library has no profile for.

A distinct type because the two failures are answered differently: an unknown mode is a caller's mistake and should be raised, while "no path for this mode" is an ordinary answer and comes back as found=False.

Source code in src/yoromaps/profiles.py

class UnknownMode(ValueError):
    """Asked to route as something this library has no profile for.

    A distinct type because the two failures are answered differently: an
    unknown mode is a caller's mistake and should be raised, while "no path
    for this mode" is an ordinary answer and comes back as `found=False`.
    """

speed_kmh(mode, road_type, surface='')

How fast mode travels on road_type, or None if it may not.

The one function the router asks. Raises for a mode that does not exist, because routing "as a helicopter" is not a routing failure to report back to a user — it is a call that should never have been made.

Source code in src/yoromaps/profiles.py

def speed_kmh(mode: str, road_type: str, surface: str = "") -> float | None:
    """How fast *mode* travels on *road_type*, or `None` if it may not.

    The one function the router asks. Raises for a mode that does not exist,
    because routing "as a helicopter" is not a routing failure to report back
    to a user — it is a call that should never have been made.
    """
    profile = PROFILES.get(mode)
    if profile is None:
        raise UnknownMode(f"No routing profile for mode {mode!r}. Known: {', '.join(MODES)}.")
    speed = profile[road_type] if road_type in profile else UNKNOWN_SPEED_KMH[mode]
    if speed is None:
        return None
    cap = SURFACE_CAPS_KMH.get(mode, {}).get(surface)
    return speed if cap is None else min(speed, cap)

top_speed_kmh(mode)

The fastest this mode ever goes, over every road it may use.

A* needs it. The heuristic has to be a lower bound on the time still to travel, or the search stops being optimal and starts returning whatever it reached first — so remaining distance is divided by the most optimistic speed the profile allows anywhere.

Source code in src/yoromaps/profiles.py

def top_speed_kmh(mode: str) -> float:
    """The fastest this mode ever goes, over every road it may use.

    A* needs it. The heuristic has to be a *lower* bound on the time still to
    travel, or the search stops being optimal and starts returning whatever it
    reached first — so remaining distance is divided by the most optimistic
    speed the profile allows anywhere.
    """
    profile = PROFILES.get(mode)
    if profile is None:
        raise UnknownMode(f"No routing profile for mode {mode!r}. Known: {', '.join(MODES)}.")
    speeds = [speed for speed in profile.values() if speed is not None]
    return max([*speeds, UNKNOWN_SPEED_KMH[mode]])

routable_highways()

Every road type some mode can use — what extraction has to keep.

Derived rather than written out again. The list used to be the car's speed table, which is why no bundle contains a footway: what a car could not drive was never extracted, so it could never be walked either.

Source code in src/yoromaps/profiles.py

def routable_highways() -> frozenset[str]:
    """Every road type some mode can use — what extraction has to keep.

    Derived rather than written out again. The list used to be the car's
    speed table, which is why no bundle contains a footway: what a car could
    not drive was never extracted, so it could never be walked either.
    """
    keep: set[str] = set()
    for profile in PROFILES.values():
        keep.update(road for road, speed in profile.items() if speed is not None)
    return frozenset(keep)