Spaces:
Sleeping
Sleeping
import pandas as pd | |
import gradio as gr | |
from typing import Dict | |
# Load the datasets | |
df1 = pd.read_csv('cleaned_climate_change_data.csv') | |
df2 = pd.read_csv('cleaned_climate_data.csv') | |
class ClimateDataTool: | |
"""Tool to get climate data from two datasets""" | |
def get_name(self) -> str: | |
return "get_climate_data" | |
def get_description(self) -> str: | |
return "Retrieve climate data for a given year and indicator from the datasets" | |
def get_params_definition(self) -> Dict[str, str]: | |
return { | |
"year": { | |
"param_type": "str", | |
"description": "The year for which to get climate data in 'yyyy' format.", | |
"required": False, | |
}, | |
"indicator": { | |
"param_type": "str", | |
"description": "The indicator for which to get climate data (e.g., 'CAT current policies median').", | |
"required": False, | |
}, | |
} | |
def run(self, params: Dict[str, str]) -> str: | |
year = params.get("year") | |
indicator = params.get("indicator", "").lower() | |
print(f"Extracted Year: {year}, Extracted Indicator: {indicator}") # Debug output | |
response_df1 = self.answer_from_df1(year, indicator) | |
response_df2 = self.answer_from_df2(year, indicator) | |
if "No data found" in response_df1 and "No data found" in response_df2: | |
return "No relevant data found in either dataset." | |
return response_df1 + "\n\n" + response_df2 | |
def answer_from_df1(self, year: str, indicator: str) -> str: | |
if year: | |
try: | |
year = int(year) | |
print(f"Checking Dataset 1 for Year: {year}") # Debug output | |
year_data = df1[df1['year'] == year] # Filter by year | |
print(f"Dataset 1 Year Data: {year_data}") # Debug output | |
if not year_data.empty: | |
if indicator and indicator in df1['indicator'].str.lower().values: | |
print(f"Checking Dataset 1 for Indicator: {indicator}") # Debug output | |
indicator_data = year_data[year_data['indicator'].str.lower() == indicator] | |
if not indicator_data.empty: | |
return indicator_data[['indicator', 'value']].to_string(index=False) | |
else: | |
return f"No data available for the indicator '{indicator}' in year {year}." | |
else: | |
return year_data[['indicator', 'value']].to_string(index=False) | |
else: | |
return f"No data available for the year {year}." | |
except ValueError: | |
return "Please specify a valid year." | |
return "No data found in Dataset 1." | |
def answer_from_df2(self, year: str, indicator: str) -> str: | |
if year: | |
try: | |
year = int(year) | |
print(f"Checking Dataset 2 for Year: {year}") # Debug output | |
year_data = df2[df2['year'] == year] # Filter by year | |
print(f"Dataset 2 Year Data: {year_data}") # Debug output | |
if not year_data.empty: | |
# Check if the indicator exists in the column names | |
indicator_columns = [col.lower() for col in df2.columns] | |
if indicator and indicator in indicator_columns: | |
return year_data[['year', indicator]].to_string(index=False) | |
else: | |
return year_data.to_string(index=False) | |
else: | |
return f"No data available for the year {year}." | |
except ValueError: | |
return "Please specify a valid year." | |
return "No data found in Dataset 2." | |
# Gradio Interface | |
def get_climate_data_from_user(query): | |
tool = ClimateDataTool() | |
# Simulate parsing query as parameters from user input | |
params = {"year": "", "indicator": ""} | |
# Extract year from the query | |
if "year" in query.lower(): | |
try: | |
year = int(query.split()[-1]) | |
params['year'] = str(year) | |
except: | |
pass | |
# Extract indicator from the query | |
if "indicator" in query.lower(): | |
params['indicator'] = query.split("indicator")[-1].strip().lower() | |
print(f"User Query Parsed Params: {params}") # Debug output | |
return tool.run(params) | |
# Gradio Interface setup | |
interface = gr.Interface( | |
fn=get_climate_data_from_user, | |
inputs="text", | |
outputs="text", | |
title="Climate Data Q&A", | |
description="Ask questions about climate data by year and indicator." | |
) | |
if __name__ == "__main__": | |
interface.launch() | |