Demo script for requesting routes from PolarRouteServer API using Python standard library.

make_request(type, url, endpoint, headers, body=None)

Sends HTTP request, prints details and returns response.

Parameters:
  • type (str) –

    HTTP request type, e.g. "GET" or "POST"

  • url (str) –

    base url to send request to

  • endpoint (str) –

    endpoint, e.g. "/api/route/some-id"

  • headers (dict) –

    HTTP headers

  • body (dict, default: None ) –

    HTTP request body. Defaults to None.

Returns:
  • HTTPResponse

    http.client.HTTPResponse

Source code in polarrouteserver/demo.py
def make_request(
    type: str, url: str, endpoint: str, headers: dict, body: dict = None
) -> http.client.HTTPResponse:
    """Sends HTTP request, prints details and returns response.

    Args:
        type (str): HTTP request type, e.g. "GET" or "POST"
        url (str): base url to send request to
        endpoint (str): endpoint, e.g. "/api/route/some-id"
        headers (dict): HTTP headers
        body (dict, optional): HTTP request body. Defaults to None.

    Returns:
        http.client.HTTPResponse
    """
    sending_str = f"Sending {type} request to {url}{endpoint}: \nHeaders: {headers}\n"

    if body:
        sending_str += f"Body: {body}\n"

    print(sending_str)

    # data = parse.urlencode(body).encode("utf-8") if body else None
    req = request.Request(url + endpoint, data=body, headers=headers)
    unverified_context = ssl._create_unverified_context()
    response = request.urlopen(req, context=unverified_context)

    print(f"Response: {response.status} {response.reason}")

    return json.loads(response.read()), response.status

parse_location(location)

Parameters:
  • location (str) –

    a location either as the name of a standard location or latitude,longitude separated by a comma, e.g. -56.7,-65.01

Returns:
  • Location

    a Location object

Source code in polarrouteserver/demo.py
def parse_location(location: str) -> Location:
    """
    Args:
     location (str): a location either as the name of a standard location or latitude,longitude separated by a comma, e.g. -56.7,-65.01

    Returns:
        a Location object
    """
    pattern = r"[+-]?([0-9]*[.])?[0-9]+,[+-]?([0-9]*[.])?[0-9]+"
    if location in STANDARD_LOCATIONS.keys():
        standard_location = STANDARD_LOCATIONS.get(location)
        return standard_location
    elif re.search(pattern, location):
        coords = re.search(pattern, location).group().split(",")
        return Location(float(coords[0]), float(coords[1]))
    else:
        raise ValueError(
            f"Expected input as the name of a standard location or latitude,longitude separated by a comma, e.g. -56.7,-65.01, got {location}"
        )

request_route(url, start, end, status_update_delay=30, num_requests=10, force_recalculation=False, mesh_id=None)

Requests a route from polarRouteServer, repeating the request for status until the route is available.

Parameters:
  • url (str) –

    Base URL to send request to.

  • start (Location) –

    Start location of route

  • end (Location) –

    End location of route

  • status_update_delay (int, default: 30 ) –

    Delay in seconds between each status request. Defaults to 10.

  • num_requests (int, default: 10 ) –

    Max number of status requests before giving up. Defaults to 10.

  • force_recalculation (bool, default: False ) –

    Force recalculation of an already existing route. Default: False.

Raises:
  • Exception

    If no status URL is returned.

Returns:
  • str( str ) –

    JSON response of route request.

Source code in polarrouteserver/demo.py
def request_route(
    url: str,
    start: Location,
    end: Location,
    status_update_delay: int = 30,
    num_requests: int = 10,
    force_recalculation: bool = False,
    mesh_id: int = None,
) -> str:
    """Requests a route from polarRouteServer, repeating the request for status until the route is available.

    Args:
        url (str): Base URL to send request to.
        start (Location): Start location of route
        end (Location): End location of route
        status_update_delay (int, optional): Delay in seconds between each status request. Defaults to 10.
        num_requests (int, optional): Max number of status requests before giving up. Defaults to 10.
        force_recalculation (bool, optional): Force recalculation of an already existing route. Default: False.

    Raises:
        Exception: If no status URL is returned.

    Returns:
        str: JSON response of route request.
    """

    # make route request
    response_body, status = make_request(
        "POST",
        url,
        "/api/route",
        {"Content-Type": "application/json"},
        json.dumps(
            {
                "start_lat": start.lat,
                "start_lon": start.lon,
                "end_lat": end.lat,
                "end_lon": end.lon,
                "start_name": start.name,
                "end_name": end.name,
                "force_recalculate": force_recalculation,
                "mesh_id": mesh_id,
            },
        ).encode("utf-8"),
    )

    print(pprint.pprint(response_body))

    if not str(status).startswith("2"):
        return None

    # if route is returned
    if response_body.get("json") is not None:
        return response_body["json"]

    # if no route returned, request status at status-url
    status_url = response_body.get("status-url")
    if status_url is None:
        raise Exception("No status URL returned.")
    id = response_body.get("id")

    status_request_count = 0
    while status_request_count <= num_requests:
        status_request_count += 1
        print(
            f"\nWaiting for {status_update_delay} seconds before sending status request."
        )
        time.sleep(status_update_delay)

        # make route request
        print(f"Status request #{status_request_count} of {num_requests}")
        response_body, status = make_request(
            "GET",
            url,
            f"/api/route/{id}",
            headers={"Content-Type": "application/json"},
        )

        print(f"Route calculation {response_body.get('status')}.")
        print(pprint.pprint(response_body))
        if response_body.get("status") == "PENDING":
            continue
        elif response_body.get("status") == "FAILURE":
            return None
        elif response_body.get("status") == "SUCCESS":
            return response_body.get("json")
    print(
        f'Max number of requests sent. Quitting.\nTo send more status requests, run: "curl {url}/api/route/{id}"'
    )
    return None