Grounding Gemini With Google Search and Other Data Sources

def get_crash_data():

 import looker_sdk
 from looker_sdk import models40 as models
 import os
 import json
 
 sdk = looker_sdk.init40("looker.ini")

 look_id = "Enter Look ID"
 
 try:
 response = sdk.run_look(look_id=look_id, result_format="json")
 print('looker done')
 return json.loads(response)
 
 except Exception as e:
 print(f"Error getting Looker data: {e}")
 return []

def get_philly_events():
 import requests
 from datetime import datetime, timedelta
 
 base_url = "https://app.ticketmaster.com/discovery/v2/events"
 
 start_date = datetime.now()
 end_date = start_date + timedelta(days=7)
 
 params = {
 "apikey": "enter",
 "city": "Philadelphia",
 "stateCode": "PA",
 "startDateTime": start_date.strftime("%Y-%m-%dT%H:%M:%SZ"),
 "endDateTime": end_date.strftime("%Y-%m-%dT%H:%M:%SZ"),
 "size": 50,
 "sort": "date,asc"
 }
 
 try:
 response = requests.get(base_url, params=params)
 if response.status_code != 200:
 return []
 
 data = response.json()
 events = []
 
 for event in data.get("_embedded", {}).get("events", []):
 venue = event["_embedded"]["venues"][0]
 event_info = {
 "name": event["name"],
 "date": event["dates"]["start"].get("dateTime", "TBA"),
 "venue": event["_embedded"]["venues"][0]["name"],
 "street": venue.get("address", {}).get("line1", "")
 }
 events.append(event_info)
 
 return events
 
 except Exception as e:
 print(f"Error getting events data: {e}")
 return []

def get_philly_weather_forecast():
 import requests
 from datetime import datetime, timedelta
 import json
 
 lat = "39.9526"
 lon = "-75.1652"
 url = f"https://api.weather.gov/points/{lat},{lon}"
 
 try:
 # Get API data
 response = requests.get(url, headers={'User-Agent': 'weatherapp/1.0'})
 response.raise_for_status()
 
 grid_data = response.json()
 forecast_url = grid_data['properties']['forecast']
 
 # Get forecast data
 forecast_response = requests.get(forecast_url)
 forecast_response.raise_for_status()
 forecast_data = forecast_response.json()
 
 weather_data = {
 "location": "Philadelphia, PA",
 "forecast_generated": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
 "data_source": "NOAA Weather API",
 "daily_forecasts": []
 }
 
 # Process forecast data - take 14 periods to get 7 full days
 periods = forecast_data['properties']['periods'][:14] # Get 14 periods (7 days × 2 periods per day)
 
 # Group periods into days
 current_date = None
 daily_data = None
 
 for period in periods:
 period_date = period['startTime'][:10] # Get just the date part of period
 is_daytime = period['isDaytime']
 
 # If we're starting a new day
 if period_date != current_date:
 # Save the previous day's data if it exists
 if daily_data is not None:
 weather_data["daily_forecasts"].append(daily_data)
 
 # Start a new daily record
 current_date = period_date
 daily_data = {
 "date": period_date,
 "forecast": {
 "day": None,
 "night": None,
 "high_temperature": None,
 "low_temperature": None,
 "conditions": None,
 "detailed_forecast": None
 }
 }
 
 # Update the daily data based on whether it's day or night
 period_data = {
 "temperature": {
 "value": period['temperature'],
 "unit": period['temperatureUnit']
 },
 "conditions": period['shortForecast'],
 "wind": {
 "speed": period['windSpeed'],
 "direction": period['windDirection']
 },
 "detailed_forecast": period['detailedForecast']
 }
 
 if is_daytime:
 daily_data["forecast"]["day"] = period_data
 daily_data["forecast"]["high_temperature"] = period_data["temperature"]
 daily_data["forecast"]["conditions"] = period_data["conditions"]
 daily_data["forecast"]["detailed_forecast"] = period_data["detailed_forecast"]
 else:
 daily_data["forecast"]["night"] = period_data
 daily_data["forecast"]["low_temperature"] = period_data["temperature"]
 
 # Append the last day's data
 if daily_data is not None:
 weather_data["daily_forecasts"].append(daily_data)
 
 # Keep only 7 days of forecast
 weather_data["daily_forecasts"] = weather_data["daily_forecasts"][:7]
 
 return json.dumps(weather_data, indent=2)
 
 except Exception as e:
 print(f"Error with NOAA API: {e}")
 return json.dumps({
 "error": str(e),
 "location": "Philadelphia, PA",
 "forecast_generated": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
 "daily_forecasts": []
 }, indent=2)

from flask import Flask, render_template, request, jsonify
import os
from google import genai
from google.genai import types
from api_handlers import get_philly_events, get_crash_data, get_philly_weather_forecast
from dotenv import load_dotenv

# Load environment variables
load_dotenv()

app = Flask(__name__)

# Initialize Gemini client
client = genai.Client(
 api_key='Enter Key Here',
)

# Global chat history
chat_history = []

def initialize_context():
 	
 try:
 # Get API data
 events = get_philly_events()
 looker_data = get_crash_data()
 weather_data = get_philly_weather_forecast()
 
 # Format events data
 events_formatted = "\n".join([
 f"- {event['name']} at {event['venue']} {event['street']} on {event['date']}" 
 for event in events
 ])
 
 # Create system context
 system_context = f"""You are a helpful AI assistant focused on Philadelphia.
You have access to the following data that was loaded when you started:

Current Philadelphia Events (Next 7 Days):
{events_formatted}

Crash Analysis Data:
{looker_data}

Instructions:
1. Use this event and crash data when answering relevant questions
2. For questions about events, reference the specific events listed above
3. For questions about crash data, use the analysis provided
4. For other questions about Philadelphia, you can provide general knowledge
5. Always maintain a natural, conversational tone
6. Use Google Search when needed for current information not in the provided data

Remember: Your events and crash data is from system initialization and represents that point in time."""

 # Add context to chat history
 chat_history.append(types.Content(
 role="user",
 parts=[types.Part.from_text(text=system_context)]
 ))

 print("Context initialized successfully")
 return True
 
 except Exception as e:
 print(f"Error initializing context: {e}")
 return False

from flask import Flask, render_template, request, jsonify
import os
from google import genai
from google.genai import types
from api_handlers import get_philly_events, get_crash_data, get_philly_weather_forecast
from dotenv import load_dotenv

# Load environment variables
load_dotenv()

app = Flask(__name__)

# Initialize Gemini client
client = genai.Client(
 api_key='Enter Key Here',
)

# Global chat history
chat_history = []

def initialize_context():
 """Initialize context with events and Looker data"""
 try:
 # Get initial data
 events = get_philly_events()
 looker_data = get_crash_data()
 weather_data = get_philly_weather_forecast()
 
 # Format events data to present better
 events_formatted = "\n".join([
 f"- {event['name']} at {event['venue']} {event['street']} on {event['date']}" 
 for event in events
 ])
 
 # Create system context
 system_context = f"""You are a helpful AI assistant focused on Philadelphia.
You have access to the following data that was loaded when you started:

Philadelphia Events for the next 7 Days:
{events_formatted}

Weather forecast for Philadelphia:
{weather_data}

Crash Analysis Data:
{looker_data}

Instructions:
1. Use this events, weather, and crash data when answering relevant questions
2. For questions about events, reference the specific events listed above
3. For questions about crash data, use the analysis provided
4. For questions about weather, use the data provided
5. For other questions about Philadelphia, you can provide general knowledge
6. Use Google Search when needed for current information not in the provided data

Remember: Your events and crash data is from system initialization and represents that point in time."""

 # Add context to chat history
 chat_history.append(types.Content(
 role="user",
 parts=[types.Part.from_text(text=system_context)]
 ))

 print("Context initialized successfully")
 return True
 
 except Exception as e:
 print(f"Error initializing context: {e}")
 return False

@app.route('/')
def home():
 return render_template('index.html')

@app.route('/chat', methods=['POST'])
def chat():
 try:
 user_message = request.json.get('message', '')
 if not user_message:
 return jsonify({'error': 'Message required'}), 400

 # Add user message to history
 chat_history.append(types.Content(
 role="user",
 parts=[types.Part.from_text(text=user_message)]
 ))

 # Configure generation settings
 generate_content_config = types.GenerateContentConfig(
 temperature=0.9,
 top_p=0.95,
 top_k=40,
 max_output_tokens=8192,
 tools=[types.Tool(google_search=types.GoogleSearch())],
 )

 # Generate response using full chat history
 response = client.models.generate_content(
 model="gemini-2.0-flash",
 contents=chat_history,
 config=generate_content_config,
 )

 # Add assistant response to history
 chat_history.append(types.Content(
 role="assistant",
 parts=[types.Part.from_text(text=response.text)]
 ))

 return jsonify({'response': response.text})

 except Exception as e:
 print(f"Error in chat endpoint: {e}")
 return jsonify({'error': str(e)}), 500

if __name__ == '__main__':
 # Initialize context before starting
 print("Initializing context...")
 if initialize_context():
 app.run(debug=True)
 else:
 print("Failed to initialize context")
 exit(1)