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 # Set up logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) # Google Sheets API setup credentials_base64 = os.getenv('GOOGLE_API_JSON_CONTENT') credentials_json = base64.b64decode(credentials_base64) # Load the credentials from the JSON string credentials_info = json.loads(credentials_json) # credentials = service_account.Credentials.from_service_account_info(credentials_info) # os.environ['GOOGLE_APPLICATION_CREDENTIALS'] = 'location-455719-d08225f31b58.json' # URL format: https://docs.google.com/spreadsheets/d/{spreadsheet_id}/edit # https://docs.google.com/spreadsheets/d/1CMdSSTxjw9ooYDzmAufL0j3EtUxAQ6vdwbAL5e3t7aY/edit?usp=sharing 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"]}' # Connect to the Google Sheet connection = connect( ":memory:", adapter_kwargs={ "gsheetsapi": { # "service_account_file": 'location-455719-d08225f31b58.json' "service_account_info": credentials_info, } } ) cursor = connection.cursor() config = {} # Get latest 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 # 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}") get_config() sample = 0 def update_location(obj): global cursor global config global location_table global sample sample += 1 try: # Update the Google Sheet with the new location 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}") # Global state for timer control is_timer_active = False current_interval = 10.0 # Default to 1 second 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: # Start the timer get_config() update_run() timer.value = config.get("interval", 300) # return f"Timer {'activated' if state else 'deactivated'}." 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 # Set your OpenAI API key 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: # TODO: remove commentted code client = openai.OpenAI() completion = client.chat.completions.create( model="gpt-4o-mini-search-preview", web_search_options={ "search_context_size": "low", }, messages=messages, # max_tokens=150, ) j = completion return j r = completion.choices[0].message.content.strip() return r # return "Debug text" except Exception as e: return f"Error retrieving street information: {e}" def preprocess_text(input_text): # Remove URLs text_without_urls = re.sub(r'http\S+', '', input_text) # Additional preprocessing steps can be added here 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 the Google Sheet with the new location update_location(location_obj) return combined_info, ai_info, audio_file except ValueError: return "Invalid coordinates provided.", "Unable to retrieve street information.", None # Gradio Interface 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" ) # JavaScript to fetch geolocation and populate coords_box 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) # Update outputs when coordinates change coords_box.change( fn=process_location, inputs=coords_box, outputs=[address_output, street_info_output, audio_output] ) # Trigger text-to-speech when "Read Aloud" button is clicked read_aloud_btn.click( fn=text_to_speech, inputs=street_info_output, outputs=audio_output ) # Toggle Timer On/Off toggle_button.change(toggle_timer, inputs=toggle_button, outputs=None) # Hidden textbox that gets updated by the Timer to trigger JS code trigger_box = gr.Textbox(label="Trigger Box", visible=True) # Timer component - triggers every `current_interval` seconds 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, ) # Launch the Gradio application demo.launch()