File size: 7,321 Bytes
9be4956 6159f52 9be4956 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 |
import requests
from annotation.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
os.environ["http_proxy"] = "http://127.0.0.1:7890"
os.environ["https_proxy"] = "http://127.0.0.1:7890"
class GoogleDistanceMatrix:
def __init__(self, subscription_key: str="AIzaSyClbhCKqk2QS97jadE0RQ3qYCAteY7sK4I") -> 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']
endpoint = "https://maps.googleapis.com/maps/api/distancematrix/json"
params = {
"origins": extract_before_parenthesis(origin),
"destinations": extract_before_parenthesis(destination),
"mode": mode if mode!="taxi" else "driving",
"key": self.gplaces_api_key
}
response = requests.get(endpoint, params=params)
data = response.json()
info = {}
if data['status'] == "OK":
element = data['rows'][0]['elements'][0]
if element['status'] == "OK":
info["duration"] = element['duration']['text']
info["distance"] = element['distance']['text']
info["cost"] = None
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(",","")))
else:
info = {"duration": "N/A", "distance": "N/A", "cost": "N/A", "Hint":"Please check the input."}
return info
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 |