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 
26
27
28
29
30
31
32
33
34
35
@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

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

Optional callable(message, current, total).

 None

Returns:

Type Description
Path

Path to the created .yoromaps file.
Source code in src/yoromaps/download.py 
 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
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=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

open_db(path, create=False)

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

Source code in src/yoromaps/db.py 
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
def open_db(path: str | Path, create: bool = False) -> sqlite3.Connection:
    """Open a .yoromaps database. Creates schema if *create* is True."""
    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")

    if create and not exists:
        conn.executescript(SCHEMA_SQL)
        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 
107
108
109
110
111
112
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 
228
229
230
231
232
233
234
235
236
237
238
239
240
241
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)

Route through multiple Yoro codes in order.

Returns a list of RouteResult, one per leg.

Source code in src/yoromaps/routing.py 
244
245
246
247
248
249
250
251
252
253
254
255
256
def route_from_codes(conn: sqlite3.Connection, codes: list[str]) -> list[RouteResult]:
    """Route through multiple Yoro codes in order.

    Returns a list of RouteResult, one per leg.
    """
    import yoro

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

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 
26
27
28
29
30
31
32
33
34
35
@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

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 
228
229
230
231
232
233
234
235
236
237
238
239
240
241
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)

Route through multiple Yoro codes in order.

Returns a list of RouteResult, one per leg.

Source code in src/yoromaps/routing.py 
244
245
246
247
248
249
250
251
252
253
254
255
256
def route_from_codes(conn: sqlite3.Connection, codes: list[str]) -> list[RouteResult]:
    """Route through multiple Yoro codes in order.

    Returns a list of RouteResult, one per leg.
    """
    import yoro

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

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 = 1 module-attribute

open_db(path, create=False)

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

Source code in src/yoromaps/db.py 
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
def open_db(path: str | Path, create: bool = False) -> sqlite3.Connection:
    """Open a .yoromaps database. Creates schema if *create* is True."""
    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")

    if create and not exists:
        conn.executescript(SCHEMA_SQL)
        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 
107
108
109
110
111
112
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 
97
98
99
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 
102
103
104
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

Optional callable(message, current, total).

 None

Returns:

Type Description
Path

Path to the created .yoromaps file.
Source code in src/yoromaps/download.py 
 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
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=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

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 
19
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
def download_pbf(
    country_code: str,
    output_dir: str | Path,
    progress=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 
13
14
15
16
17
18
19
20
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 
23
24
25
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='https://tile.openstreetmap.org/{z}/{x}/{y}.png', progress=None)

Download map tiles for a bounding box into the database.

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.

 'https://tile.openstreetmap.org/{z}/{x}/{y}.png'
progress

Optional callable(message, current, total).

 None

Returns:

Type Description
int

Number of tiles downloaded.
Source code in src/yoromaps/tiles.py 
 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
 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
def download_tiles(
    conn: sqlite3.Connection,
    bbox: tuple[float, float, float, float],
    zoom_min: int = 6,
    zoom_max: int = 14,
    tile_url: str = "https://tile.openstreetmap.org/{z}/{x}/{y}.png",
    progress=None,
) -> int:
    """Download map tiles for a bounding box into the database.

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

    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
    client = httpx.Client(
        timeout=15,
        headers={"User-Agent": "yoromaps/0.1 (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 URLs and metadata.

Focused on Africa, extensible to other regions.

COUNTRIES = {'ML': Country('ML', 'Mali', 'africa/mali', (-12.5, 10.0, 4.5, 25.0)), 'NE': Country('NE', 'Niger', 'africa/niger', (0.0, 11.5, 16.0, 24.0)), 'BF': Country('BF', 'Burkina Faso', 'africa/burkina-faso', (-5.5, 9.0, 2.5, 15.5)), 'CI': Country('CI', "Cote d'Ivoire", 'africa/ivory-coast', (-8.6, 4.36, -2.49, 10.74)), 'GH': Country('GH', 'Ghana', 'africa/ghana', (-3.26, 4.74, 1.2, 11.17)), 'SN': Country('SN', 'Senegal', 'africa/senegal-and-gambia', (-17.54, 12.31, -11.36, 16.69)), 'GN': Country('GN', 'Guinee', 'africa/guinea', (-15.08, 7.19, -7.64, 12.68)), 'TG': Country('TG', 'Togo', 'africa/togo', (-0.15, 6.1, 1.81, 11.14)), 'BJ': Country('BJ', 'Benin', 'africa/benin', (0.77, 6.14, 3.85, 12.41)), 'NG': Country('NG', 'Nigeria', 'africa/nigeria', (2.69, 4.27, 14.68, 13.89)), 'CM': Country('CM', 'Cameroun', 'africa/cameroon', (8.49, 1.65, 16.19, 13.08)), 'SL': Country('SL', 'Sierra Leone', 'africa/sierra-leone', (-13.3, 6.93, -10.27, 10.0)), 'LR': Country('LR', 'Liberia', 'africa/liberia', (-11.49, 4.35, -7.37, 8.55)), 'MR': Country('MR', 'Mauritanie', 'africa/mauritania', (-17.07, 14.72, -4.83, 27.3)), 'GW': Country('GW', 'Guinee-Bissau', 'africa/guinea-bissau', (-16.71, 10.87, -13.64, 12.68)), 'CD': Country('CD', 'RD Congo', 'africa/congo-democratic-republic', (12.18, -13.46, 31.31, 5.39)), 'KE': Country('KE', 'Kenya', 'africa/kenya', (33.91, -4.68, 41.91, 5.03)), 'TZ': Country('TZ', 'Tanzanie', 'africa/tanzania', (29.33, -11.75, 40.44, -0.99)), 'UG': Country('UG', 'Ouganda', 'africa/uganda', (29.57, -1.48, 35.04, 4.23)), 'ET': Country('ET', 'Ethiopie', 'africa/ethiopia', (32.99, 3.4, 47.99, 14.89)), 'MG': Country('MG', 'Madagascar', 'africa/madagascar', (43.18, -25.61, 50.48, -11.95))} module-attribute

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

Country dataclass

Source code in src/yoromaps/countries.py 
12
13
14
15
16
17
@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 
49
50
51
52
53
54
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

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 
 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
def extract_graph(pbf_path: str | Path, conn: sqlite3.Connection, progress=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 
35
36
37
38
39
40
41
42
43
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))