Skip to content

Extract

OpenStreetMap extract class.

OpenStreetMapExtract(
    id, name, parent, url, geometry, file_name=""
)

dataclass

OSM Extract metadata object.

OsmExtractSource

Bases: str, Enum

Enum of available OSM extract sources.

load_index_decorator(extract_source)

Decorator for loading OSM extracts index.

PARAMETER DESCRIPTION
extract_source

OpenStreetMap extract source. Used to save the index to cache.

TYPE: OsmExtractSource

Source code in quackosm/osm_extracts/extract.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
def load_index_decorator(
    extract_source: OsmExtractSource,
) -> Callable[[Callable[[], "GeoDataFrame"]], Callable[[], "GeoDataFrame"]]:
    """
    Decorator for loading OSM extracts index.

    Args:
        extract_source (OsmExtractSource): OpenStreetMap extract source.
            Used to save the index to cache.
    """

    def inner(function: Callable[[], "GeoDataFrame"]) -> Callable[[], "GeoDataFrame"]:
        def wrapper() -> "GeoDataFrame":
            global_cache_file_path = _get_global_cache_file_path(extract_source)
            global_cache_file_path.parent.mkdir(exist_ok=True, parents=True)
            expected_columns = ["id", "name", "file_name", "parent", "geometry", "area", "url"]

            # Check if index exists in cache
            if global_cache_file_path.exists():
                import geopandas as gpd

                index_gdf = gpd.read_file(global_cache_file_path)
            elif (local_cache_file_path := _get_local_cache_file_path(extract_source)).exists():
                import shutil

                import geopandas as gpd

                shutil.copy(local_cache_file_path, global_cache_file_path)
                index_gdf = gpd.read_file(global_cache_file_path)
            # Download index
            else:  # pragma: no cover
                index_gdf = function()
                # calculate extracts area
                index_gdf["area"] = index_gdf.geometry.apply(_calculate_geodetic_area)
                index_gdf.sort_values(by="area", ignore_index=True, inplace=True)

                # generate full file names
                apply_function = _get_full_file_name_function(index_gdf)
                index_gdf["file_name"] = index_gdf["id"].apply(apply_function)

                index_gdf = index_gdf[expected_columns]

            # Check if columns are right
            if set(expected_columns).symmetric_difference(index_gdf.columns):
                from quackosm._exceptions import OsmExtractIndexOutdatedWarning

                warnings.warn(
                    "Existing cached index has outdated structure. New index will be redownloaded.",
                    OsmExtractIndexOutdatedWarning,
                    stacklevel=0,
                )
                # Invalidate previous cached index
                global_cache_file_path.replace(global_cache_file_path.with_suffix(".geojson.old"))
                # Download index again
                index_gdf = wrapper()

            # Save index to cache
            if not global_cache_file_path.exists():
                global_cache_file_path.parent.mkdir(parents=True, exist_ok=True)
                index_gdf[expected_columns].to_file(global_cache_file_path, driver="GeoJSON")

            return index_gdf

        return wrapper

    return inner


extracts_to_geodataframe(extracts)

Transforms a list of OpenStreetMapExtracts to a GeoDataFrame.

Source code in quackosm/osm_extracts/extract.py 
116
117
118
119
120
121
122
def extracts_to_geodataframe(extracts: list[OpenStreetMapExtract]) -> "GeoDataFrame":
    """Transforms a list of OpenStreetMapExtracts to a GeoDataFrame."""
    import geopandas as gpd

    return gpd.GeoDataFrame(
        data=[asdict(extract) for extract in extracts], geometry="geometry"
    ).set_crs(WGS84_CRS)