Spaces:
Sleeping
Sleeping
import logging | |
import pandas as pd | |
import os | |
from typing import Optional, Dict, Any | |
import gradio as gr | |
import googlemaps | |
from PIL import Image | |
from langchain.utilities.google_places_api import GooglePlacesAPIWrapper | |
import plotly.graph_objects as go | |
import requests | |
from PIL import Image | |
from io import BytesIO | |
import tempfile | |
class GooglePlacesAPIWrapperExtended(GooglePlacesAPIWrapper): | |
api_key = os.environ["GPLACES_API_KEY"] | |
def __init__(self, **kwargs): | |
super().__init__(**kwargs) | |
def run(self, query: str, **kwargs) -> pd.DataFrame: | |
"""Run Places search and get k number of places that exist that match.""" | |
search_results = self.google_map_client.places(query, **kwargs)["results"] | |
num_to_return = len(search_results) | |
places = [] | |
if num_to_return == 0: | |
return pd.DataFrame(columns=["Name", "Address", "Phone Number", "Website", | |
"Opening Hours", "Is Open Now", "latitude", "longitude", | |
"Summary", "Rating", "Image", "Reviews"]) | |
num_to_return = ( | |
num_to_return | |
if self.top_k_results is None | |
else min(num_to_return, self.top_k_results) | |
) | |
for i in range(num_to_return): | |
result = search_results[i] | |
details = self.fetch_place_details(result["place_id"]) | |
if details is not None: | |
places.append(details) | |
return pd.DataFrame(places) | |
def fetch_place_details(self, place_id: str) -> Optional[Dict[str, Any]]: | |
try: | |
place_details = self.google_map_client.place(place_id) | |
formatted_details = self.format_place_details(place_details) | |
return formatted_details | |
except Exception as e: | |
logging.error(f"An Error occurred while fetching place details: {e}") | |
return None | |
def format_place_details(self, place_details: Dict[str, Any]) -> Optional[Dict[str, Any]]: | |
try: | |
name = place_details.get("result", {}).get("name", "Unknown") | |
address = place_details.get("result", {}).get("formatted_address", "Unknown") | |
phone_number = place_details.get("result", {}).get("formatted_phone_number", "Unknown") | |
website = place_details.get("result", {}).get("website", "Unknown") | |
weekday_text = place_details.get("result", {}).get("opening_hours", {}).get("weekday_text", []) | |
is_open = place_details.get("result", {}).get("opening_hours", {}).get("open_now", "Unknown") | |
location = place_details.get("result", {}).get("geometry", {}).get("location", {}) | |
latitude = location.get("lat", "Unknown") | |
longitude = location.get("lng", "Unknown") | |
summary = place_details.get("result", {}).get("editorial_summary", {}).get("overview", "Unknown") | |
rating = place_details.get("result", {}).get("rating", "Unknown") | |
image = place_details.get("result", {}).get("photos", [{}])[0].get("photo_reference", "Unknown") | |
image_url = f"https://maps.googleapis.com/maps/api/place/photo?maxwidth=400&photoreference={image}&key={self.api_key}" | |
first_three_reviews = place_details.get("result", {}).get("reviews", [])[:3] | |
formatted_details = { | |
"name": name, | |
"address": address, | |
"phone_number": phone_number, | |
"website": website, | |
"opening_hours": weekday_text, | |
"is_open_now": is_open, | |
"latitude": latitude, | |
"longitude": longitude, | |
"summary": summary, | |
"rating": rating, | |
"image": image_url, | |
"reviews": first_three_reviews | |
} | |
return formatted_details | |
except Exception as e: | |
logging.error(f"An error occurred while formatting place details: {e}") | |
return None | |
#pd.set_option("display.max_columns", None) | |
#pd.set_option("display.max_rows", None) | |
#gplaceapi = GooglePlacesAPIWrapperExtended() | |
#query = "Louvre, Paris" | |
#result_df = gplaceapi.run(query) | |
#print(result_df) | |
#query = gr.inputs.Textbox(lines=2, label="Query") | |
#result_df = gr.outputs.Dataframe(type="pandas") | |
#gr.Interface(fn=GooglePlacesAPIWrapperExtended().run, inputs=query, outputs=result_df).launch(debug=True) | |
def filter_map(locations): | |
dataframe = pd.DataFrame() | |
for location in locations: | |
dataframe = pd.concat([dataframe, GooglePlacesAPIWrapperExtended().run(location)]) | |
names = dataframe["name"].tolist() | |
summaries = dataframe["summary"].tolist() | |
image_urls = dataframe["image"].tolist() | |
fig = go.Figure(go.Scattermapbox( | |
lat=dataframe['latitude'].tolist(), | |
lon=dataframe['longitude'].tolist(), | |
mode='markers', | |
marker=go.scattermapbox.Marker( | |
size=13, | |
color='rgb(255, 123, 0)', | |
), | |
hovertemplate='Name: %{customdata[0]}<br>Summary: %{customdata[1]}', | |
customdata=list(zip(names, summaries)), | |
name='Places' | |
)) | |
fig.update_layout( | |
mapbox_style="open-street-map", | |
hovermode='closest', | |
mapbox=dict( | |
bearing=0, | |
center=go.layout.mapbox.Center( | |
lat=dataframe['latitude'].tolist()[0], | |
lon=dataframe['longitude'].tolist()[0] | |
), | |
pitch=0, | |
zoom=12 | |
), | |
) | |
# Add images using layout.images attribute | |
#for i, url in enumerate(image_urls): | |
# response = requests.get(url) | |
# img = Image.open(BytesIO(response.content)) | |
# with tempfile.NamedTemporaryFile(delete=False, suffix=".png") as temp: | |
# img.save(temp.name) | |
# fig.add_layout_image( | |
# dict( | |
# source=temp.name, | |
# xref='x', | |
# yref='y', | |
# x=dataframe['longitude'].iloc[i], | |
# y=dataframe['latitude'].iloc[i], | |
# sizex=0.05, | |
# sizey=0.05, | |
# sizing='stretch', | |
# opacity=0.7, | |
# layer='above' | |
# ) | |
# ) | |
# | |
#fig.update_layout( | |
# xaxis=dict(range=[dataframe['longitude'].min(), dataframe['longitude'].max()]), | |
# yaxis=dict(range=[dataframe['latitude'].min(), dataframe['latitude'].max()]) | |
#) | |
# | |
return fig, dataframe | |
if __name__ == "main": | |
with gr.Blocks() as demo: | |
with gr.Column(): | |
location = gr.Textbox(lines=2, label="Location") | |
btn = gr.Button(value="Update Filter") | |
map = gr.Plot().style() | |
result_df = gr.Dataframe(type="pandas") | |
btn.click(filter_map, [location], [map, result_df]) | |
demo.queue(concurrency_count=6).launch() |