Overture Maps Downloader simplifies geospatial data manipulation by integrating the powerful DuckDB, Dask DataFrames, and GDAL/OGR open source tools.
import logging
import warnings
from rich.console import Console

console = Console()

warnings.simplefilter(action="ignore", category=FutureWarning)
# Setup logging configuration
logging.basicConfig(
    level=logging.INFO,
    format="[%(asctime)s] %(levelname)s: %(message)s",
    datefmt="%Y-%m-%d %H:%M:%S",
)

overturemapsdownloader/__init__.py
Youssef-Harby/OvertureMapsDownloader
import logging
from pathlib import Path
from overturemapsdownloader.yaml_helper import read_yaml_file

# Initialize logging
logging.basicConfig(
    level=logging.INFO,
    format="[%(asctime)s] %(levelname)s: %(message)s",
    datefmt="%Y-%m-%d %H:%M:%S",
)


class Config:
    def __init__(self, config_dict):
        for key, value in config_dict.items():
            if key == "themes":
                self.themes = [Config(theme_dict) for theme_dict in value]
            elif isinstance(value, dict):
                setattr(self, key.lower(), Config(value))
            else:
                setattr(self, key.lower(), value)

    def update_attribute(self, attribute_name, new_values):
        attribute = getattr(self, attribute_name, None)
        if attribute is not None:
            for key, value in new_values.items():
                setattr(attribute, key.lower(), value)
        else:
            raise AttributeError(f"{attribute_name} not found in Config.")

    def to_dict(self):
        result = {}
        for key, value in self.__dict__.items():
            if isinstance(value, Config):
                result[key] = value.to_dict()
            elif isinstance(value, list):
                result[key] = [
                    item.to_dict() if isinstance(item, Config) else item
                    for item in value
                ]
            else:
                result[key] = value
        return result

    def format_url(self, url_name, theme=None, ptype=None):
        if hasattr(self.urls, url_name.lower()):
            url_template = getattr(self.urls, url_name.lower())
            final_theme = theme if theme else self.global_variables.default_theme
            final_type = ptype if ptype else self.global_variables.default_type
            return url_template.format(
                release=self.global_variables.release,
                theme=final_theme,
                ptype=final_type,
            )
        else:
            return None


def load_config(yaml_file=None, **kwargs):
    if yaml_file is None:
        default_path = Path(__file__).parent.parent / "config.yml"
        if default_path.exists():
            yaml_file = default_path
        else:
            yaml_file = input(
                "Default config.yml not found. Please enter the path to the config.yml file: "
            )
            yaml_file = Path(yaml_file)
            if not yaml_file.exists():
                logging.error(f"No such file found: {yaml_file}")
                return None

    config_data = read_yaml_file(yaml_file)
    if config_data is None:
        logging.error("Configuration not loaded due to an error in the YAML file.")
        return None

    config = Config(config_data)
    return config


if __name__ == "__main__":
    config = load_config("config.yml")

    if config is None:
        logging.error("Failed to load config.")
        exit(1)

    # Scenario 1: Retrieve S3 URL with default theme and type
    url1 = config.format_url("Amazon_S3")
    logging.info(f"Scenario 1: S3 URL with default theme and type: {url1}")

    # Scenario 2: Retrieve Azure URL with default theme and type
    url2 = config.format_url("Microsoft_Azure")
    logging.info(f"Scenario 2: Azure URL with default theme and type: {url2}")

    # Scenario 3: Retrieve S3 URL with specific theme and type
    url3 = config.format_url("Amazon_S3", theme="buildings", ptype="building")
    logging.info(f"Scenario 3: S3 URL with specific theme and type: {url3}")

    # Scenario 4: Retrieve Azure URL with specific theme and default type
    url4 = config.format_url("Microsoft_Azure", theme="admins")
    logging.info(f"Scenario 4: Azure URL with specific theme and default type: {url4}")

    # Scenario 5: Update global variables and retrieve S3 URL
    config.update_attribute("global_variables", {"release": "2023-08-01-beta.1"})
    url5 = config.format_url("Amazon_S3")
    logging.info(f"Scenario 5: S3 URL with updated release: {url5}")

    # Scenario 6: Print the entire configuration as a dictionary
    logging.info(f"Scenario 6: Config as dict: {config.to_dict()}")

overturemapsdownloader/config.py
Youssef-Harby/OvertureMapsDownloader
import logging
import dask.dataframe as dd
import geopandas as gpd
import dask_geopandas as dgpd
from overturemapsdownloader.utils_helper import read_geospatial_data
from dask.diagnostics import ProgressBar

ProgressBar().register()


def get_df_from_parquet(
    parquet_path,
    engine="pyarrow",
    # columns=["geometry"],  # comment to get all columns by default
    storage_options={"anon": True},
    parquet_file_extensions=False,
):
    """
    Reads a Dask DataFrame from a Parquet file.
    """
    try:
        logging.info(f"Reading Parquet file from {parquet_path}")
        df = dd.read_parquet(
            parquet_path,
            # columns=columns,  # comment to get all columns by default
            engine=engine,
            index="id",
            dtype_backend=engine,
            storage_options=storage_options,
            parquet_file_extensions=parquet_file_extensions,
        )
        return df
    except Exception as e:
        logging.error(f"An error occurred: {e}")
        return None


def make_gdf_from_df(df, crs=4326):
    """
    Converts a Dask DataFrame to a Dask GeoDataFrame.
    """
    geometry = (
        df["geometry"]
        .map_partitions(gpd.GeoSeries.from_wkb, meta=gpd.GeoSeries(name="geometry"))
        .set_crs(crs)
    )
    return dgpd.from_dask_dataframe(df, geometry=geometry)


def get_clipped_gdf(gdf, bbox_filter):
    """
    Clips the GeoDataFrame using a bounding box.
    """
    return gdf[gdf.geometry.within(bbox_filter)]


if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO)

    # TODO: Handle columns with official schemas
    schema_yaml_path = "overturemapsdownloader/schemas/schema/places/place.yaml"

    bbox_filter = read_geospatial_data(
        "examples/bbox.geojson", as_shapely_str=True, output_format="Custom"
    )

    df = get_df_from_parquet(
        parquet_path="s3://overturemaps-us-west-2/release/2023-07-26-alpha.0/theme=places/type=*/*",
        # columns=get_columns_from_om_schema_yaml(schema_yaml_path),
    )

    if df is not None:
        gdf = make_gdf_from_df(df)

        # TODO: Add filter by country (also in config)
        clipped_gdf = get_clipped_gdf(gdf, bbox_filter)

        print(clipped_gdf.head())
    else:
        logging.error("Could not read the DataFrame from the Parquet file.")

    # TODO: Write to file; Parquet by default. Allow user to convert to other formats (e.g., via ogr2ogr).

overturemapsdownloader/dask_qrys.py
Youssef-Harby/OvertureMapsDownloader
import typer
import warnings
import logging
from pathlib import Path
from overturemapsdownloader.om_logic import om_dask_to_parquet
from overturemapsdownloader.config import load_config
from rich import print


def callback():
    logging.info("Running a command")


app = typer.Typer(callback=callback)


@app.command()
def omaps(
    theme: str = typer.Option(None, help="Theme of the download, e.g., building"),
    ptype: str = typer.Option(None, help="Type of the parquet, e.g., buildings"),
    bbox: Path = typer.Option(
        None, help="Path to GeoJSON file containing bounding box coordinates"
    ),
    output: Path = typer.Option(None, help="Path to store the output GeoPackage file"),
):
    config = load_config()
    if config is None:
        logging.warning("Configuration not loaded.")
        exit(1)
    if theme:
        config.update_attribute("global_variables", {"default_theme": theme})
        logging.info(f"Theme: {config.global_variables.default_theme}")
    if ptype:
        if theme == "transportation" and ptype in ["connector", "segment"]:
            config.update_attribute("global_variables", {"default_type": ptype})
            query_url = config.format_url("Wherobots_S3", theme=theme, ptype=ptype)
            logging.info(f"Parquet url: {query_url}")
        else:
            config.update_attribute("global_variables", {"default_type": ptype})
            query_url = config.format_url("Amazon_S3", theme=theme, ptype=ptype)
            logging.info(f"Parquet url: {query_url}")
    if bbox:
        config.update_attribute("global_variables", {"filter_by_bbox": True})
        config.update_attribute("global_variables", {"bbox_file_path": str(bbox)})
        logging.info(f"Bounding Box File: {config.global_variables.bbox_file_path}")
    if output:
        config.update_attribute("global_variables", {"output_file_path": str(output)})
        logging.info(f"Output File: {config.global_variables.output_file_path}")

    om_dask_to_parquet(config, query_url)


if __name__ == "__main__":
    app()

overturemapsdownloader/main.py
Youssef-Harby/OvertureMapsDownloader
import logging
from overturemapsdownloader.dask_qrys import (
    get_clipped_gdf,
    get_df_from_parquet,
    make_gdf_from_df,
)
from overturemapsdownloader.utils_helper import read_geospatial_data


def om_dask_to_parquet(config, query_url):
    print(f"Query URL: {query_url}")
    bbox_filter = read_geospatial_data(
        config.global_variables.bbox_file_path,
        as_shapely_str=True,
        output_format="Custom",
    )

    df = get_df_from_parquet(
        parquet_path=query_url,
        # columns=get_columns_from_om_schema_yaml(schema_yaml_path),
    )

    if df is not None:
        gdf = make_gdf_from_df(df, int(config.global_variables.default_epsg))

        # TODO: Add filter by country (also in config)
        if config.global_variables.filter_by_bbox:
            clipped_gdf = get_clipped_gdf(gdf, bbox_filter)
            logging.info(f"Clipped GeoDataFrame: {clipped_gdf}")
            print(clipped_gdf.head(10))
            if clipped_gdf is not None:
                logging.info(
                    f'Writing GeoDataFrame to "{config.global_variables.output_file_path}"'
                )
                clipped_gdf.to_parquet(config.global_variables.output_file_path)
                logging.info("GeoDataFrame written to Parquet file.")
            else:
                logging.error("Could not clip the GeoDataFrame.")
        elif config.global_variables.filter_by_country:
            pass

    else:
        logging.error("Could not read the DataFrame from the Parquet file.")


if __name__ == "__main__":
    om_dask_to_parquet()

overturemapsdownloader/om_logic.py
Youssef-Harby/OvertureMapsDownloader
from overturemapsdownloader.yaml_helper import read_yaml_file


def get_columns_from_om_schema_yaml(yaml_path):
    columns = []
    schema = read_yaml_file(yaml_path)

    # handle geometry field
    if "geometry" in schema["properties"]:
        columns.append("geometry")

    # handle properties
    if "properties" in schema["properties"]:
        prop_schema = schema["properties"]["properties"]["properties"]

        for k, v in prop_schema.items():
            if "type" in v and v["type"] == "object":
                for sub_k in v["properties"].keys():
                    columns.append(f"{k}.{sub_k}")
            else:
                columns.append(k)

    return columns


if __name__ == "__main__":
    yaml_path = "overturemapsdownloader/schemas/schema/places/place.yaml"
    columns = get_columns_from_om_schema_yaml(yaml_path)
    print("Columns:", columns)

overturemapsdownloader/read_om_schemas.py
Youssef-Harby/OvertureMapsDownloader
import geopandas as gpd
import pandas as pd
from pathlib import Path
import fiona
from shapely import wkt, wkb
from shapely.geometry import Polygon, MultiPolygon
import logging


def detect_file_type(filepath: str):
    """
    Detect the type of geospatial file based on its extension.

    :param filepath: The path to the geospatial data file.
    :return: A string representing the file type.
    """
    # Create a Path object and get the file extension (without the dot)
    path = Path(filepath)
    extension = path.suffix[1:].lower()

    # Check if the extension is one of the recognized types
    if extension in ["geojson", "gpkg", "shp", "gdb", "csv", "xls", "xlsx"]:
        return extension
    else:
        raise ValueError(f"Unsupported file type: {extension}")


def remove_extension(full_path):
    """
    Removes the file extension from a given path.

    :param full_path: The full path to the file.
    :return: The file path without the extension.
    """
    path = Path(full_path)
    return str(path.with_suffix(""))


def read_geospatial_data(
    filepath,
    table_name=None,
    as_shapely_str=False,
    geometry_types=["Polygon", "MultiPolygon"],
    output_format="WKT",
):
    """
    Reads a geospatial data file and returns it as a GeoDataFrame or with geometries as Shapely strings.

    :param filepath: The path to the geospatial data file.
    :param table_name: The name of the table or layer to load (for GeoPackage or File Geodatabase).
    :param as_shapely_str: If True, return geometries as Shapely strings.
    :param geometry_types: List of geometry types to filter. Default includes 'Polygon' and 'MultiPolygon'.
    :return: A GeoDataFrame containing the data from the file.
    """
    # Ensure file exists
    if not Path(filepath).exists():
        raise FileNotFoundError(f"File not found: {filepath}")

    # Detect the file type
    file_type = detect_file_type(filepath)

    # Map file type to driver
    driver_map = {
        "geojson": "GeoJSON",
        "shp": "ESRI Shapefile",
        "gpkg": "GPKG",
        "gdb": "OpenFileGDB",
    }

    # Check if the driver is one of the recognized types
    if file_type not in driver_map:
        raise ValueError(f"Unsupported file type: {file_type}")

    try:
        # Read the file
        if file_type in ["gpkg", "gdb"]:
            if table_name is None:
                raise ValueError(f"A table name must be provided for {file_type} files")
            else:
                gdf = gpd.read_file(
                    filepath, layer=table_name, driver=driver_map[file_type]
                )
        else:
            gdf = gpd.read_file(filepath, driver=driver_map[file_type])

        # Filter by geometry type
        gdf = gdf[gdf["geometry"].apply(lambda geom: geom.geom_type in geometry_types)]

        if as_shapely_str:
            first_geom = gdf.loc[0, "geometry"]
            if first_geom.geom_type in geometry_types:
                if output_format == "WKT":
                    return str(first_geom)
                elif output_format == "Custom":
                    if isinstance(first_geom, Polygon) or isinstance(
                        first_geom, MultiPolygon
                    ):
                        return first_geom
                    else:
                        logging.info(
                            f"The geometry is not a Polygon, it's a {first_geom.geom_type}"
                        )
                        exit(1)
                        return None
                else:
                    raise ValueError(
                        "Invalid output_format. Choose from 'WKT' or 'Custom'."
                    )
        return gdf

    except Exception as e:
        logging.error(f"An error occurred while reading the file: {e}")


def write_geospatial_data(gdf, filepath, driver=None, layer=None, encoding="utf-8"):
    """
    Writes a GeoDataFrame to a geospatial data file.

    :param gdf: The GeoDataFrame to write.
    :param filepath: The path to the file to write.
    :param driver: The driver to use for writing the file.
    :param layer: The layer to write (for GeoPackage or File Geodatabase).
    :param encoding: The encoding to use for writing the file. Defaults to 'utf-8'.
    :return: newfilepath: The path of the file that was written.
    """
    # Detect the file type if no driver is given
    if driver is None:
        driver = detect_file_type(filepath)
    newfilepath = remove_extension(filepath) + "." + driver

    # Map file type to driver
    driver_map = {
        "geojson": "GeoJSON",
        "shp": "ESRI Shapefile",
        "gpkg": "GPKG",
        "gdb": "OpenFileGDB",
    }

    # Check if the driver is one of the recognized types
    if driver not in driver_map:
        raise ValueError(f"Unsupported file type: {driver}")

    # Write the GeoDataFrame to file
    try:
        if driver == "gdb":
            # Make sure the OpenFileGDB driver is supported
            assert (
                "OpenFileGDB" in fiona.supported_drivers
            ), "OpenFileGDB driver is not supported"
            # Make sure the OpenFileGDB driver supports writing ('w')
            assert (
                "w" in fiona.supported_drivers["OpenFileGDB"]
            ), "OpenFileGDB driver does not support writing"
            gdf.to_file(
                newfilepath, driver=driver_map[driver], layer=layer, encoding=encoding
            )
        elif driver == "gpkg":
            gdf.to_file(
                newfilepath, driver=driver_map[driver], layer=layer, encoding=encoding
            )
        else:
            gdf.to_file(newfilepath, driver=driver_map[driver], encoding=encoding)
    except Exception as e:
        logging.error(f"An error occurred while writing the file: {e}")
    return newfilepath


def detect_geometry_format(geometry_sample):
    """
    Detect the format of the geometry (WKT or WKB).

    :param geometry_sample: A sample geometry to detect.
    :return: A string representing the geometry format ('WKT' or 'WKB').
    """
    if isinstance(geometry_sample, str):
        return "WKT"
    elif isinstance(geometry_sample, bytes):
        return "WKB"
    else:
        try:
            wkt.loads(str(geometry_sample))
            return "WKT"
        except:
            try:
                wkb.loads(bytes(geometry_sample))
                return "WKB"
            except:
                return None

overturemapsdownloader/utils_helper.py
Youssef-Harby/OvertureMapsDownloader
# yaml.py
import yaml
import logging


def read_yaml_file(filepath):
    try:
        with open(filepath, "r", encoding="utf8") as stream:
            schema_yaml = yaml.safe_load(stream)
            return schema_yaml
    except yaml.YAMLError as exc:
        logging.error(f"Error in reading YAML file {filepath}: {exc}")
        return None
    except FileNotFoundError as fnf_error:
        logging.error(f"File not found: {filepath}")
        return None


def create_yaml_file(filepath, data):
    try:
        with open(filepath, "a", encoding="utf-8") as file:
            yaml.dump(data, file, encoding="utf-8")
    except yaml.YAMLError as exc:
        logging.error(f"Error in writing YAML file {filepath}: {exc}")
        return None

overturemapsdownloader/yaml_helper.py
Youssef-Harby/OvertureMapsDownloader