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
    from_snap_m: float = 0.0
    to_snap_m: float = 0.0
    #: 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.

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.

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
    from_snap_m: float = 0.0
    to_snap_m: float = 0.0
    #: 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, road_type, name)]
        #
        # The road type, 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, str, 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])

        for row in conn.execute(
            "SELECT from_node, to_node, distance_m, road_type, name FROM edges"
        ):
            from_n, to_n, dist, road_type, name = row
            if from_n not in g.adj:
                g.adj[from_n] = []
            # Interned: a country has millions of edges and about twenty road
            # types between them. Without this the same short string is held
            # a million times over, for tens of megabytes of nothing.
            g.adj[from_n].append((to_n, dist, sys.intern(road_type or "road"), 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,
    ) -> tuple[int | None, float]:
        """Best node among *cells*, in squared degree distance."""
        assert self._grid is not None
        for key in cells:
            for nid in self._grid.get(key, ()):
                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) -> 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.
        """
        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)

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

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

    def _same_node_result(
        self, node: int, from_snap_m: float, to_snap_m: float, mode: str
    ) -> RouteResult:
        lat, lon = self.coords[node]
        return RouteResult(
            distance_km=0,
            duration_min=0,
            nodes=[node],
            geometry={"type": "LineString", "coordinates": [[lon, lat]]},
            steps=[],
            found=True,
            from_snap_m=from_snap_m,
            to_snap_m=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) -> _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.
        """
        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},
            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.cost_s[end_node],
                )

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

        return None

    def _relax(
        self,
        current: int,
        frontier: _Frontier,
        mode: str,
        top_ms: float,
        goal: tuple[float, float],
    ) -> None:
        """Relax every edge leaving *current* that this mode may travel."""
        end_lat, end_lon = goal
        for to_node, dist_m, road_type, _ in self.adj.get(current, []):
            if to_node in frontier.visited:
                continue
            speed = speed_kmh(mode, road_type)
            # 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
            tentative = frontier.cost_s[current] + dist_m / (speed * _KMH_TO_MS)
            if tentative >= frontier.cost_s.get(to_node, float("inf")):
                continue
            frontier.came_from[to_node] = current
            frontier.cost_s[to_node] = tentative
            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.
        """
        start_node, from_snap_m = self.nearest_with_distance(start_lat, start_lon)
        end_node, to_snap_m = self.nearest_with_distance(end_lat, end_lon)

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

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

        search = self._astar(start_node, end_node, mode)
        if search is None:
            return self._empty_result(from_snap_m, to_snap_m, mode)

        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),
            steps=self._build_steps(search.path),
            found=True,
            from_snap_m=from_snap_m,
            to_snap_m=to_snap_m,
            mode=mode,
        )

    def _build_geometry(self, node_ids: list[int]) -> GeoJson:
        coords = []
        for nid in node_ids:
            c = self.coords.get(nid)
            if c:
                coords.append([c[1], c[0]])  # [lon, lat]
        return {"type": "LineString", "coordinates": coords}

    def _build_steps(self, node_ids: list[int]) -> list[dict[str, object]]:
        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])

    for row in conn.execute(
        "SELECT from_node, to_node, distance_m, road_type, name FROM edges"
    ):
        from_n, to_n, dist, road_type, name = row
        if from_n not in g.adj:
            g.adj[from_n] = []
        # Interned: a country has millions of edges and about twenty road
        # types between them. Without this the same short string is held
        # a million times over, for tens of megabytes of nothing.
        g.adj[from_n].append((to_n, dist, sys.intern(road_type or "road"), name or ""))

    return g

nearest_with_distance(lat, lon)

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.

Source code in src/yoromaps/routing.py

def nearest_with_distance(self, lat: float, lon: float) -> 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.
    """
    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)

    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.

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.
    """
    start_node, from_snap_m = self.nearest_with_distance(start_lat, start_lon)
    end_node, to_snap_m = self.nearest_with_distance(end_lat, end_lon)

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

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

    search = self._astar(start_node, end_node, mode)
    if search is None:
        return self._empty_result(from_snap_m, to_snap_m, mode)

    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),
        steps=self._build_steps(search.path),
        found=True,
        from_snap_m=from_snap_m,
        to_snap_m=to_snap_m,
        mode=mode,
    )

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

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 = 4 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.

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

• One-way streets are baked into the graph's shape. A two-way road is written as two rows, a one-way as one, so the reverse edge simply does not exist. A pedestrian cannot be allowed up a one-way street from here — there is nothing to allow. Fixing that means always writing both directions and letting the router read the oneway flag per mode.
• 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)

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) -> 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)}.")
    if road_type in profile:
        return profile[road_type]
    return UNKNOWN_SPEED_KMH[mode]

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)