accessmod/osm.py
def extract_pbf(
    src_files: Sequence[str],
    dst_file: str,
    area_of_interest: Polygon,
    expressions: Sequence[str],
    properties: Sequence[str],
    geom_type: str = "LineString",
) -> str:
    """Filter records from multiple .osm.pbf files based on location and tags.

    Records from each .osm.pbf file are filtered based on their tags and the
    osmium tag expressions provided by the user. Records are then exported to
    GeoJSON files and processed with Fiona to drop records outside of the area
    of interest.pbfs

    All valid records are merged into a final Geopackage.

    Parameters
    ----------
    src_files : list of str
        Input .osm.pbf files.
    dst_file : str
        Output .gpkg gile.
    area_of_interest : shapely polygon
        Geographic area of interest.
    expressions : list of str
        OSM tag filter expression (e.g. "w/highway").
    properties : list of str
        List of OSM tags to keep as column in the output .gpkg.
    geom_type : str, optional
        Geometry type (e.g. LineString or Polygon)

    Return
    ------
    str
        Path to output .gpkg file.
    """
    # minimal `osmium export` json config to drop unwanted attributes
    osmium_export_config = {
        "linear_tags": True,
        "area_tags": True,
        "include_tags": properties,
    }

    # schema of the output .gpkg file
    dst_schema = {
        "properties": OrderedDict({tag: "str" for tag in properties}),
        "geometry": geom_type,
    }

    # we will convert Polygon to MultiPolygon anyway
    if dst_schema.get("geometry") == "Polygon":
        dst_schema.update(geometry="MultiPolygon")

    fs_in = utils.filesystem(src_files[0])
    fs_out = utils.filesystem(dst_file)

    with tempfile.TemporaryDirectory(prefix="AccessMod_") as tmp_dir:

        dst_file_tmp = os.path.join(tmp_dir, os.path.basename(dst_file))

        for src_file_i, src_file in enumerate(src_files):

            src_file_tmp = os.path.join(tmp_dir, os.path.basename(src_file))
            fs_in.get(src_file, src_file_tmp)

            for expression_i, expression in enumerate(expressions):

                filtered_fp = src_file_tmp.replace(".osm.pbf", "_filtered.osm.pbf")
                geojson_fp = src_file_tmp.replace(".osm.pbf", ".geojson")

                osmium(
                    "tags-filter",
                    "--overwrite",
                    "-o",
                    filtered_fp,
                    src_file_tmp,
                    expression,
                )
                logger.info(f"Filtered {src_file_tmp} based on expression {expression}")

                # osmium export needs some config options to be provided in a
                # json file so we temporarily create one
                with tempfile.NamedTemporaryFile(suffix=".json") as tmp_file:
                    with open(tmp_file.name, "w") as f:
                        json.dump(osmium_export_config, f)
                    osmium(
                        "export",
                        filtered_fp,
                        "-f",
                        "json",
                        f"--geometry-types={geom_type.lower()}",
                        "-c",
                        tmp_file.name,
                        "-o",
                        geojson_fp,
                        "--overwrite",
                    )
                logger.info(f"Exported {filtered_fp} to {geojson_fp}")

                with fiona.open(geojson_fp, driver="GeoJSON") as src:

                    # create a new gpkg if we are processing the first file, then
                    # append to existing one
                    mode = "w" if src_file_i == 0 and expression_i == 0 else "a"

                    def match_schema(record: dict, dst_schema: dict) -> dict:
                        """Ensure that the Fiona record matches the target schema.

                        If a property is in the record but not available in the
                        target schema, then the property is dropped. If the property
                        is in the target schema but not in the record, then it is
                        added to the record with None as value.
                        """
                        dst_properties = OrderedDict(
                            {
                                prop: record["properties"].get(prop)
                                for prop in dst_schema["properties"]
                            }
                        )
                        record.update(properties=dst_properties)
                        return record

                    def to_multipolygon(record: dict) -> dict:
                        """Convert polygon record to multipolygon."""
                        if record["geometry"].get("type") == "MultiPolygon":
                            return record
                        record["geometry"]["type"] = "MultiPolygon"
                        record["geometry"]["coordinates"] = [
                            record["geometry"]["coordinates"]
                        ]
                        return record

                    # we filter and write feature per feature in the destination
                    # gpkg to avoid memory issues in the largest areas of interest
                    n_records = 0
                    with fiona.open(
                        dst_file_tmp, mode, driver="GPKG", schema=dst_schema
                    ) as dst:
                        for record in src:
                            if "geometry" in record:
                                if geom_type == "Polygon":
                                    record = to_multipolygon(record)
                                geom = shape(record["geometry"])
                                if geom.intersects(area_of_interest):
                                    record = match_schema(record, dst_schema)
                                    dst.write(record)
                                    n_records += 1

                    logger.info(
                        f"Written {n_records} records from {geojson_fp} into {dst_file_tmp}"
                    )

        fs_out.put(dst_file_tmp, dst_file)
        logger.info(f"Put {dst_file_tmp} into {dst_file}")

        return dst_file
Filter records from multiple .osm.pbf files based on location and tags.

    Records from each .osm.pbf file are filtered based on their tags and the
    osmium tag expressions provided by the user. Records are then exported to
    GeoJSON files and processed with Fiona to drop records outside of the area
    of interest.pbfs

    All valid records are merged into a final Geopackage.

    Parameters
    ----------
    src_files : list of str
        Input .osm.pbf files.
    dst_file : str
        Output .gpkg gile.
    area_of_interest : shapely polygon
        Geographic area of interest.
    expressions : list of str
        OSM tag filter expression (e.g. "w/highway").
    properties : list of str
        List of OSM tags to keep as column in the output .gpkg.
    geom_type : str, optional
        Geometry type (e.g. LineString or Polygon)

    Return
    ------
    str
        Path to output .gpkg file.
    
Filter records from multiple .osm.pbf files based on location and tags.
Records from each .osm.pbf file are filtered based on their tags and the
osmium tag expressions provided by the user. Records are then exported to
GeoJSON files and processed with Fiona to drop records outside of the area
of interest.pbfs

All valid records are merged into a final Geopackage.

Parameters

src_files : list of str
Input .osm.pbf files.
dst_file : str
Output .gpkg gile.
area_of_interest : shapely polygon
Geographic area of interest.
expressions : list of str
OSM tag filter expression .
properties : list of str
List of OSM tags to keep as column in the output .gpkg.
geom_type : str, optional
Geometry type

Return

str
Path to output .gpkg file.
Filter records from multiple . osm . pbf files based on location and tags . Records from each . osm . pbf file are filtered based on their tags and the osmium tag expressions provided by the user . Records are then exported to GeoJSON files and processed with Fiona to drop records outside of the area of interest . pbfs All valid records are merged into a final Geopackage . Parameters src_files : list of str Input . osm . pbf files . dst_file : str Output . gpkg gile . area_of_interest : shapely polygon Geographic area of interest . expressions : list of str OSM tag filter expression . properties : list of str List of OSM tags to keep as column in the output . gpkg . geom_type : str optional Geometry type Return str Path to output . gpkg file .
def extract_pbf(
    src_files: Sequence[str],
    dst_file: str,
    area_of_interest: Polygon,
    expressions: Sequence[str],
    properties: Sequence[str],
    geom_type: str = "LineString",
) -> str:
    osmium_export_config = {
        "linear_tags": True,
        "area_tags": True,
        "include_tags": properties,
    }
    dst_schema = {
        "properties": OrderedDict({tag: "str" for tag in properties}),
        "geometry": geom_type,
    }
    if dst_schema.get("geometry") == "Polygon":
        dst_schema.update(geometry="MultiPolygon")
    fs_in = utils.filesystem(src_files[0])
    fs_out = utils.filesystem(dst_file)
    with tempfile.TemporaryDirectory(prefix="AccessMod_") as tmp_dir:
        dst_file_tmp = os.path.join(tmp_dir, os.path.basename(dst_file))
        for src_file_i, src_file in enumerate(src_files):
            src_file_tmp = os.path.join(tmp_dir, os.path.basename(src_file))
            fs_in.get(src_file, src_file_tmp)
            for expression_i, expression in enumerate(expressions):
                filtered_fp = src_file_tmp.replace(".osm.pbf", "_filtered.osm.pbf")
                geojson_fp = src_file_tmp.replace(".osm.pbf", ".geojson")
                osmium(
                    "tags-filter",
                    "--overwrite",
                    "-o",
                    filtered_fp,
                    src_file_tmp,
                    expression,
                )
                logger.info(f"Filtered {src_file_tmp} based on expression {expression}")
                with tempfile.NamedTemporaryFile(suffix=".json") as tmp_file:
                    with open(tmp_file.name, "w") as f:
                        json.dump(osmium_export_config, f)
                    osmium(
                        "export",
                        filtered_fp,
                        "-f",
                        "json",
                        f"--geometry-types={geom_type.lower()}",
                        "-c",
                        tmp_file.name,
                        "-o",
                        geojson_fp,
                        "--overwrite",
                    )
                logger.info(f"Exported {filtered_fp} to {geojson_fp}")
                with fiona.open(geojson_fp, driver="GeoJSON") as src:
                    mode = "w" if src_file_i == 0 and expression_i == 0 else "a"
                    def match_schema(record: dict, dst_schema: dict) -> dict:
                        dst_properties = OrderedDict(
                            {
                                prop: record["properties"].get(prop)
                                for prop in dst_schema["properties"]
                            }
                        )
                        record.update(properties=dst_properties)
                        return record
                    def to_multipolygon(record: dict) -> dict:
                        if record["geometry"].get("type") == "MultiPolygon":
                            return record
                        record["geometry"]["type"] = "MultiPolygon"
                        record["geometry"]["coordinates"] = [
                            record["geometry"]["coordinates"]
                        ]
                        return record
                    n_records = 0
                    with fiona.open(
                        dst_file_tmp, mode, driver="GPKG", schema=dst_schema
                    ) as dst:
                        for record in src:
                            if "geometry" in record:
                                if geom_type == "Polygon":
                                    record = to_multipolygon(record)
                                geom = shape(record["geometry"])
                                if geom.intersects(area_of_interest):
                                    record = match_schema(record, dst_schema)
                                    dst.write(record)
                                    n_records += 1
                    logger.info(
                        f"Written {n_records} records from {geojson_fp} into {dst_file_tmp}"
                    )
        fs_out.put(dst_file_tmp, dst_file)
        logger.info(f"Put {dst_file_tmp} into {dst_file}")
        return dst_file
def extract_pbf ( src_files : Sequence [ str ] , dst_file : str , area_of_interest : Polygon , expressions : Sequence [ str ] , properties : Sequence [ str ] , geom_type : str = "LineString" , ) -> str : osmium_export_config = { "linear_tags" : True , "area_tags" : True , "include_tags" : properties , } dst_schema = { "properties" : OrderedDict ( { tag : "str" for tag in properties } ) , "geometry" : geom_type , } if dst_schema . get ( "geometry" ) == "Polygon" : dst_schema . update ( geometry = "MultiPolygon" ) fs_in = utils . filesystem ( src_files [ 0 ] ) fs_out = utils . filesystem ( dst_file ) with tempfile . TemporaryDirectory ( prefix = "AccessMod_" ) as tmp_dir : dst_file_tmp = os . path . join ( tmp_dir , os . path . basename ( dst_file ) ) for src_file_i , src_file in enumerate ( src_files ) : src_file_tmp = os . path . join ( tmp_dir , os . path . basename ( src_file ) ) fs_in . get ( src_file , src_file_tmp ) for expression_i , expression in enumerate ( expressions ) : filtered_fp = src_file_tmp . replace ( ".osm.pbf" , "_filtered.osm.pbf" ) geojson_fp = src_file_tmp . replace ( ".osm.pbf" , ".geojson" ) osmium ( "tags-filter" , "--overwrite" , "-o" , filtered_fp , src_file_tmp , expression , ) logger . info ( f"Filtered {src_file_tmp} based on expression {expression}" ) with tempfile . NamedTemporaryFile ( suffix = ".json" ) as tmp_file : with open ( tmp_file . name , "w" ) as f : json . dump ( osmium_export_config , f ) osmium ( "export" , filtered_fp , "-f" , "json" , f"--geometry-types={geom_type.lower()}" , "-c" , tmp_file . name , "-o" , geojson_fp , "--overwrite" , ) logger . info ( f"Exported {filtered_fp} to {geojson_fp}" ) with fiona . open ( geojson_fp , driver = "GeoJSON" ) as src : mode = "w" if src_file_i == 0 and expression_i == 0 else "a" def match_schema ( record : dict , dst_schema : dict ) -> dict : """Ensure that the Fiona record matches the target schema.

                        If a property is in the record but not available in the
                        target schema, then the property is dropped. If the property
                        is in the target schema but not in the record, then it is
                        added to the record with None as value.
                        """ dst_properties = OrderedDict ( { prop : record [ "properties" ] . get ( prop ) for prop in dst_schema [ "properties" ] } ) record . update ( properties = dst_properties ) return record def to_multipolygon ( record : dict ) -> dict : """Convert polygon record to multipolygon.""" if record [ "geometry" ] . get ( "type" ) == "MultiPolygon" : return record record [ "geometry" ] [ "type" ] = "MultiPolygon" record [ "geometry" ] [ "coordinates" ] = [ record [ "geometry" ] [ "coordinates" ] ] return record n_records = 0 with fiona . open ( dst_file_tmp , mode , driver = "GPKG" , schema = dst_schema ) as dst : for record in src : if "geometry" in record : if geom_type == "Polygon" : record = to_multipolygon ( record ) geom = shape ( record [ "geometry" ] ) if geom . intersects ( area_of_interest ) : record = match_schema ( record , dst_schema ) dst . write ( record ) n_records += 1 logger . info ( f"Written {n_records} records from {geojson_fp} into {dst_file_tmp}" ) fs_out . put ( dst_file_tmp , dst_file ) logger . info ( f"Put {dst_file_tmp} into {dst_file}" ) return dst_file
Filter records from multiple .osm.pbf files based on location and tags.
Filter records from multiple . osm . pbf files based on location and tags .
"""Filter records from multiple .osm.pbf files based on location and tags.

    Records from each .osm.pbf file are filtered based on their tags and the
    osmium tag expressions provided by the user. Records are then exported to
    GeoJSON files and processed with Fiona to drop records outside of the area
    of interest.pbfs

    All valid records are merged into a final Geopackage.

    Parameters
    ----------
    src_files : list of str
        Input .osm.pbf files.
    dst_file : str
        Output .gpkg gile.
    area_of_interest : shapely polygon
        Geographic area of interest.
    expressions : list of str
        OSM tag filter expression (e.g. "w/highway").
    properties : list of str
        List of OSM tags to keep as column in the output .gpkg.
    geom_type : str, optional
        Geometry type (e.g. LineString or Polygon)

    Return
    ------
    str
        Path to output .gpkg file.
    """ # minimal `osmium export` json config to drop unwanted attributes # schema of the output .gpkg file # we will convert Polygon to MultiPolygon anyway # osmium export needs some config options to be provided in a # json file so we temporarily create one # create a new gpkg if we are processing the first file, then # append to existing one """Ensure that the Fiona record matches the target schema.

                        If a property is in the record but not available in the
                        target schema, then the property is dropped. If the property
                        is in the target schema but not in the record, then it is
                        added to the record with None as value.
                        """ """Convert polygon record to multipolygon.""" # we filter and write feature per feature in the destination # gpkg to avoid memory issues in the largest areas of interest
src_files
Sequence[str]
dst_file
str
area_of_interest
Polygon
expressions
Sequence[str]
properties
Sequence[str]
geom_type
str