Download app.py from dn-scribe/location: direct link, hf CLI and curl.
- Browser
- Download file 15.8 kB
-
https://huggingface.co/spaces/dn-scribe/location/resolve/main/app.py
- Command line
-
hf download hf://spaces/dn-scribe/location/app.py
-
curl -L -o app.py https://huggingface.co/spaces/dn-scribe/location/resolve/main/app.py
15.8 kB
| 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() | |