Spaces:
Sleeping
Sleeping
| from dotenv import load_dotenv | |
| from location import Location | |
| import time | |
| import pandas as pd | |
| import requests | |
| import os | |
| HOME_COORDS = [32.85195, 35.09126] | |
| load_dotenv() | |
| def call_tomtom_api(street_name=None, street_number=None, city=None, bbox=None): | |
| # avoid exceeding rate limit | |
| time.sleep(1) | |
| base_url = os.getenv("TOMTOM_URL") | |
| api_key = os.getenv("TOMTOM_API_KEY") | |
| params = { | |
| "key": api_key, | |
| "countryCode": "ISR" | |
| } | |
| if street_name: | |
| params["streetName"] = street_name | |
| if street_number: | |
| params["streetNumber"] = street_number | |
| if city: | |
| params["municipality"] = city | |
| if bbox: | |
| # Format: topLeftLat,topLeftLon:btmRightLat,btmRightLon | |
| params["boundingBox"] = f"{bbox[0]},{bbox[1]}:{bbox[2]},{bbox[3]}" | |
| try: | |
| response = requests.get(base_url, params=params) | |
| response.raise_for_status() | |
| return response | |
| except requests.exceptions.HTTPError as e: | |
| print(f"[TomTom Structured] HTTP error: {e.response.status_code} - {e.response.text}") | |
| return None | |
| except requests.exceptions.RequestException as e: | |
| print(f"[TomTom Structured] Request error: {str(e)}") | |
| return None | |
| def is_in_bbox(pos, bbox): | |
| lat, lon = pos | |
| top_left_lat, top_left_lon, bottom_right_lat, bottom_right_lon = bbox | |
| return (bottom_right_lat <= lat <= top_left_lat) and (top_left_lon <= lon <= bottom_right_lon) | |
| def parse_tomtom_response(res, bbox=False): | |
| if res is None: | |
| return None | |
| try: | |
| data = res.json() | |
| pos = data['results'][0]['position'] | |
| if bbox: | |
| bbox_points = data['results'][0]['boundingBox'] | |
| return [pos['lat'], pos['lon']], bbox_points | |
| else: | |
| return [pos['lat'], pos['lon']] | |
| except (KeyError, IndexError, ValueError) as e: | |
| print(f"[TomTom] Failed to parse response: {str(e)}") | |
| return None | |
| def addresses_to_coordinates(addresses): | |
| locations = [] | |
| for addr in addresses: | |
| loc = Location(addr) | |
| # get the city's central location and bounding box | |
| result, bbox = parse_tomtom_response(call_tomtom_api(city=loc.city), bbox=True) | |
| if result: | |
| locations.append(loc) | |
| loc.set_coords(*result) | |
| loc.set_city_bbox(tl_lat=bbox['topLeftPoint']['lat'], tl_lon=bbox['topLeftPoint']['lon'], | |
| br_lat=bbox['btmRightPoint']['lat'], br_lon=bbox['btmRightPoint']['lon']) | |
| else: | |
| print(f"[TomTom] Failed address {addr}.") | |
| # after we have the city bbox, get the address location and check if it is in the bbox | |
| result = parse_tomtom_response(call_tomtom_api(street_name=loc.street, | |
| street_number=loc.building, | |
| city=loc.city, | |
| bbox=loc.city_bbox)) | |
| if result: | |
| if is_in_bbox(result, loc.city_bbox): | |
| loc.set_coords_level('address') | |
| loc.set_coords(*result) | |
| else: | |
| continue | |
| return locations | |
| def call_graphhopper_api(coords): | |
| if not coords: | |
| return {"error": "No coordinates provided"} | |
| lat, lon = HOME_COORDS | |
| services = [ | |
| { | |
| "id": f"s{idx}", | |
| "type": "service", | |
| "address": { | |
| "location_id": f"s{idx}", | |
| "lat": service_lat, | |
| "lon": service_lon | |
| }, | |
| "duration": 120 | |
| } | |
| for idx, (service_lat, service_lon) in enumerate(coords) | |
| ] | |
| body = { | |
| "configuration": { | |
| "routing": { | |
| "calc_points": True, | |
| "return_snapped_waypoints": True | |
| } | |
| }, | |
| "objectives": [{"type": "min", "value": "completion_time"}], | |
| "vehicles": [{ | |
| "vehicle_id": "v1", | |
| "type_id": "custom_vehicle_type", | |
| "start_address": { | |
| "location_id": "v1", | |
| "lat": lat, | |
| "lon": lon | |
| }, | |
| "return_to_depot": True | |
| }], | |
| "vehicle_types": [{"type_id": "custom_vehicle_type", "profile": "car"}], | |
| "services": services | |
| } | |
| api_key = os.getenv("GRAPHHOPPER_API_KEY") | |
| post_url = f"https://graphhopper.com/api/1/vrp/optimize?key={api_key}" | |
| try: | |
| post_res = requests.post(post_url, json=body) | |
| post_res.raise_for_status() | |
| job_id = post_res.json().get("job_id") | |
| if not job_id: | |
| return {"error": "No job_id in response", "response": post_res.text} | |
| get_url = f"https://graphhopper.com/api/1/vrp/solution/{job_id}?key={api_key}" | |
| get_res = requests.get(get_url) | |
| get_res.raise_for_status() | |
| return get_res | |
| except requests.exceptions.HTTPError as e: | |
| return {"error": f"GraphHopper HTTP error: {e.response.status_code}", "response": e.response.text} | |
| except requests.exceptions.RequestException as e: | |
| return {"error": f"GraphHopper request failed: {str(e)}"} | |
| def parse_graphhopper_response(res): | |
| if isinstance(res, dict) and "error" in res: | |
| print(f"[GraphHopper] Error: {res['error']}") | |
| if "response" in res: | |
| print(f"[GraphHopper] Response: {res['response']}") | |
| return pd.DataFrame() | |
| try: | |
| json_data = res.json() | |
| route = json_data['solution']['routes'][0]['activities'] | |
| return pd.DataFrame([{ | |
| "point_type": step["type"], | |
| "location_id": step["address"]["location_id"], | |
| "lat": step["address"]["lat"], | |
| "lon": step["address"]["lon"] | |
| } for step in route]) | |
| except (KeyError, IndexError, ValueError) as e: | |
| print(f"[GraphHopper] Failed to parse solution: {str(e)}") | |
| return pd.DataFrame() | |
| def coordinates_to_route(coords): | |
| res = call_graphhopper_api(coords) | |
| return parse_graphhopper_response(res) | |