basemapper.py

Thread to handle downloads for Queue.

Parameters:

Name	Type	Description	Default
dest	str	The filespec of the tile cache.	required
mirrors	list	The list of mirrors to get imagery.	required
tiles	list	The list of tiles to download.	required

Source code in osm_fieldwork/basemapper.py

def dlthread(dest: str, mirrors: list[dict], tiles: list[tuple]) -> None:
    """Thread to handle downloads for Queue.

    Args:
        dest (str): The filespec of the tile cache.
        mirrors (list): The list of mirrors to get imagery.
        tiles (list): The list of tiles to download.
    """
    if len(tiles) == 0:
        # epdb.st()
        return

    # Create the subdirectories as pySmartDL doesn't do it for us
    Path(dest).mkdir(parents=True, exist_ok=True)

    log.info(f"Downloading {len(tiles)} tiles in thread {threading.get_ident()} to {dest}")

    with concurrent.futures.ThreadPoolExecutor(max_workers=4) as executor:
        futures = [executor.submit(download_tile, dest, tile, mirrors) for tile in tiles]
        concurrent.futures.wait(futures)

options: show_source: false heading_level: 3

Bases: object

Basemapper parent class.

Parameters:

Name	Type	Description	Default
boundary	Union[str, BytesIO]	A BBOX string or GeoJSON provided as BytesIO object of the AOI. The GeoJSON can contain multiple geometries.	required
base	str	The base directory to cache map tile in	required
source	str	The upstream data source for map tiles	required

Returns:

Type	Description
BaseMapper	An instance of this class

Source code in osm_fieldwork/basemapper.py

def __init__(
    self,
    boundary: Union[str, BytesIO],
    base: str,
    source: str,
):
    """Create an tile basemap for ODK Collect.

    Args:
        boundary (Union[str, BytesIO]): A BBOX string or GeoJSON provided as BytesIO object of the AOI.
            The GeoJSON can contain multiple geometries.
        base (str): The base directory to cache map tile in
        source (str): The upstream data source for map tiles

    Returns:
        (BaseMapper): An instance of this class
    """
    bbox_factory = BoundaryHandlerFactory(boundary)
    self.bbox = bbox_factory.get_bounding_box()
    self.tiles = list()
    self.base = base
    # sources for imagery
    self.source = source
    self.sources = dict()

    path = xlsforms_path.replace("xlsforms", "imagery.yaml")
    self.yaml = YamlFile(path)

    for entry in self.yaml.yaml["sources"]:
        for k, v in entry.items():
            src = dict()
            for item in v:
                src["source"] = k
                for k1, v1 in item.items():
                    # print(f"\tFIXME2: {k1} - {v1}")
                    src[k1] = v1
            self.sources[k] = src

customTMS

customTMS(url, is_oam=False, is_xy=False)

Add a custom TMS URL to the list of sources.

The url must end in %s to be replaced with the tile xyz values.

Format examples: https://basemap.nationalmap.gov/ArcGIS/rest/services/USGSTopo/MapServer/tile/{z}/{y}/{x} https://maps.nyc.gov/xyz/1.0.0/carto/basemap/%s https://maps.nyc.gov/xyz/1.0.0/carto/basemap/{z}/{x}/{y}.jpg

The method will replace {z}/{x}/{y}.jpg with %s

Parameters:

Name	Type	Description	Default
url	str	The URL string	required
source	str	The provier source, for setting attribution	required
is_xy	bool	Swap the x and y for the provider --> 'zxy'	False

Source code in osm_fieldwork/basemapper.py

def customTMS(self, url: str, is_oam: bool = False, is_xy: bool = False):
    """Add a custom TMS URL to the list of sources.

    The url must end in %s to be replaced with the tile xyz values.

    Format examples:
    https://basemap.nationalmap.gov/ArcGIS/rest/services/USGSTopo/MapServer/tile/{z}/{y}/{x}
    https://maps.nyc.gov/xyz/1.0.0/carto/basemap/%s
    https://maps.nyc.gov/xyz/1.0.0/carto/basemap/{z}/{x}/{y}.jpg

    The method will replace {z}/{x}/{y}.jpg with %s

    Args:
        url (str): The URL string
        source (str): The provier source, for setting attribution
        is_xy (bool): Swap the x and y for the provider --> 'zxy'
    """
    # Remove any file extensions if present and update the 'suffix' parameter
    # NOTE the file extension gets added again later for the download URL
    if url.endswith(".jpg"):
        suffix = "jpg"
        url = url[:-4]  # Remove the last 4 characters (".jpg")
    elif url.endswith(".png"):
        suffix = "png"
        url = url[:-4]  # Remove the last 4 characters (".png")
    else:
        # FIXME handle other formats for custom TMS
        suffix = "jpg"

    # If placeholders present, validate they have no additional spaces
    if "{" in url and "}" in url:
        pattern = r".*/\{[zxy]\}/\{[zxy]\}/\{[zxy]\}(?:/|/?)"
        if not bool(re.search(pattern, url)):
            msg = "Invalid TMS URL format. Please check the URL placeholders {z}/{x}/{y}."
            log.error(msg)
            raise ValueError(msg)

    # Remove "{z}/{x}/{y}" placeholders if they are present
    url = re.sub(r"/{[xyz]+\}", "", url)
    # Append "%s" to the end of the URL to later add the tile path
    url = url + r"/%s"

    if is_oam:
        # Override dummy OAM URL
        source = "oam"
        self.sources[source]["url"] = url
    else:
        source = "custom"
        tms_params = {"name": source, "url": url, "suffix": suffix, "source": source, "xy": is_xy}
        log.debug(f"Setting custom TMS with params: {tms_params}")
        self.sources[source] = tms_params

    # Select the source
    self.source = source

getFormat

getFormat()

Get the image format of the map tiles.

Returns:

Type	Description
str	the upstream source for map tiles.

Source code in osm_fieldwork/basemapper.py

def getFormat(self):
    """Get the image format of the map tiles.

    Returns:
        (str): the upstream source for map tiles.
    """
    return self.sources[self.source]["suffix"]

getTiles

getTiles(zoom)

Get a list of tiles for the specified zoom level.

Parameters:

Name	Type	Description	Default
zoom	int	The Zoom level of the desired map tiles.	required

Returns:

Name	Type	Description
int	int	The total number of map tiles downloaded.

Source code in osm_fieldwork/basemapper.py

def getTiles(self, zoom: int) -> int:
    """Get a list of tiles for the specified zoom level.

    Args:
        zoom (int): The Zoom level of the desired map tiles.

    Returns:
        int: The total number of map tiles downloaded.
    """
    info = get_cpu_info()
    cores = info["count"]

    self.tiles = list(mercantile.tiles(self.bbox[0], self.bbox[1], self.bbox[2], self.bbox[3], zoom))
    total = len(self.tiles)
    log.info(f"{total} tiles for zoom level {zoom}")

    mirrors = [self.sources[self.source]]
    chunk_size = max(1, round(total / cores))

    if total < chunk_size or chunk_size == 0:
        dlthread(self.base, mirrors, self.tiles)
    else:
        with concurrent.futures.ThreadPoolExecutor(max_workers=cores) as executor:
            futures = []
            for i in range(0, total, chunk_size):
                chunk = self.tiles[i : i + chunk_size]
                futures.append(executor.submit(dlthread, self.base, mirrors, chunk))
                log.debug(f"Dispatching Block {i}:{i + chunk_size}")
            concurrent.futures.wait(futures)

    return total

tileExists

tileExists(tile)

See if a map tile already exists.

Parameters:

Name	Type	Description	Default
tile	MapTile	The map tile to check for the existence of	required

Returns:

Type	Description
bool	Whether the tile exists in the map tile cache

Source code in osm_fieldwork/basemapper.py

def tileExists(
    self,
    tile: MapTile,
):
    """See if a map tile already exists.

    Args:
        tile (MapTile): The map tile to check for the existence of

    Returns:
        (bool): Whether the tile exists in the map tile cache
    """
    filespec = f"{self.base}{tile[2]}/{tile[1]}/{tile[0]}.{self.sources[{self.source}]['suffix']}"
    if Path(filespec).exists():
        log.debug("%s exists" % filespec)
        return True
    else:
        log.debug("%s doesn't exists" % filespec)
        return False

options: show_source: false heading_level: 3

Create a basemap with given parameters.

Parameters:

Name	Type	Description	Default
boundary	str | BytesIO	The boundary for the area you want.	None
tms	str	Custom TMS URL.	None
xy	bool	Swap the X & Y coordinates when using a custom TMS if True.	False
outfile	str	Output file name for the basemap.	None
zooms	str	The Zoom levels, specified as a range (e.g., "12-17") or comma-separated levels (e.g., "12,13,14").	'12-17'
outdir	str	Output directory name for tile cache.	None
source	str	Imagery source, one of ["esri", "bing", "topo", "google", "oam", "custom"] (default is "esri").	'esri'
append	bool	Whether to append to an existing file	False

Returns:

Type	Description
None	None

Source code in osm_fieldwork/basemapper.py

def create_basemap_file(
    boundary=None,
    tms=None,
    xy=False,
    outfile=None,
    zooms="12-17",
    outdir=None,
    source="esri",
    append: bool = False,
) -> None:
    """Create a basemap with given parameters.

    Args:
        boundary (str | BytesIO, optional): The boundary for the area you want.
        tms (str, optional): Custom TMS URL.
        xy (bool, optional): Swap the X & Y coordinates when using a
            custom TMS if True.
        outfile (str, optional): Output file name for the basemap.
        zooms (str, optional): The Zoom levels, specified as a range
            (e.g., "12-17") or comma-separated levels (e.g., "12,13,14").
        outdir (str, optional): Output directory name for tile cache.
        source (str, optional): Imagery source, one of
            ["esri", "bing", "topo", "google", "oam", "custom"] (default is "esri").
        append (bool, optional): Whether to append to an existing file

    Returns:
        None
    """
    log.debug(
        "Creating basemap with params: "
        f"boundary={boundary} | "
        f"outfile={outfile} | "
        f"zooms={zooms} | "
        f"outdir={outdir} | "
        f"source={source} | "
        f"tms={tms}"
    )

    # Validation
    if not boundary:
        err = "You need to specify a boundary! (in-memory object or bbox)"
        log.error(err)
        raise ValueError(err)

    # Get all the zoom levels we want
    zoom_levels = list()
    if zooms:
        if zooms.find("-") > 0:
            start = int(zooms.split("-")[0])
            end = int(zooms.split("-")[1]) + 1
            x = range(start, end)
            for i in x:
                zoom_levels.append(i)
        elif zooms.find(",") > 0:
            levels = zooms.split(",")
            for level in levels:
                zoom_levels.append(int(level))
        else:
            zoom_levels.append(int(zooms))

    if not outdir:
        base = Path.cwd().absolute()
    else:
        base = Path(outdir).absolute()

    # Source / TMS validation
    if not source and not tms:
        err = "You need to specify a source!"
        log.error(err)
        raise ValueError(err)
    if source == "oam" and not tms:
        err = "A TMS URL must be provided for OpenAerialMap!"
        log.error(err)
        raise ValueError(err)
    # A custom TMS provider
    if source != "oam" and tms:
        source = "custom"

    tiledir = base / f"{source}tiles"
    # Make tile download directory
    tiledir.mkdir(parents=True, exist_ok=True)
    # Convert to string for other methods
    tiledir = str(tiledir)

    basemap = BaseMapper(boundary, tiledir, source)

    if tms:
        # Add TMS URL to sources for download
        basemap.customTMS(tms, True if source == "oam" else False, xy)

    # Args parsed, main code:
    tiles = list()
    for zoom_level in zoom_levels:
        # Download the tile directory
        basemap.getTiles(zoom_level)
        tiles += basemap.tiles

    if not outfile:
        log.info(f"No outfile specified, tile download finished: {tiledir}")
        return

    suffix = Path(outfile).suffix.lower()
    image_format = basemap.sources[source].get("suffix", "jpg")
    log.debug(f"Basemap output format: {suffix} | Image format: {image_format}")

    if any(substring in suffix for substring in ["sqlite", "mbtiles"]):
        outf = DataFile(outfile, basemap.getFormat(), append)
        if suffix == ".mbtiles":
            outf.addBounds(basemap.bbox)
            outf.addZoomLevels(zoom_levels)
        # Create output database and specify image format, png, jpg, or tif
        outf.writeTiles(tiles, tiledir, image_format)

    elif suffix == ".pmtiles":
        tile_dir_to_pmtiles(outfile, tiledir, basemap.bbox, image_format, zoom_levels, source)

    else:
        msg = f"Format {suffix} not supported"
        log.error(msg)
        raise ValueError(msg) from None
    log.info(f"Wrote {outfile}")

options: show_source: false heading_level: 3

Write PMTiles archive from tiles in the specified directory.

Parameters:

Name	Type	Description	Default
outfile	str	The output PMTiles archive file path.	required
tile_dir	str | Path	The directory containing the tile images.	required
bbox	tuple	Bounding box in format (min_lon, min_lat, max_lon, max_lat).	required
attribution	str	Attribution string to include in PMTile archive.	required

Returns:

Type	Description
	None

Source code in osm_fieldwork/basemapper.py

def tile_dir_to_pmtiles(
    outfile: str,
    tile_dir: str | Path,
    bbox: tuple,
    image_format: str,
    zoom_levels: list[int],
    attribution: str,
):
    """Write PMTiles archive from tiles in the specified directory.

    Args:
        outfile (str): The output PMTiles archive file path.
        tile_dir (str | Path): The directory containing the tile images.
        bbox (tuple): Bounding box in format (min_lon, min_lat, max_lon, max_lat).
        attribution (str): Attribution string to include in PMTile archive.

    Returns:
        None
    """
    tile_dir = Path(tile_dir)

    # Abort if no files are present
    first_file = next((file for file in tile_dir.rglob("*.*") if file.is_file()), None)
    if not first_file:
        err = "No tile files found in the specified directory. Aborting PMTile creation."
        log.error(err)
        raise ValueError(err)

    tile_format = image_format.upper()
    # NOTE JPEG exception / flexible extension (.jpg, .jpeg)
    if tile_format == "JPG":
        tile_format = "JPEG"
    log.debug(f"PMTile determind internal file format: {tile_format}")
    possible_tile_formats = [f".{e.name.lower()}" for e in PMTileType]
    possible_tile_formats.append(".jpg")
    possible_tile_formats.remove(".unknown")

    with open(outfile, "wb") as pmtile_file:
        writer = PMTileWriter(pmtile_file)

        for tile_path in tile_dir.rglob("*"):
            if tile_path.is_file() and tile_path.suffix.lower() in possible_tile_formats:
                tile_id = tileid_from_zyx_dir_path(tile_path)

                with open(tile_path, "rb") as tile:
                    writer.write_tile(tile_id, tile.read())

        min_lon, min_lat, max_lon, max_lat = bbox
        log.debug(
            f"Writing PMTiles file with min_zoom ({zoom_levels[0]}) "
            f"max_zoom ({zoom_levels[-1]}) bbox ({bbox}) tile_compression None"
        )

        # Write PMTile metadata
        writer.finalize(
            header={
                "tile_type": PMTileType[tile_format],
                "tile_compression": PMTileCompression.NONE,
                "min_zoom": zoom_levels[0],
                "max_zoom": zoom_levels[-1],
                "min_lon_e7": int(min_lon * 10000000),
                "min_lat_e7": int(min_lat * 10000000),
                "max_lon_e7": int(max_lon * 10000000),
                "max_lat_e7": int(max_lat * 10000000),
                "center_zoom": zoom_levels[0],
                "center_lon_e7": int(min_lon + ((max_lon - min_lon) / 2)),
                "center_lat_e7": int(min_lat + ((max_lat - min_lat) / 2)),
            },
            metadata={"attribution": f"© {attribution}"},
        )

options: show_source: false heading_level: 3

Helper function to get the tile id from a tile in xyz (zyx) directory structure.

TMS typically has structure z/y/x.png If the --xy flag was used previously, the TMS was downloaed into directories of z/y/x structure from their z/x/y URL.

Parameters:

Name	Type	Description	Default
filepath	Union[Path, str]	The path to tile image within the xyz directory.	required

Returns:

Name	Type	Description
int	int	The globally defined tile id from the xyz definition.

Source code in osm_fieldwork/basemapper.py

def tileid_from_zyx_dir_path(filepath: Union[Path, str]) -> int:
    """Helper function to get the tile id from a tile in xyz (zyx) directory structure.

    TMS typically has structure z/y/x.png
    If the --xy flag was used previously, the TMS was downloaed into
    directories of z/y/x structure from their z/x/y URL.

    Args:
        filepath (Union[Path, str]): The path to tile image within the xyz directory.

    Returns:
        int: The globally defined tile id from the xyz definition.
    """
    # Extract the final 3 parts from the TMS file path
    tile_image_path = Path(filepath).parts[-3:]

    try:
        final_tile = int(Path(tile_image_path[-1]).stem)
    except ValueError as e:
        msg = f"Invalid tile path (cannot parse as int): {str(tile_image_path)}"
        log.error(msg)
        raise ValueError(msg) from e

    x = final_tile
    z, y = map(int, tile_image_path[:-1])

    return zxy_to_tileid(z, x, y)

options: show_source: false heading_level: 3