|
|
import os |
|
|
import json |
|
|
import gradio as gr |
|
|
import requests |
|
|
import openai |
|
|
import logging |
|
|
from gtts import gTTS |
|
|
import re |
|
|
from shillelagh.backends.apsw.db import connect |
|
|
import base64 |
|
|
|
|
|
|
|
|
|
|
|
logging.basicConfig(level=logging.INFO) |
|
|
logger = logging.getLogger(__name__) |
|
|
|
|
|
|
|
|
credentials_base64 = os.getenv('GOOGLE_API_JSON_CONTENT') |
|
|
credentials_json = base64.b64decode(credentials_base64) |
|
|
|
|
|
credentials_info = json.loads(credentials_json) |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
spreadsheet_id = '1CMdSSTxjw9ooYDzmAufL0j3EtUxAQ6vdwbAL5e3t7aY' |
|
|
sheet_table = { |
|
|
"Location": 0, |
|
|
"Config": 1310222727 |
|
|
} |
|
|
location_table = f'https://docs.google.com/spreadsheets/d/{spreadsheet_id}/edit?gid={sheet_table["Location"]}' |
|
|
config_table = f'https://docs.google.com/spreadsheets/d/{spreadsheet_id}/edit?gid={sheet_table["Config"]}' |
|
|
|
|
|
|
|
|
connection = connect( |
|
|
":memory:", |
|
|
adapter_kwargs={ |
|
|
"gsheetsapi": { |
|
|
|
|
|
"service_account_info": credentials_info, |
|
|
} |
|
|
} |
|
|
) |
|
|
cursor = connection.cursor() |
|
|
|
|
|
config = {} |
|
|
|
|
|
def get_config(): |
|
|
global config_table |
|
|
global cursor |
|
|
global config |
|
|
rows = cursor.execute(f"SELECT * FROM '{config_table}' LIMIT 1").fetchall() |
|
|
config = rows[0][1] |
|
|
config = json.loads(config) |
|
|
logging.info(f"Config loaded: {config}") |
|
|
|
|
|
def update_run(): |
|
|
global config |
|
|
global config_table |
|
|
global cursor |
|
|
config["last_run"] = config.get("last_run", 0) + 1 |
|
|
updated_config = json.dumps(config) |
|
|
try: |
|
|
cursor.execute( |
|
|
f"UPDATE '{config_table}' SET config = ? WHERE id = ?", |
|
|
(updated_config, 1) |
|
|
) |
|
|
except Exception as e: |
|
|
logger.error(f"Error updating config: {e}") |
|
|
|
|
|
def update_config(): |
|
|
global config |
|
|
global config_table |
|
|
global cursor |
|
|
|
|
|
updated_config = json.dumps(config) |
|
|
try: |
|
|
cursor.execute( |
|
|
f"UPDATE '{config_table}' SET config = ? WHERE id = ?", |
|
|
(updated_config, 1) |
|
|
) |
|
|
except Exception as e: |
|
|
logger.error(f"Error updating config: {e}") |
|
|
|
|
|
get_config() |
|
|
|
|
|
sample = 0 |
|
|
def update_location(obj): |
|
|
global cursor |
|
|
global config |
|
|
global location_table |
|
|
global sample |
|
|
sample += 1 |
|
|
try: |
|
|
|
|
|
cursor.execute( |
|
|
f"INSERT INTO '{location_table}' (Run, Sample, LocationObj, Address, AI) VALUES (?, ?, ?, ?, ?)", |
|
|
(config["last_run"], sample, json.dumps(obj), obj["address"], obj["ai_info"]) |
|
|
) |
|
|
connection.commit() |
|
|
logger.info(f"Updated location in Google Sheets: {obj}") |
|
|
except Exception as e: |
|
|
logger.error(f"Error updating location in Google Sheets: {e}") |
|
|
|
|
|
|
|
|
is_timer_active = False |
|
|
current_interval = 10.0 |
|
|
|
|
|
timer = None |
|
|
|
|
|
def modify_timer(interval): |
|
|
logging.info(f"Timer interval changed to: {interval}") |
|
|
global timer |
|
|
if interval == "10 sec": |
|
|
interval_secs = 10 |
|
|
elif interval == "1 min": |
|
|
interval_secs = 60 |
|
|
elif interval == "2 min": |
|
|
interval_secs = 120 |
|
|
elif interval == "5 min": |
|
|
interval_secs = 300 |
|
|
|
|
|
timer.value = interval_secs |
|
|
global config |
|
|
config["interval"] = interval_secs |
|
|
update_config() |
|
|
|
|
|
|
|
|
def toggle_timer(state): |
|
|
global is_timer_active, timer |
|
|
global config |
|
|
is_timer_active = state |
|
|
if state: |
|
|
|
|
|
get_config() |
|
|
update_run() |
|
|
|
|
|
timer.value = config.get("interval", 300) |
|
|
|
|
|
|
|
|
|
|
|
counter = 0 |
|
|
def auto_trigger(): |
|
|
""" |
|
|
Simulates pressing the '📍 Get My Location and Street Info' button. |
|
|
This function is triggered by the timer. |
|
|
""" |
|
|
global counter |
|
|
global is_timer_active |
|
|
counter += 1 |
|
|
if is_timer_active: |
|
|
get_config() |
|
|
return f"Auto-triggered: Getting location...{counter}" |
|
|
return None |
|
|
|
|
|
|
|
|
def text_to_speech(text, lang='en'): |
|
|
""" |
|
|
Converts text to speech using gTTS and returns the path to the audio file. |
|
|
""" |
|
|
if not text: |
|
|
return None |
|
|
tts = gTTS(text=text, lang=lang) |
|
|
audio_file = "street_info.mp3" |
|
|
tts.save(audio_file) |
|
|
return audio_file |
|
|
|
|
|
|
|
|
|
|
|
openai.api_key = os.getenv("OPENAI_API_KEY", "your_openai_api_key") |
|
|
|
|
|
def get_address(lat, lon): |
|
|
""" |
|
|
Reverse geocodes the given latitude and longitude to retrieve the address. |
|
|
""" |
|
|
headers = {"User-Agent": "gradio-gps-app"} |
|
|
url = "https://nominatim.openstreetmap.org/reverse" |
|
|
params = { |
|
|
"format": "jsonv2", |
|
|
"lat": lat, |
|
|
"lon": lon, |
|
|
"addressdetails": 1, |
|
|
"accept-language": "en", |
|
|
} |
|
|
try: |
|
|
response = requests.get(url, headers=headers, params=params, timeout=5) |
|
|
response.raise_for_status() |
|
|
data = response.json() |
|
|
datas = json.dumps(data, indent=4) |
|
|
logging.info(f"Reverse geocoding response: {datas}") |
|
|
return data |
|
|
except Exception as e: |
|
|
data = {"display_name": "Error retrieving address"} |
|
|
return data |
|
|
|
|
|
|
|
|
def get_nearby_pois(lat, lon, radius=500, categories=None): |
|
|
""" |
|
|
Retrieves nearby points of interest (POIs) within a specified radius. |
|
|
|
|
|
Parameters: |
|
|
- lat (float): Latitude of the location. |
|
|
- lon (float): Longitude of the location. |
|
|
- radius (int): Search radius in meters. Default is 500 meters. |
|
|
- categories (dict): Dictionary containing OSM tag-value pairs for different categories. |
|
|
|
|
|
Returns: |
|
|
- list: A list of dictionaries containing POI names and types. |
|
|
""" |
|
|
if categories is None: |
|
|
categories = { |
|
|
'historical': [ |
|
|
('historic', 'archaeological_site'), |
|
|
('historic', 'castle'), |
|
|
('historic', 'monument'), |
|
|
('historic', 'ruins'), |
|
|
('historic', 'memorial') |
|
|
], |
|
|
'religious': [ |
|
|
('amenity', 'place_of_worship'), |
|
|
('building', 'church'), |
|
|
('building', 'mosque'), |
|
|
('building', 'synagogue'), |
|
|
('building', 'temple') |
|
|
], |
|
|
'political': [ |
|
|
('amenity', 'townhall'), |
|
|
('office', 'government'), |
|
|
('building', 'government') |
|
|
] |
|
|
} |
|
|
|
|
|
overpass_url = "http://overpass-api.de/api/interpreter" |
|
|
overpass_query = f""" |
|
|
[out:json]; |
|
|
( |
|
|
{"".join(f'node(around:{radius},{lat},{lon})["{key}"="{value}"];' for category in categories.values() for key, value in category)} |
|
|
{"".join(f'way(around:{radius},{lat},{lon})["{key}"="{value}"];' for category in categories.values() for key, value in category)} |
|
|
{"".join(f'relation(around:{radius},{lat},{lon})["{key}"="{value}"];' for category in categories.values() for key, value in category)} |
|
|
); |
|
|
out body; |
|
|
""" |
|
|
try: |
|
|
response = requests.get(overpass_url, params={'data': overpass_query}, timeout=10) |
|
|
response.raise_for_status() |
|
|
data = response.json() |
|
|
pois = [] |
|
|
for element in data.get('elements', []): |
|
|
tags = element.get('tags', {}) |
|
|
poi = { |
|
|
'name': tags.get('name:en', 'Unnamed'), |
|
|
'type': tags.get('amenity') or tags.get('building') or tags.get('historic') or 'Unknown', |
|
|
'category': next((cat for cat, tag_list in categories.items() if any(tags.get(k) == v for k, v in tag_list)), 'Other') |
|
|
} |
|
|
pois.append(poi) |
|
|
return pois |
|
|
except Exception as e: |
|
|
logging.error(f"Error retrieving POIs: {e}") |
|
|
return [] |
|
|
|
|
|
def get_ai_info(street_name): |
|
|
""" |
|
|
Queries OpenAI's GPT model to obtain information about the specified street. |
|
|
""" |
|
|
messages = [ |
|
|
{"role": "system", "content": """ |
|
|
You are a whitty tour guide. You know a lot about any area and name, and you can search if needed. |
|
|
--- |
|
|
Answer only in english, and with no special characters, links or other technical details - the output is to be read out loud. |
|
|
No markdown parts, no '#', '##' '**' etc. |
|
|
--- |
|
|
You will get as input a town, road name, and points of interest. |
|
|
You randomly decide about the direction of your answer: either elaborate about the actual location, or the meaining and history of the name given to the street, |
|
|
or choose a point of interest and elaborate about it. Focus on history, and stories. |
|
|
Try to find a joke about the subject, or some fun facts. |
|
|
--- |
|
|
Maker your answer ready for narration: interesting, storytelling-style with as many puncutiaion marks, pauses and not links or other technical details. |
|
|
Your answer should be read out in about than 2 minute. |
|
|
"""}, |
|
|
{"role": "user", "content": f"Location information:{street_name}."} |
|
|
] |
|
|
try: |
|
|
|
|
|
client = openai.OpenAI() |
|
|
completion = client.chat.completions.create( |
|
|
model="gpt-4o-mini-search-preview", |
|
|
web_search_options={ |
|
|
"search_context_size": "low", |
|
|
}, |
|
|
messages=messages, |
|
|
|
|
|
) |
|
|
j = completion |
|
|
return j |
|
|
r = completion.choices[0].message.content.strip() |
|
|
return r |
|
|
|
|
|
except Exception as e: |
|
|
return f"Error retrieving street information: {e}" |
|
|
|
|
|
def preprocess_text(input_text): |
|
|
|
|
|
text_without_urls = re.sub(r'http\S+', '', input_text) |
|
|
|
|
|
return text_without_urls |
|
|
|
|
|
|
|
|
|
|
|
def process_location(coords): |
|
|
""" |
|
|
Processes the user's coordinates to retrieve the address, street information, |
|
|
and nearby points of interest. |
|
|
""" |
|
|
global is_timer_active |
|
|
if not is_timer_active: |
|
|
return "Timer is not active. Please enable the timer to get location updates.", "Unable to retrieve street information.", None |
|
|
try: |
|
|
lat, lon = map(float, coords.split(",")) |
|
|
address_data = get_address(lat, lon) |
|
|
if address_data['display_name'] == "Error retrieving address": |
|
|
return "Error retrieving address", "Unable to retrieve street information.", None |
|
|
town = address_data['address'].get('town', 'Unknown town') |
|
|
road = address_data['address'].get('road', 'Unknown road') |
|
|
address = f"{road}, {town}" |
|
|
pois = get_nearby_pois(lat, lon) |
|
|
poi_descriptions = '; '.join(f"{poi['name']} ({poi['type']})" for poi in pois) |
|
|
combined_info = f"{address} Nearby points of interest include: {poi_descriptions}." |
|
|
logging.info(f"Retrieved address: {combined_info}") |
|
|
ai_obj = get_ai_info(address) |
|
|
ai_info = ai_obj.choices[0].message.content.strip() |
|
|
ai_obj = ai_obj.to_json() |
|
|
ai_info = preprocess_text(ai_info) |
|
|
audio_file = text_to_speech(ai_info) |
|
|
location_obj = { |
|
|
"lat": lat, |
|
|
"lon": lon, |
|
|
"address": address, |
|
|
"combined_info": combined_info, |
|
|
"ai_obj": ai_obj, |
|
|
"ai_info": ai_info, |
|
|
} |
|
|
|
|
|
update_location(location_obj) |
|
|
return combined_info, ai_info, audio_file |
|
|
except ValueError: |
|
|
return "Invalid coordinates provided.", "Unable to retrieve street information.", None |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
with gr.Blocks() as demo: |
|
|
coords_box = gr.Textbox(label="Your Coordinates (Lat, Lon)", visible=False) |
|
|
address_output = gr.Textbox(label="Resolved Address") |
|
|
street_info_output = gr.Textbox(label="Street Information", lines=5) |
|
|
audio_output = gr.Audio(label="Street Information Audio", autoplay=True, elem_id="audio_player") |
|
|
|
|
|
with gr.Row(): |
|
|
get_location_btn = gr.Button("📍 Get My Location and Street Info") |
|
|
read_aloud_btn = gr.Button("🔊 Read Aloud") |
|
|
toggle_button = gr.Checkbox(label="Enable Auto Trigger", value=False) |
|
|
interval_selector = gr.Dropdown( |
|
|
choices=["10 sec", "1 min", "2 min", "5 min"], value="10 sec", label="Set Timer Interval" |
|
|
) |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
debug_geolocation_js = """ |
|
|
() => new Promise((resolve, reject) => { |
|
|
try { |
|
|
// Generate realistic coordinates within populated areas |
|
|
const locations = [ |
|
|
{lat: 40.7128, lon: -74.0060}, // New York, USA |
|
|
{lat: 48.8566, lon: 2.3522}, // Paris, France |
|
|
{lat: 35.6762, lon: 139.6503}, // Tokyo, Japan |
|
|
{lat: 51.5074, lon: -0.1278}, // London, UK |
|
|
{lat: -33.8688, lon: 151.2093},// Sydney, Australia |
|
|
{lat: 19.0760, lon: 72.8777}, // Mumbai, India |
|
|
{lat: -23.5505, lon: -46.6333} // São Paulo, Brazil |
|
|
]; |
|
|
|
|
|
// Select a random location |
|
|
const randomLocation = locations[Math.floor(Math.random() * locations.length)]; |
|
|
const coords = `${randomLocation.lat.toFixed(6)},${randomLocation.lon.toFixed(6)}`; |
|
|
|
|
|
console.log("Debug Geolocation fetched:", coords); |
|
|
resolve(coords); |
|
|
} catch (error) { |
|
|
console.error("Error generating random geolocation: " + error.message); |
|
|
reject(); |
|
|
} |
|
|
}) |
|
|
""" |
|
|
|
|
|
real_geolocation_js = """ |
|
|
() => new Promise((resolve, reject) => { |
|
|
if (navigator.geolocation) { |
|
|
navigator.geolocation.getCurrentPosition( |
|
|
(position) => { |
|
|
const coords = `${position.coords.latitude},${position.coords.longitude}`; |
|
|
resolve(coords); |
|
|
}, |
|
|
(error) => { |
|
|
alert("Geolocation error: " + error.message); |
|
|
reject(); |
|
|
}, |
|
|
{ enableHighAccuracy: true } |
|
|
); |
|
|
} else { |
|
|
alert("Geolocation is not supported by this browser."); |
|
|
reject(); |
|
|
} |
|
|
}) |
|
|
""" |
|
|
|
|
|
geolocation_js = real_geolocation_js |
|
|
|
|
|
get_location_btn.click(None, js=geolocation_js, outputs=coords_box) |
|
|
|
|
|
|
|
|
coords_box.change( |
|
|
fn=process_location, |
|
|
inputs=coords_box, |
|
|
outputs=[address_output, street_info_output, audio_output] |
|
|
) |
|
|
|
|
|
|
|
|
read_aloud_btn.click( |
|
|
fn=text_to_speech, |
|
|
inputs=street_info_output, |
|
|
outputs=audio_output |
|
|
) |
|
|
|
|
|
|
|
|
toggle_button.change(toggle_timer, inputs=toggle_button, outputs=None) |
|
|
|
|
|
|
|
|
trigger_box = gr.Textbox(label="Trigger Box", visible=True) |
|
|
|
|
|
timer = gr.Timer(config.get("interval", 300)) |
|
|
timer.tick(fn=auto_trigger, outputs=trigger_box) |
|
|
trigger_box.change(fn=None, inputs = None, outputs = coords_box, js=geolocation_js) |
|
|
|
|
|
interval_selector.change( |
|
|
modify_timer, |
|
|
inputs=interval_selector, |
|
|
outputs=None, |
|
|
) |
|
|
|
|
|
|
|
|
demo.launch() |
|
|
|
