Spaces:
Running
Running
| import requests | |
| from src.utils import extract_before_parenthesis | |
| import os | |
| from requests.exceptions import SSLError | |
| import time | |
| import sys | |
| import pandas as pd | |
| import numpy as np | |
| class GoogleDistanceMatrix: | |
| def __init__(self, subscription_key: str="") -> None: | |
| self.gplaces_api_key: str = subscription_key | |
| self.data = pd.read_csv('/home/user/app/database/googleDistanceMatrix/distance.csv') | |
| print("GoogleDistanceMatrix loaded.") | |
| def run(self, origin, destination, mode='driving'): | |
| origin = extract_before_parenthesis(origin) | |
| destination = extract_before_parenthesis(destination) | |
| info = {"origin": origin, "destination": destination,"cost": None, "duration": None, "distance": None} | |
| response = self.data[(self.data['origin'] == origin) & (self.data['destination'] == destination)] | |
| if len(response) > 0: | |
| if response['duration'].values[0] is None or response['distance'].values[0] is None or response['duration'].values[0] is np.nan or response['distance'].values[0] is np.nan: | |
| return "No valid information." | |
| info["duration"] = response['duration'].values[0] | |
| info["distance"] = response['distance'].values[0] | |
| if 'driving' in mode: | |
| info["cost"] = int(eval(info["distance"].replace("km","").replace(",","")) * 0.05) | |
| elif mode == "taxi": | |
| info["cost"] = int(eval(info["distance"].replace("km","").replace(",",""))) | |
| if 'day' in info["duration"]: | |
| return "No valid information." | |
| return f"{mode}, from {origin} to {destination}, duration: {info['duration']}, distance: {info['distance']}, cost: {info['cost']}" | |
| return f"{mode}, from {origin} to {destination}, no valid information." | |
| def run_for_evaluation(self, origin, destination, mode='driving'): | |
| origin = extract_before_parenthesis(origin) | |
| destination = extract_before_parenthesis(destination) | |
| info = {"origin": origin, "destination": destination,"cost": None, "duration": None, "distance": None} | |
| response = self.data[(self.data['origin'] == origin) & (self.data['destination'] == destination)] | |
| if len(response) > 0: | |
| if response['duration'].values[0] is None or response['distance'].values[0] is None or response['duration'].values[0] is np.nan or response['distance'].values[0] is np.nan: | |
| return info | |
| info["duration"] = response['duration'].values[0] | |
| info["distance"] = response['distance'].values[0] | |
| if 'day' not in info["duration"]: | |
| if 'driving' in mode: | |
| info["cost"] = int(eval(info["distance"].replace("km","").replace(",","")) * 0.05) | |
| elif mode == "taxi": | |
| info["cost"] = int(eval(info["distance"].replace("km","").replace(",",""))) | |
| return info | |
| return info | |
| def run_online(self, origin, destination, mode="driving"): | |
| # mode in ['driving','taxi','walking', 'distance','transit'] | |
| endpoint = "https://maps.googleapis.com/maps/api/distancematrix/json" | |
| params = { | |
| "origins": origin, | |
| "destinations": destination, | |
| "mode": mode if mode=="taxi" else "driving", | |
| "key": self.gplaces_api_key | |
| } | |
| while True: | |
| try: | |
| response = requests.get(endpoint, params=params) | |
| break | |
| except SSLError: | |
| time.sleep(30) | |
| data = response.json() | |
| info = {"origin": origin, "destination": destination,"cost": None, "duration": None, "distance": None} | |
| if data['status'] == "OK": | |
| element = data['rows'][0]['elements'][0] | |
| if element['status'] == "OK": | |
| info["duration"] = element['duration']['text'] | |
| info["distance"] = element['distance']['text'] | |
| if 'driving' in mode: | |
| info["cost"] = int(eval(info["distance"].replace("km","").replace(",","")) * 0.05) | |
| elif mode == "taxi": | |
| info["cost"] = int(eval(info["distance"].replace("km","").replace(",",""))) | |
| # if 'day' in info["duration"]: | |
| # return "No valid information." | |
| return f"{mode}, from {origin} to {destination}, duration: {info['duration']}, distance: {info['distance']}, cost: {info['cost']}" | |
| return "No valid information." | |
| def run_for_annotation(self, origin, destination, mode="driving"): | |
| # mode in ['driving','taxi','walking', 'distance','transit'] | |
| origin = extract_before_parenthesis(origin) | |
| destination = extract_before_parenthesis(destination) | |
| info = {"cost": None, "duration": None, "distance": None} | |
| response = self.data[(self.data['origin'] == origin) & (self.data['destination'] == destination)] | |
| if len(response) > 0: | |
| if response['duration'].values[0] is None or response['distance'].values[0] is None or response['duration'].values[0] is np.nan or response['distance'].values[0] is np.nan: | |
| return info | |
| info["duration"] = response['duration'].values[0] | |
| info["distance"] = response['distance'].values[0] | |
| if 'day' not in info["duration"]: | |
| if 'driving' in mode: | |
| info["cost"] = int(eval(info["distance"].replace("km","").replace(",","")) * 0.05) | |
| elif mode == "taxi": | |
| info["cost"] = int(eval(info["distance"].replace("km","").replace(",",""))) | |
| return info | |
| return {"duration": "N/A", "distance": "N/A", "cost": "N/A", "Hint":"Please check the input."} | |
| def run_for_build_database(self, origin, destination): | |
| # mode in ['driving','taxi','walking', 'distance','transit'] | |
| endpoint = "https://maps.googleapis.com/maps/api/distancematrix/json" | |
| params = { | |
| "origins": extract_before_parenthesis(origin), | |
| "destinations": extract_before_parenthesis(destination), | |
| "mode": "driving", | |
| "key": self.gplaces_api_key | |
| } | |
| # response = requests.get(endpoint, params=params) | |
| while True: | |
| try: | |
| response = requests.get(endpoint, params=params) | |
| break | |
| except: | |
| # print the error | |
| error = sys.exc_info()[0] | |
| print(error) | |
| time.sleep(30) | |
| data = response.json() | |
| info = {"origin": extract_before_parenthesis(origin), "destination": extract_before_parenthesis(destination),"cost": None, "duration": None, "distance": None} | |
| if data['status'] == "OK": | |
| element = data['rows'][0]['elements'][0] | |
| if element['status'] == "OK": | |
| info["duration"] = element['duration']['text'] | |
| info["distance"] = element['distance']['text'] | |
| return info |