Requests a route from polarRouteServer, monitors job status until complete, then retrieves route data.
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_new_route
(bool , default:
False
)
–
Force recalculation of an already existing route. Default: False.
-
mesh_id
(int , default:
None
)
–
Custom mesh ID to use for route calculation. Default: None.
|
Raises: |
-
Exception
–
If no status URL is returned.
|
Returns: |
-
str ( str
) –
JSON response of route data, or None if request failed.
|
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_new_route: bool = False,
mesh_id: int = None,
) -> str:
"""Requests a route from polarRouteServer, monitors job status until complete, then retrieves route data.
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_new_route (bool, optional): Force recalculation of an already existing route. Default: False.
mesh_id (int, optional): Custom mesh ID to use for route calculation. Default: None.
Raises:
Exception: If no status URL is returned.
Returns:
str: JSON response of route data, or None if request failed.
"""
# 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_new_route": force_new_route,
"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:
print(
"No status URL returned. Route may have failed or been returned immediately."
)
return None
job_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 job status request
print(f"Status request #{status_request_count} of {num_requests}")
status_response, status_code = make_request(
"GET",
url,
f"/api/job/{job_id}",
headers={"Content-Type": "application/json"},
)
print(f"Route calculation {status_response.get('status')}.")
print(pprint.pprint(status_response))
if status_response.get("status") == "PENDING":
continue
elif status_response.get("status") == "FAILURE":
return None
elif status_response.get("status") == "SUCCESS":
# Job is complete, now get the actual route data
route_url = status_response.get("route_url")
if route_url:
# Extract route ID from the route_url (e.g., "/api/route/123")
route_id = route_url.split("/")[-1]
print(f"Job complete! Fetching route data from route ID: {route_id}")
route_response, route_status = make_request(
"GET",
url,
f"/api/route/{route_id}",
headers={"Content-Type": "application/json"},
)
if str(route_status).startswith("2"):
return route_response
else:
print(f"Failed to fetch route data: {route_status}")
return None
else:
print("Job completed but no route_url provided")
return None
print(
f'Max number of requests sent. Quitting.\nTo send more status requests, run: "curl {url}/api/job/{job_id}"'
)
return None