Skip to content

Core API Reference

yoromaps

Top-level module with convenience imports.

yoromaps

Yoro Maps — Offline maps, routing, and POI for Africa.

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"])

RouteResult dataclass

Result of a routing query.

Source code in src/yoromaps/routing.py 
34
35
36
37
38
39
40
41
42
43
44
45
@dataclass
class RouteResult:
    """Result of a routing query."""

    distance_km: float
    duration_min: float
    nodes: list[int]
    geometry: dict  # GeoJSON LineString
    steps: list[dict]
    found: bool = True
    from_snap_m: float = 0.0
    to_snap_m: float = 0.0

build(country_code, output, pbf_path=None, include_tiles=False, zoom_min=6, zoom_max=12, progress=None)

Build a .yoromaps file for a country.

Parameters:

Name Type Description Default
country_code str

ISO country code (e.g. "ML", "BF", "CI").

 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
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 
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
def build(
    country_code: str,
    output: str | Path,
    pbf_path: str | Path | None = None,
    include_tiles: bool = False,
    zoom_min: int = 6,
    zoom_max: int = 12,
    progress: ProgressCallback | None = None,
) -> Path:
    """Build a .yoromaps file for a country.

    Args:
        country_code: ISO country code (e.g. "ML", "BF", "CI").
        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).
        progress: Optional callable(message, current, total).

    Returns:
        Path to the created .yoromaps file.
    """
    country_code = country_code.upper()
    country = COUNTRIES[country_code]
    output = Path(output)

    # Download PBF if needed
    if pbf_path is None:
        tmp_dir = tempfile.mkdtemp(prefix="yoromaps_")
        pbf_path = download_pbf(country_code, tmp_dir, progress=progress)
    else:
        pbf_path = Path(pbf_path)

    # Create database
    conn = open_db(output, create=True)
    set_metadata(conn, "country", country_code)
    set_metadata(conn, "country_name", country.name)
    set_metadata(conn, "bbox", ",".join(str(v) for v in country.bbox))

    # Extract road graph
    if progress:
        progress("Extracting road graph...", 0, 1)
    stats = extract_graph(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, "osm_update_date", now)

    if progress:
        progress(f"Graph: {stats['nodes']} nodes, {stats['edges']} edges", 1, 1)

    # Download tiles if requested
    if include_tiles:
        from yoromaps.tiles import download_tiles
        if progress:
            progress("Downloading tiles...", 0, 1)
        n = download_tiles(conn, country.bbox, zoom_min=zoom_min, zoom_max=zoom_max, progress=progress)
        set_metadata(conn, "tiles_count", str(n))

    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 while preserving local POI data.

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
dict

Dict with updated (bool), nodes, edges, previous_date, new_date.
Source code in src/yoromaps/download.py 
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
def update(
    db_path: str | Path,
    pbf_cache_dir: str | Path | None = None,
    progress: ProgressCallback | None = None,
) -> dict:
    """Update an existing .yoromaps file with the latest OSM data.

    Downloads the PBF only if Geofabrik has a newer version.
    Rebuilds the road graph while preserving local POI data.

    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)

    country_code = get_metadata(conn, "country")
    if not country_code:
        conn.close()
        raise ValueError(f"No country 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 ({country_code})...", 0, 3)

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

    if pbf_path is None:
        nodes = int(get_metadata(conn, "graph_nodes") or 0)
        edges = int(get_metadata(conn, "graph_edges") or 0)
        conn.close()
        return {
            "updated": False,
            "nodes": nodes,
            "edges": edges,
            "previous_date": last_update,
            "new_date": last_update,
        }

    # Replace the graph atomically: the DELETE is only committed together
    # with the freshly extracted graph (extract_graph commits at the end).
    # A failure mid-extraction rolls back and keeps the previous graph.
    if progress:
        progress("Clearing old graph...", 1, 3)
    try:
        conn.execute("DELETE FROM edges")
        conn.execute("DELETE FROM nodes")

        if progress:
            progress("Extracting new graph...", 2, 3)
        stats = extract_graph(str(pbf_path), conn, progress=progress)
    except BaseException:
        conn.rollback()
        conn.close()
        raise

    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, "osm_update_date", now)

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

    conn.close()

    return {
        "updated": True,
        "nodes": stats["nodes"],
        "edges": stats["edges"],
        "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 fully idempotent (CREATE ... IF NOT EXISTS), so files built by older versions gain new tables/indexes on first open.

Source code in src/yoromaps/db.py 
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
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 fully
    idempotent (``CREATE ... IF NOT EXISTS``), so files built by older
    versions gain new tables/indexes on first open.
    """
    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)
    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 
116
117
118
119
120
121
def db_config(path: str) -> dict:
    """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)

Find the shortest 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 
339
340
341
342
343
344
345
346
347
348
349
350
351
352
def route(
    conn: sqlite3.Connection,
    start_lat: float,
    start_lon: float,
    end_lat: float,
    end_lon: float,
) -> RouteResult:
    """Find the shortest 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)

route_from_codes(conn, codes, graph=None)

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.

Source code in src/yoromaps/routing.py 
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
def route_from_codes(
    conn: sqlite3.Connection,
    codes: list[str],
    graph: Graph | None = None,
) -> 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.
    """
    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])
        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 
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
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)

Insert a POI; its Yoro code is computed automatically.

Source code in src/yoromaps/poi.py 
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
def add_poi(
    conn: sqlite3.Connection,
    lat: float,
    lon: float,
    name: str,
    category: str = "other",
    source: str = "local",
    osm_id: int | None = None,
) -> dict:
    """Insert a POI; its Yoro code is computed automatically."""
    code = _encode_for_db(conn, lat, lon)
    cur = conn.execute(
        "INSERT INTO pois (lat, lon, yoro_code, name, category, source, osm_id) "
        "VALUES (?, ?, ?, ?, ?, ?, ?)",
        (lat, lon, code, name, category, source, osm_id),
    )
    conn.commit()
    return {
        "id": cur.lastrowid,
        "lat": lat,
        "lon": lon,
        "yoro_code": code,
        "name": name,
        "category": category,
        "source": source,
    }

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 
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
def pois_near(
    conn: sqlite3.Connection,
    lat: float,
    lon: float,
    radius_m: float = 1000,
    category: str | None = None,
    limit: int = 100,
) -> list[dict]:
    """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 = [lat - dlat, lat + dlat, lon - dlon, lon + dlon]
    if category:
        sql += " AND category = ?"
        params.append(category)

    hits = []
    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 
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
def pois_in_cell(
    conn: sqlite3.Connection,
    code: str,
    category: str | None = None,
) -> list[dict]:
    """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 = [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 
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
def search_pois(
    conn: sqlite3.Connection,
    query: str | None = None,
    category: str | None = None,
    limit: int = 50,
) -> list[dict]:
    """Search POIs by name substring and/or category."""
    sql = "SELECT * FROM pois WHERE 1=1"
    params: list = []
    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 
151
152
153
154
155
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.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.

RouteResult dataclass

Result of a routing query.

Source code in src/yoromaps/routing.py 
34
35
36
37
38
39
40
41
42
43
44
45
@dataclass
class RouteResult:
    """Result of a routing query."""

    distance_km: float
    duration_min: float
    nodes: list[int]
    geometry: dict  # GeoJSON LineString
    steps: list[dict]
    found: bool = True
    from_snap_m: float = 0.0
    to_snap_m: float = 0.0

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 
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
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):
        self.coords: dict[int, tuple[float, float]] = {}  # node_id → (lat, lon)
        self.adj: dict[int, list[tuple[int, float, float, str]]] = {}  # node_id → [(to, dist_m, dur_s, name)]
        self._grid: dict[tuple[int, int], list[int]] | None = None
        self._grid_extent: tuple[int, int, int, int] | None = None  # i_min, i_max, j_min, j_max

    @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, duration_s, name FROM edges"):
            from_n, to_n, dist, dur, name = row
            if from_n not in g.adj:
                g.adj[from_n] = []
            g.adj[from_n].append((to_n, dist, dur, 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 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)

        best_id, best_sq = None, float("inf")
        # Upper bound on rings: enough to reach the farthest grid cell.
        i_min, i_max, j_min, j_max = self._grid_extent
        max_ring = max(
            abs(ci - i_min), abs(ci - i_max), abs(cj - j_min), abs(cj - j_max)
        ) + 1

        grid = self._grid
        coords = self.coords

        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

            # Skip rings whose square lies entirely outside the grid extent
            if (ci + ring < i_min or ci - ring > i_max
                    or cj + ring < j_min or cj - ring > j_max):
                continue

            # Walk only the ring's perimeter, clipped to the extent — O(ring),
            # not O(ring²) (a full-square filter made far-away queries O(r³)).
            cells = []
            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 ring:
                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))

            for key in cells:
                for nid in grid.get(key, ()):
                    nlat, nlon = coords[nid]
                    d = (nlat - lat) ** 2 + (nlon - lon) ** 2
                    if d < best_sq:
                        best_sq = d
                        best_id = nid

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

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

        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)

        empty = 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,
        )

        if not start_node or not end_node:
            return empty

        if start_node == end_node:
            c = self.coords[start_node]
            return RouteResult(
                distance_km=0, duration_min=0, nodes=[start_node],
                geometry={"type": "LineString", "coordinates": [[c[1], c[0]]]},
                steps=[], found=True,
                from_snap_m=from_snap_m, to_snap_m=to_snap_m,
            )

        end_lat_n, end_lon_n = self.coords[end_node]

        # A* algorithm — in-memory, fast
        open_set: list[tuple[float, int]] = [(0.0, start_node)]
        came_from: dict[int, int] = {}
        g_score: dict[int, float] = {start_node: 0.0}
        duration: dict[int, float] = {start_node: 0.0}
        visited: set[int] = set()

        while open_set:
            _, current = heapq.heappop(open_set)

            if current == end_node:
                path = [current]
                while current in came_from:
                    current = came_from[current]
                    path.append(current)
                path.reverse()

                return RouteResult(
                    distance_km=round(g_score[end_node] / 1000, 1),
                    duration_min=round(duration[end_node] / 60, 1),
                    nodes=path,
                    geometry=self._build_geometry(path),
                    steps=self._build_steps(path),
                    found=True,
                    from_snap_m=from_snap_m, to_snap_m=to_snap_m,
                )

            if current in visited:
                continue
            visited.add(current)

            for to_node, dist_m, dur_s, _ in self.adj.get(current, []):
                if to_node in visited:
                    continue

                tentative_g = g_score[current] + dist_m
                if tentative_g < g_score.get(to_node, float("inf")):
                    came_from[to_node] = current
                    g_score[to_node] = tentative_g
                    duration[to_node] = duration[current] + dur_s

                    nlat, nlon = self.coords.get(to_node, (0, 0))
                    h = _haversine(nlat, nlon, end_lat_n, end_lon_n)
                    heapq.heappush(open_set, (tentative_g + h, to_node))

        return empty

    def _build_geometry(self, node_ids: list[int]) -> dict:
        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]:
        steps: list[dict] = []
        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 
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
@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, duration_s, name FROM edges"):
        from_n, to_n, dist, dur, name = row
        if from_n not in g.adj:
            g.adj[from_n] = []
        g.adj[from_n].append((to_n, dist, dur, 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 
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
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)

    best_id, best_sq = None, float("inf")
    # Upper bound on rings: enough to reach the farthest grid cell.
    i_min, i_max, j_min, j_max = self._grid_extent
    max_ring = max(
        abs(ci - i_min), abs(ci - i_max), abs(cj - j_min), abs(cj - j_max)
    ) + 1

    grid = self._grid
    coords = self.coords

    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

        # Skip rings whose square lies entirely outside the grid extent
        if (ci + ring < i_min or ci - ring > i_max
                or cj + ring < j_min or cj - ring > j_max):
            continue

        # Walk only the ring's perimeter, clipped to the extent — O(ring),
        # not O(ring²) (a full-square filter made far-away queries O(r³)).
        cells = []
        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 ring:
            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))

        for key in cells:
            for nid in grid.get(key, ()):
                nlat, nlon = coords[nid]
                d = (nlat - lat) ** 2 + (nlon - lon) ** 2
                if d < best_sq:
                    best_sq = d
                    best_id = nid

    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)

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

Source code in src/yoromaps/routing.py 
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
def route(
    self,
    start_lat: float,
    start_lon: float,
    end_lat: float,
    end_lon: float,
) -> RouteResult:
    """Find the shortest route between two GPS coordinates using A*."""

    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)

    empty = 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,
    )

    if not start_node or not end_node:
        return empty

    if start_node == end_node:
        c = self.coords[start_node]
        return RouteResult(
            distance_km=0, duration_min=0, nodes=[start_node],
            geometry={"type": "LineString", "coordinates": [[c[1], c[0]]]},
            steps=[], found=True,
            from_snap_m=from_snap_m, to_snap_m=to_snap_m,
        )

    end_lat_n, end_lon_n = self.coords[end_node]

    # A* algorithm — in-memory, fast
    open_set: list[tuple[float, int]] = [(0.0, start_node)]
    came_from: dict[int, int] = {}
    g_score: dict[int, float] = {start_node: 0.0}
    duration: dict[int, float] = {start_node: 0.0}
    visited: set[int] = set()

    while open_set:
        _, current = heapq.heappop(open_set)

        if current == end_node:
            path = [current]
            while current in came_from:
                current = came_from[current]
                path.append(current)
            path.reverse()

            return RouteResult(
                distance_km=round(g_score[end_node] / 1000, 1),
                duration_min=round(duration[end_node] / 60, 1),
                nodes=path,
                geometry=self._build_geometry(path),
                steps=self._build_steps(path),
                found=True,
                from_snap_m=from_snap_m, to_snap_m=to_snap_m,
            )

        if current in visited:
            continue
        visited.add(current)

        for to_node, dist_m, dur_s, _ in self.adj.get(current, []):
            if to_node in visited:
                continue

            tentative_g = g_score[current] + dist_m
            if tentative_g < g_score.get(to_node, float("inf")):
                came_from[to_node] = current
                g_score[to_node] = tentative_g
                duration[to_node] = duration[current] + dur_s

                nlat, nlon = self.coords.get(to_node, (0, 0))
                h = _haversine(nlat, nlon, end_lat_n, end_lon_n)
                heapq.heappush(open_set, (tentative_g + h, to_node))

    return empty

route(conn, start_lat, start_lon, end_lat, end_lon)

Find the shortest 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 
339
340
341
342
343
344
345
346
347
348
349
350
351
352
def route(
    conn: sqlite3.Connection,
    start_lat: float,
    start_lon: float,
    end_lat: float,
    end_lon: float,
) -> RouteResult:
    """Find the shortest 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)

route_from_codes(conn, codes, graph=None)

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.

Source code in src/yoromaps/routing.py 
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
def route_from_codes(
    conn: sqlite3.Connection,
    codes: list[str],
    graph: Graph | None = None,
) -> 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.
    """
    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])
        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 
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
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)

Insert a POI; its Yoro code is computed automatically.

Source code in src/yoromaps/poi.py 
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
def add_poi(
    conn: sqlite3.Connection,
    lat: float,
    lon: float,
    name: str,
    category: str = "other",
    source: str = "local",
    osm_id: int | None = None,
) -> dict:
    """Insert a POI; its Yoro code is computed automatically."""
    code = _encode_for_db(conn, lat, lon)
    cur = conn.execute(
        "INSERT INTO pois (lat, lon, yoro_code, name, category, source, osm_id) "
        "VALUES (?, ?, ?, ?, ?, ?, ?)",
        (lat, lon, code, name, category, source, osm_id),
    )
    conn.commit()
    return {
        "id": cur.lastrowid,
        "lat": lat,
        "lon": lon,
        "yoro_code": code,
        "name": name,
        "category": category,
        "source": source,
    }

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 
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
def pois_near(
    conn: sqlite3.Connection,
    lat: float,
    lon: float,
    radius_m: float = 1000,
    category: str | None = None,
    limit: int = 100,
) -> list[dict]:
    """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 = [lat - dlat, lat + dlat, lon - dlon, lon + dlon]
    if category:
        sql += " AND category = ?"
        params.append(category)

    hits = []
    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 
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
def pois_in_cell(
    conn: sqlite3.Connection,
    code: str,
    category: str | None = None,
) -> list[dict]:
    """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 = [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 
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
def search_pois(
    conn: sqlite3.Connection,
    query: str | None = None,
    category: str | None = None,
    limit: int = 50,
) -> list[dict]:
    """Search POIs by name substring and/or category."""
    sql = "SELECT * FROM pois WHERE 1=1"
    params: list = []
    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 
151
152
153
154
155
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.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 = 2 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 fully idempotent (CREATE ... IF NOT EXISTS), so files built by older versions gain new tables/indexes on first open.

Source code in src/yoromaps/db.py 
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
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 fully
    idempotent (``CREATE ... IF NOT EXISTS``), so files built by older
    versions gain new tables/indexes on first open.
    """
    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)
    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 
116
117
118
119
120
121
def db_config(path: str) -> dict:
    """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 
106
107
108
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 
111
112
113
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(country_code, output, pbf_path=None, include_tiles=False, zoom_min=6, zoom_max=12, progress=None)

Build a .yoromaps file for a country.

Parameters:

Name Type Description Default
country_code str

ISO country code (e.g. "ML", "BF", "CI").

 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
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 
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
def build(
    country_code: str,
    output: str | Path,
    pbf_path: str | Path | None = None,
    include_tiles: bool = False,
    zoom_min: int = 6,
    zoom_max: int = 12,
    progress: ProgressCallback | None = None,
) -> Path:
    """Build a .yoromaps file for a country.

    Args:
        country_code: ISO country code (e.g. "ML", "BF", "CI").
        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).
        progress: Optional callable(message, current, total).

    Returns:
        Path to the created .yoromaps file.
    """
    country_code = country_code.upper()
    country = COUNTRIES[country_code]
    output = Path(output)

    # Download PBF if needed
    if pbf_path is None:
        tmp_dir = tempfile.mkdtemp(prefix="yoromaps_")
        pbf_path = download_pbf(country_code, tmp_dir, progress=progress)
    else:
        pbf_path = Path(pbf_path)

    # Create database
    conn = open_db(output, create=True)
    set_metadata(conn, "country", country_code)
    set_metadata(conn, "country_name", country.name)
    set_metadata(conn, "bbox", ",".join(str(v) for v in country.bbox))

    # Extract road graph
    if progress:
        progress("Extracting road graph...", 0, 1)
    stats = extract_graph(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, "osm_update_date", now)

    if progress:
        progress(f"Graph: {stats['nodes']} nodes, {stats['edges']} edges", 1, 1)

    # Download tiles if requested
    if include_tiles:
        from yoromaps.tiles import download_tiles
        if progress:
            progress("Downloading tiles...", 0, 1)
        n = download_tiles(conn, country.bbox, zoom_min=zoom_min, zoom_max=zoom_max, progress=progress)
        set_metadata(conn, "tiles_count", str(n))

    conn.close()
    return output

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

Download the latest OSM PBF for a country from Geofabrik.

Parameters:

Name Type Description Default
country_code str

ISO country code.

 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 
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
def download_pbf(
    country_code: str,
    output_dir: str | Path,
    progress: ProgressCallback | None = None,
    if_newer_than: str | None = None,
) -> Path | None:
    """Download the latest OSM PBF for a country from Geofabrik.

    Args:
        country_code: ISO country code.
        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 = geofabrik_url(country_code)
    output_dir = Path(output_dir)
    output_dir.mkdir(parents=True, exist_ok=True)
    output_path = output_dir / f"{country_code.lower()}.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 
18
19
20
21
22
23
24
25
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 
28
29
30
def tile_count(conn: sqlite3.Connection) -> int:
    row = conn.execute("SELECT COUNT(*) FROM tiles").fetchone()
    return row[0]

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 (the default stops at 12) 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 downloaded.
Source code in src/yoromaps/tiles.py 
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
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 (the default stops at 12) 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 downloaded.
    """
    try:
        import mercantile
    except ImportError:
        raise ImportError("Install tiles extra: pip install yoro-maps[tiles]")

    if tile_url == DEFAULT_TILE_URL and zoom_max > 12:
        warnings.warn(
            "Bulk-downloading osm.org tiles above zoom 12 conflicts with the "
            "OSM tile usage policy; use your own tile server via tile_url.",
            stacklevel=2,
        )

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

    total_tiles = 0
    for z in range(zoom_min, zoom_max + 1):
        total_tiles += len(list(mercantile.tiles(*bbox, zooms=z)))

    downloaded = 0
    from yoromaps import __version__

    client = httpx.Client(
        timeout=15,
        headers={"User-Agent": f"yoromaps/{__version__} (https://github.com/Altius-Academy-SNC/yoro-maps)"},
        follow_redirects=True,
    )

    try:
        for z in range(zoom_min, zoom_max + 1):
            tiles = list(mercantile.tiles(*bbox, zooms=z))
            for t in tiles:
                # Skip if already exists
                y_tms = (1 << t.z) - 1 - t.y
                existing = conn.execute(
                    "SELECT 1 FROM tiles WHERE zoom_level=? AND tile_column=? AND tile_row=?",
                    (t.z, t.x, y_tms),
                ).fetchone()
                if existing:
                    downloaded += 1
                    continue

                url = tile_url.format(z=t.z, x=t.x, y=t.y)
                try:
                    resp = client.get(url)
                    if resp.status_code == 200:
                        conn.execute(
                            "INSERT OR REPLACE INTO tiles VALUES (?, ?, ?, ?)",
                            (t.z, t.x, y_tms, resp.content),
                        )
                except httpx.HTTPError:
                    pass

                downloaded += 1
                if progress and downloaded % 50 == 0:
                    progress(f"z={z}", downloaded, total_tiles)

                # Rate limiting
                time.sleep(0.05)

            conn.commit()
    finally:
        client.close()

    if progress:
        progress("Done", downloaded, total_tiles)

    return downloaded

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.

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

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

Country dataclass

Source code in src/yoromaps/countries.py 
16
17
18
19
20
21
@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 
73
74
75
76
77
78
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:
        raise ValueError(f"Unknown country: {country_code}. Available: {list(COUNTRIES.keys())}")
    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.

HIGHWAY_SPEEDS = {'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': 5} module-attribute

extract_graph(pbf_path, conn, progress=None)

Extract road graph from a PBF file into the database.

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
dict

Dict with keys: nodes, edges.
Source code in src/yoromaps/extract.py 
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
def extract_graph(
    pbf_path: str | Path,
    conn: sqlite3.Connection,
    progress: ProgressCallback | None = None,
) -> dict:
    """Extract road graph from a PBF file into the database.

    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:
        Dict with keys: ``nodes``, ``edges``.
    """
    _check_osmium()

    class _NodeCollector(osmium.SimpleHandler):
        def __init__(self):
            super().__init__()
            self.highway_node_ids: set[int] = set()
            self.ways_data: list[tuple] = []

        def way(self, w):
            tags = dict(w.tags)
            hw = tags.get("highway")
            if hw not in HIGHWAY_SPEEDS:
                return
            node_ids = [n.ref for n in w.nodes]
            if len(node_ids) < 2:
                return
            oneway = tags.get("oneway", "no") in ("yes", "true", "1")
            name = tags.get("name", "")
            self.ways_data.append((node_ids, hw, oneway, name))
            self.highway_node_ids.update(node_ids)

    class _NodeLocator(osmium.SimpleHandler):
        def __init__(self, wanted_ids: set[int]):
            super().__init__()
            self.wanted = wanted_ids
            self.coords: dict[int, tuple[float, float]] = {}

        def node(self, n):
            if n.id in self.wanted:
                self.coords[n.id] = (n.location.lat, n.location.lon)

    pbf_path = str(pbf_path)

    # Pass 1: collect ways and their node references
    if progress:
        progress("Scanning ways...", 0, 2)
    collector = _NodeCollector()
    collector.apply_file(pbf_path)

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

    # Pass 2: collect coordinates for relevant nodes
    locator = _NodeLocator(collector.highway_node_ids)
    locator.apply_file(pbf_path, locations=True)

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

    # Insert nodes
    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)

    # Insert edges
    edge_count = 0
    edge_rows = []
    for node_ids, hw_type, oneway, name in collector.ways_data:
        speed_kmh = HIGHWAY_SPEEDS[hw_type]
        speed_ms = speed_kmh / 3.6

        for i in range(len(node_ids) - 1):
            n1, n2 = node_ids[i], node_ids[i + 1]
            c1 = locator.coords.get(n1)
            c2 = locator.coords.get(n2)
            if not c1 or not c2:
                continue

            dist = haversine(c1[0], c1[1], c2[0], c2[1])
            duration = dist / speed_ms if speed_ms > 0 else 0

            # Forward edge
            edge_rows.append((n1, n2, dist, duration, hw_type, 1 if oneway else 0, name))

            # Reverse edge (if not oneway)
            if not oneway:
                edge_rows.append((n2, n1, dist, duration, hw_type, 0, name))

            edge_count += 1

    conn.executemany(
        "INSERT INTO edges (from_node, to_node, distance_m, duration_s, road_type, oneway, name) "
        "VALUES (?, ?, ?, ?, ?, ?, ?)",
        edge_rows,
    )
    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 
36
37
38
39
40
41
42
43
44
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))