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 )
–
-
end
(Location )
–
-
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