Blame for app.py

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()