Spaces:
Build error
Build error
| import os | |
| import pandas as pd | |
| from dotenv import load_dotenv | |
| import requests | |
| from geopy.distance import geodesic | |
| load_dotenv() | |
| def find_place(input_text, api_key): | |
| # Define the base URL and parameters for the request | |
| base_url = "https://maps.googleapis.com/maps/api/place/findplacefromtext/json" | |
| params = { | |
| "fields": "formatted_address,name,rating,opening_hours,geometry", | |
| "input": input_text, | |
| "inputtype": "textquery", | |
| "key": api_key # Replace with your actual API key | |
| } | |
| # Make the GET request | |
| response = requests.get(base_url, params=params) | |
| # Check if the request was successful | |
| if response.status_code == 200: | |
| # Return the response content (JSON) | |
| return response.json() | |
| else: | |
| # Return an error message if the request failed | |
| return f"Request failed with status code: {response.status_code}" | |
| def get_zip_code(address): | |
| base_url = "https://maps.googleapis.com/maps/api/geocode/json" | |
| params = { | |
| "address": address, | |
| "key": "AIzaSyDXCm-fwucr6R93CHa9Bb4j5qArup5D4kY" | |
| } | |
| response = requests.get(base_url, params=params) | |
| if response.status_code == 200: | |
| data = response.json() | |
| if data['status'] == 'OK': | |
| # Extract the postal code from the address components | |
| for component in data['results'][0]['address_components']: | |
| if 'postal_code' in component['types']: | |
| return component['long_name'] | |
| return None | |
| else: | |
| return f"API request returned status: {data['status']}" | |
| else: | |
| return f"Request failed with status code: {response.status_code}" | |
| def get_distances_for_multiple_destinations(origin, destinations, api_key): | |
| destinations_param = '|'.join(destinations) | |
| base_url = "https://maps.googleapis.com/maps/api/distancematrix/json" | |
| params = { | |
| "origins": origin, | |
| "destinations": destinations_param, | |
| "key": api_key | |
| } | |
| response = requests.get(base_url, params=params) | |
| if response.status_code == 200: | |
| data = response.json() | |
| if data['status'] == 'OK': | |
| results = [] | |
| for element in data['rows'][0]['elements']: | |
| if element['status'] == 'OK': | |
| distance = element['distance']['text'] | |
| results.append(distance) | |
| else: | |
| results.append("Distance calculation failed") | |
| return results | |
| else: | |
| return ["API request returned error: " + data['status']] | |
| else: | |
| return [f"Request failed with status code: {response.status_code}"] | |
| class LocationDataHandler: | |
| def __init__(self, data_frame): | |
| # self.data_file = data_file | |
| self.api_key = os.getenv('GOOGLE_MAPS_API_KEY') | |
| self.data_frame = data_frame | |
| def get_latlong_from_address(self, address): | |
| try: | |
| result = find_place(address, self.api_key) | |
| lat = result['candidates'][0]['geometry']['location']['lat'] | |
| lng = result['candidates'][0]['geometry']['location']['lng'] | |
| return lat, lng | |
| except Exception as e: | |
| print(f"Error retrieving location for {address}: {e}") | |
| return None | |
| def calculate_distance(self, ref_point, latitude, longitude): | |
| try: | |
| location = (latitude, longitude) | |
| return geodesic(ref_point, location).kilometers | |
| except Exception as e: | |
| print(f"Error calculating distance: {e}") | |
| return None | |
| def filter_by_address(self, address, max_distance_km=30): | |
| ref_point = self.get_latlong_from_address(address) | |
| if not ref_point: | |
| return None | |
| def distance_calculation(row): | |
| return self.calculate_distance(ref_point, row['Latitude'], row['Longitude']) | |
| # Calculate distance for each row | |
| self.data_frame['Distance'] = self.data_frame.apply(distance_calculation, axis=1) | |
| # Select top 10 within the maximum distance | |
| df = self.data_frame[self.data_frame['Distance'] <= max_distance_km].sort_values(by='Distance').head(10).reset_index(drop=True) | |
| # Calculate Actual Distance Using GMaps API | |
| destinations = df.apply(lambda row: f"{row['Latitude']},{row['Longitude']}", axis=1).tolist() | |
| distances = get_distances_for_multiple_destinations(f"{ref_point[0]},{ref_point[1]}", destinations, self.api_key) | |
| # Add the distances back to the DataFrame | |
| df['Distance'] = distances | |
| return df |