covermap.py
async def fetch_uplinks(h3_index, session):
    """Fetch H3 hex coverage based on Mappers uplinks."""
    mappers_coverage = False
    mapper_urls = [
        MAPPERS_URL + h for h in h3.k_ring(h3.h3_to_center_child(h3_index))
    ]
    for mapper_url in mapper_urls:
        while rate_limit and not limiter.consume("mappers", 1):
            time.sleep(0.01)
        async with session.get(mapper_url) as response:
            if response.status == 200:
                uplinks = (await response.json())['uplinks']
                mappers_coverage |= len(uplinks) > 0
            elif response.status == 500:
                logger_tenacity.warning(
                    "response status 500 for %s, assuming no coverage",
                    mapper_url)
                mappers_coverage |= False
            else:
                response.raise_for_status()
        if mappers_coverage:
            break
    return [h3_index, mappers_coverage, datetime.datetime.utcnow()]
Fetch H3 hex coverage based on Mappers uplinks.
Fetch H3 hex coverage based on Mappers uplinks.
Fetch H3 hex coverage based on Mappers uplinks .
async def fetch_uplinks(h3_index, session):
    mappers_coverage = False
    mapper_urls = [
        MAPPERS_URL + h for h in h3.k_ring(h3.h3_to_center_child(h3_index))
    ]
    for mapper_url in mapper_urls:
        while rate_limit and not limiter.consume("mappers", 1):
            time.sleep(0.01)
        async with session.get(mapper_url) as response:
            if response.status == 200:
                uplinks = (await response.json())['uplinks']
                mappers_coverage |= len(uplinks) > 0
            elif response.status == 500:
                logger_tenacity.warning(
                    "response status 500 for %s, assuming no coverage",
                    mapper_url)
                mappers_coverage |= False
            else:
                response.raise_for_status()
        if mappers_coverage:
            break
    return [h3_index, mappers_coverage, datetime.datetime.utcnow()]
async def fetch_uplinks ( h3_index , session ) : mappers_coverage = False mapper_urls = [ MAPPERS_URL + h for h in h3 . k_ring ( h3 . h3_to_center_child ( h3_index ) ) ] for mapper_url in mapper_urls : while rate_limit and not limiter . consume ( "mappers" , 1 ) : time . sleep ( 0.01 ) async with session . get ( mapper_url ) as response : if response . status == 200 : uplinks = ( await response . json ( ) ) [ 'uplinks' ] mappers_coverage |= len ( uplinks ) > 0 elif response . status == 500 : logger_tenacity . warning ( "response status 500 for %s, assuming no coverage" , mapper_url ) mappers_coverage |= False else : response . raise_for_status ( ) if mappers_coverage : break return [ h3_index , mappers_coverage , datetime . datetime . utcnow ( ) ]
Fetch H3 hex coverage based on Mappers uplinks.
Fetch H3 hex coverage based on Mappers uplinks .
"""Fetch H3 hex coverage based on Mappers uplinks."""
h3_index
session