location / app.py
dn-scribe's picture
fix: better prompt
e4e488d unverified
Raw History Blame Contribute Delete
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()