From 5b0a0768cb340fa053c3732ed6e4cc7a21df135e Mon Sep 17 00:00:00 2001 From: Amy Bowersox Date: Tue, 18 Aug 2020 12:49:20 -0600 Subject: [PATCH] added transit time calculations to DCalc and Batch-Distance --- src/batch_distance.py | 52 ++++++++++++++++++++++++++++++++++++++++++ src/dcalc.py | 53 +++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 105 insertions(+) diff --git a/src/batch_distance.py b/src/batch_distance.py index 8b5f687..682e707 100755 --- a/src/batch_distance.py +++ b/src/batch_distance.py @@ -66,6 +66,53 @@ def classify_distance(dist): return 'GREEN' +def transit_time_google(config, org, dest): + apikey = config['google']['apikey'] + if not apikey: + raise GeocodingError("Google API key not specified") + str_origin = f"{org[0]},{org[1]}" + str_dest = f"{dest[0]},{dest[1]}" + query = {'key': apikey, 'origin': str_origin, 'destination': str_dest, 'mode': 'driving', 'region': 'us'} + url = 'https://maps.googleapis.com/maps/api/directions/json?' + \ + urllib.parse.urlencode(query, quote_via=urllib.parse.quote) + with urllib.request.urlopen(url) as response: + if response.status == 200: + apireturn = json.loads(response.read()) + stat = apireturn['status'] + if stat == 'OK': + best_time = None + best_summary = None + for route in apireturn['routes']: + summary = route['summary'] + current_time = 0 + for leg in route['legs']: + current_time += leg['duration']['value'] + if best_time is None or current_time < best_time: + best_time = current_time + best_summary = summary + if best_time is None: + return None + return best_time, best_summary + elif stat == 'ZERO_RESULTS': + return None + else: + raise GeocodingError(f"Google API returns status of {stat}") + else: + raise GeocodingError(f"Google API returns {response.status} HTTP status code") + + +def stringize_seconds(total_secs): + secs = total_secs % 60 + remainder = total_secs // 60 + if remainder == 0: + return f"{secs}s" + mins = remainder % 60 + hours = remainder // 60 + if hours == 0: + return f"{mins}m{secs}s" + return f"{hours}h{mins}m{secs}s" + + cmdline_parser = argparse.ArgumentParser() cmdline_parser.add_argument('input', help='The input file containing locations to be checked.') cmdline_parser.add_argument('output', nargs='?', help='The output file name that will receive the processed data.') @@ -90,6 +137,11 @@ def main(args): if coords: dist = distance_miles(OFFICE_LOCATION, coords) out_row = [in_row[0], in_row[1], coords[0], coords[1], dist, classify_distance(dist)] + transit_time = transit_time_google(config, coords, OFFICE_LOCATION) + if transit_time: + out_row += [transit_time[0], stringize_seconds(transit_time[0]), transit_time[1]] + else: + out_row += ['*** Unable to calculate transit time'] else: out_row = [in_row[0], in_row[1], '*** Unable to locate'] writer.writerow(out_row) diff --git a/src/dcalc.py b/src/dcalc.py index f73dc59..a36890f 100755 --- a/src/dcalc.py +++ b/src/dcalc.py @@ -53,6 +53,53 @@ def distance_miles(point1, point2): return RADIUS * c / METERS_PER_MILE +def transit_time_google(config, org, dest): + apikey = config['google']['apikey'] + if not apikey: + raise GeocodingError("Google API key not specified") + str_origin = f"{org[0]},{org[1]}" + str_dest = f"{dest[0]},{dest[1]}" + query = {'key': apikey, 'origin': str_origin, 'destination': str_dest, 'mode': 'driving', 'region': 'us'} + url = 'https://maps.googleapis.com/maps/api/directions/json?' + \ + urllib.parse.urlencode(query, quote_via=urllib.parse.quote) + with urllib.request.urlopen(url) as response: + if response.status == 200: + apireturn = json.loads(response.read()) + stat = apireturn['status'] + if stat == 'OK': + best_time = None + best_summary = None + for route in apireturn['routes']: + summary = route['summary'] + current_time = 0 + for leg in route['legs']: + current_time += leg['duration']['value'] + if best_time is None or current_time < best_time: + best_time = current_time + best_summary = summary + if best_time is None: + return None + return best_time, best_summary + elif stat == 'ZERO_RESULTS': + return None + else: + raise GeocodingError(f"Google API returns status of {stat}") + else: + raise GeocodingError(f"Google API returns {response.status} HTTP status code") + + +def stringize_seconds(total_secs): + secs = total_secs % 60 + remainder = total_secs // 60 + if remainder == 0: + return f"{secs}s" + mins = remainder % 60 + hours = remainder // 60 + if hours == 0: + return f"{mins}m{secs}s" + return f"{hours}h{mins}m{secs}s" + + cmdline_parser = argparse.ArgumentParser() cmdline_parser.add_argument('address', nargs='+', help='The address to be calculated') cmdline_parser.add_argument('-C', '--config', default='geoapi.ini', help='The geocoding API configuration file') @@ -69,6 +116,12 @@ def main(args): print(f"Coordinates: Latitude {coords[0]}, longitude {coords[1]}") dist = distance_miles(OFFICE_LOCATION, coords) print(f"Distance from Boulder office: {dist} miles") + transit_time = transit_time_google(config, coords, OFFICE_LOCATION) + if transit_time: + str_time = stringize_seconds(transit_time[0]) + print(f"Estimated transit time: {str_time} - route: {transit_time[1]} ({transit_time[0]} seconds)") + else: + print("Unable to calculate transit time.") else: print("Location was not found.")