Update app.py

app.py

@@ -1,111 +1,179 @@
-from smolagents import CodeAgent, DuckDuckGoSearchTool, HfApiModel, load_tool, tool
 import datetime
 import requests
 import pytz
 import yaml
 import os
+from typing import Dict, Union
+
+from smolagents import CodeAgent, DuckDuckGoSearchTool, HfApiModel, load_tool, tool
 from tools.final_answer import FinalAnswerTool
 from Gradio_UI import GradioUI
 
-# Fetch USGS API credentials from environment variables (set in Hugging Face Spaces secrets)
-USGS_USERNAME = os.getenv("USGS_USERNAME")
-USGS_TOKEN = os.getenv("USGS_TOKEN")
+# Load the API key from Hugging Face Spaces secrets (environment variable)
+API_KEY = os.getenv("CRICKETDATA_API_KEY", "default_key_if_not_found")
+BASE_URL = "https://api.cricketdata.org/v1"
 
+# Custom Tool: Analyze Cricketer Form
 @tool
-def fetch_landsat_image(lat_min: float, lon_min: float, lat_max: float, lon_max: float, start_date: str, end_date: str) -> str:
-    """Fetches download URLs for all available Landsat images from USGS M2M API and returns them as clickable HTML links.
+def analyze_cricketer_form(player_name: str, role: str) -> str:
+    """A tool to analyze a cricketer's recent form based on their role and name.
+    
     Args:
-        lat_min: Minimum latitude of the bounding box
-        lon_min: Minimum longitude of the bounding box
-        lat_max: Maximum latitude of the bounding box
-        lon_max: Maximum longitude of the bounding box
-        start_date: Start date in 'YYYY-MM-DD' format
-        end_date: End date in 'YYYY-MM-DD' format
+        player_name: The name of the cricketer (e.g., 'Virat Kohli').
+        role: The player's role ('batter', 'bowler', 'wicket_keeper', 'fielder').
+    
+    Returns:
+        A string summary of the player's career stats and recent form.
     """
-    if not USGS_USERNAME or not USGS_TOKEN:
-        return "<p>Error: USGS_USERNAME or USGS_TOKEN not set in environment variables.</p>"
-
-    service_url = "https://m2m.cr.usgs.gov/api/api/json/stable/"
-
-    def send_request(url, data, api_key=None):
-        json_data = json.dumps(data)
-        headers = {'X-Auth-Token': api_key} if api_key else {}
-        response = requests.post(url, json_data, headers=headers)
-        output = response.json()
-        if output.get('errorCode'):
-            return f"Error: {output['errorCode']} - {output['errorMessage']}"
-        return output['data']
-
+    valid_roles = ['batter', 'bowler', 'wicket_keeper', 'fielder']
+    if role.lower() not in valid_roles:
+        return f"Error: Invalid role '{role}'. Choose from {valid_roles}."
+
+    player_id = get_player_id(player_name)
+    if not player_id:
+        return f"Error: Player '{player_name}' not found."
+
+    career_stats = get_career_stats(player_id)
+    if not career_stats:
+        return f"Error: Could not fetch career stats for '{player_name}'."
+
+    recent_matches = get_recent_matches(player_id)
+    
+    response = f"Analysis of {player_name} ({role.capitalize()}):\n\n"
+    response += "Career Stats:\n" + format_career_stats(career_stats, role) + "\n\n"
+    response += "Recent Form (Last 5 Matches):\n" + analyze_recent_form(recent_matches, role, player_name)
+    
+    return response
+
+def get_player_id(player_name: str) -> Union[str, None]:
+    url = f"{BASE_URL}/players/search?query={player_name}&apikey={API_KEY}"
     try:
-        # Step 1: Authenticate
-        login_payload = {'username': USGS_USERNAME, 'token': USGS_TOKEN}
-        api_key = send_request(service_url + "login-token", login_payload)
-        if not api_key or isinstance(api_key, str) and "Error" in api_key:
-            return "<p>Authentication failed. Check username and token in secrets.</p>"
-
-        # Step 2: Search scenes
-        spatial_filter = {
-            'filterType': "mbr",
-            'lowerLeft': {'latitude': lat_min, 'longitude': lon_min},
-            'upperRight': {'latitude': lat_max, 'longitude': lon_max}
-        }
-        temporal_filter = {'start': start_date, 'end': end_date}
-        search_payload = {
-            'datasetName': 'landsat_ot_c2_l1',
-            'spatialFilter': spatial_filter,
-            'temporalFilter': temporal_filter,
-            'maxResults': 50,
-            'sortField': 'cloudCover',
-            'sortDirection': 'ASC'
-        }
-        scenes = send_request(service_url + "scene-search", search_payload, api_key)
-        if isinstance(scenes, str) and "Error" in scenes:
-            return f"<p>{scenes}</p>"
-        if not scenes['results']:
-            return "<p>No Landsat scenes found for the given area and time range.</p>"
-
-        # Step 3: Get download options
-        entity_ids = [scene['entityId'] for scene in scenes['results']]
-        download_options_payload = {
-            'datasetName': 'landsat_ot_c2_l1',
-            'entityIds': entity_ids
-        }
-        download_options = send_request(service_url + "download-options", download_options_payload, api_key)
-        if isinstance(download_options, str) and "Error" in download_options:
-            return f"<p>{download_options}</p>"
-
-        # Step 4: Request downloads
-        downloads = [opt for opt in download_options if opt['available']]
-        if not downloads:
-            return "<p>No available downloads for these scenes.</p>"
-
-        label = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
-        download_request_payload = {
-            'downloads': [{'entityId': opt['entityId'], 'productId': opt['id']} for opt in downloads],
-            'label': label
-        }
-        request_results = send_request(service_url + "download-request", download_request_payload, api_key)
-        if isinstance(request_results, str) and "Error" in request_results:
-            return f"<p>{request_results}</p>"
-
-        # Step 5: Generate HTML with clickable links
-        urls = [download['url'] for download in request_results['availableDownloads']]
-        if not urls:
-            return "<p>No download URLs immediately available. Try again later.</p>"
-
-        html_output = f"<p>Found {len(urls)} available Landsat scenes:</p><ul>"
-        for i, (scene, url) in enumerate(zip(scenes['results'][:len(urls)], urls)):
-            html_output += f'<li>Cloud Cover: {scene["cloudCover"]}% - <a href="{url}" target="_blank">Download Scene {i+1}</a></li>'
-        html_output += "</ul>"
-        return html_output
-
+        response = requests.get(url)
+        response.raise_for_status()
+        data = response.json()
+        if data.get("data") and len(data["data"]) > 0:
+            return data["data"][0]["id"]
+        return None
+    except requests.RequestException as e:
+        print(f"Error fetching player ID: {e}")
+        return None
+
+def get_career_stats(player_id: str) -> Union[Dict, None]:
+    url = f"{BASE_URL}/players/{player_id}/stats?apikey={API_KEY}"
+    try:
+        response = requests.get(url)
+        response.raise_for_status()
+        return response.json()
+    except requests.RequestException as e:
+        print(f"Error fetching career stats: {e}")
+        return None
+
+def get_recent_matches(player_id: str) -> list:
+    url = f"{BASE_URL}/players/{player_id}/matches?apikey={API_KEY}"
+    try:
+        response = requests.get(url)
+        response.raise_for_status()
+        matches = response.json().get("data", [])
+        return matches[:5]  # Limit to last 5 matches
+    except requests.RequestException as e:
+        print(f"Error fetching recent matches: {e}")
+        return []
+
+def format_career_stats(stats: Dict, role: str) -> str:
+    if not stats or "data" not in stats:
+        return "No stats available."
+    
+    data = stats["data"]
+    output = ""
+    
+    if role == "batter":
+        batting = data.get("batting", {})
+        output += f"Matches: {batting.get('matches', 'N/A')}\n"
+        output += f"Runs: {batting.get('runs', 'N/A')}\n"
+        output += f"Average: {batting.get('average', 'N/A')}\n"
+        output += f"Strike Rate: {batting.get('strike_rate', 'N/A')}\n"
+    
+    elif role == "bowler":
+        bowling = data.get("bowling", {})
+        output += f"Matches: {bowling.get('matches', 'N/A')}\n"
+        output += f"Wickets: {bowling.get('wickets', 'N/A')}\n"
+        output += f"Economy: {bowling.get('economy', 'N/A')}\n"
+        output += f"Best: {bowling.get('best', {}).get('figures', 'N/A')}\n"
+    
+    elif role == "wicket_keeper":
+        fielding = data.get("fielding", {})
+        output += f"Catches: {fielding.get('catches', 'N/A')}\n"
+        output += f"Stumpings: {fielding.get('stumpings', 'N/A')}\n"
+    
+    elif role == "fielder":
+        fielding = data.get("fielding", {})
+        output += f"Catches: {fielding.get('catches', 'N/A')}\n"
+        output += f"Run Outs: {fielding.get('run_outs', 'N/A')}\n"
+    
+    return output
+
+def analyze_recent_form(matches: list, role: str, player_name: str) -> str:
+    if not matches:
+        return "No recent match data available."
+    
+    output = ""
+    if role == "batter":
+        total_runs = 0
+        games = len(matches)
+        for match in matches:
+            batting = match.get("batting", {})
+            runs = batting.get("runs", 0)
+            total_runs += runs
+            output += f"Match vs {match.get('opponent', 'Unknown')}: {runs} runs\n"
+        avg = total_runs / games if games > 0 else 0
+        output += f"Recent Average: {avg:.2f}\n"
+        output += f"Form Comment: {'Strong' if avg > 30 else 'Inconsistent'} batting form."
+    
+    elif role == "bowler":
+        total_wickets = 0
+        games = len(matches)
+        for match in matches:
+            bowling = match.get("bowling", {})
+            wickets = bowling.get("wickets", 0)
+            total_wickets += wickets
+            output += f"Match vs {match.get('opponent', 'Unknown')}: {wickets} wickets\n"
+        avg_wickets = total_wickets / games if games > 0 else 0
+        output += f"Recent Wickets per Game: {avg_wickets:.2f}\n"
+        output += f"Form Comment: {'Strong' if avg_wickets > 1 else 'Inconsistent'} bowling form."
+    
+    elif role == "wicket_keeper":
+        total_stumpings = sum(m.get("fielding", {}).get("stumpings", 0) for m in matches)
+        total_catches = sum(m.get("fielding", {}).get("catches", 0) for m in matches)
+        output += f"Stumpings: {total_stumpings}, Catches: {total_catches}\n"
+        output += f"Form Comment: {'Reliable' if total_stumpings + total_catches > 3 else 'Average'} keeping."
+    
+    elif role == "fielder":
+        total_catches = sum(m.get("fielding", {}).get("catches", 0) for m in matches)
+        total_run_outs = sum(m.get("fielding", {}).get("run_outs", 0) for m in matches)
+        output += f"Catches: {total_catches}, Run Outs: {total_run_outs}\n"
+        output += f"Form Comment: {'Sharp' if total_catches + total_run_outs > 3 else 'Average'} fielding."
+    
+    return output
+
+# Example tool from your template
+@tool
+def get_current_time_in_timezone(timezone: str) -> str:
+    """A tool that fetches the current local time in a specified timezone.
+    Args:
+        timezone: A string representing a valid timezone (e.g., 'America/New_York').
+    """
+    try:
+        tz = pytz.timezone(timezone)
+        local_time = datetime.datetime.now(tz).strftime("%Y-%m-%d %H:%M:%S")
+        return f"The current local time in {timezone} is: {local_time}"
     except Exception as e:
-        return f"<p>Error fetching Landsat images: {str(e)}</p>"
-
-
+        return f"Error fetching time for timezone '{timezone}': {str(e)}"
 
+# Initialize tools
 final_answer = FinalAnswerTool()
+image_generation_tool = load_tool("agents-course/text-to-image", trust_remote_code=True)
 
+# Model setup
 model = HfApiModel(
     max_tokens=2096,
     temperature=0.5,
@@ -113,23 +181,20 @@ model = HfApiModel(
     custom_role_conversions=None,
 )
 
-image_generation_tool = load_tool("agents-course/text-to-image", trust_remote_code=True)
-
-with open("prompts.yaml", "r") as stream:
+# Load prompt templates
+with open("prompts.yaml", 'r') as stream:
     prompt_templates = yaml.safe_load(stream)
 
+# Initialize the agent
 agent = CodeAgent(
     model=model,
-    tools=[fetch_landsat_image, final_answer],
+    tools=[final_answer, analyze_cricketer_form, get_current_time_in_timezone, image_generation_tool],
     max_steps=6,
     verbosity_level=1,
     grammar=None,
-    planning_interval=None,
-    name=None,
-    description=None,
     prompt_templates=prompt_templates
 )
 
-# Launch Gradio UI
-interface = GradioUI(agent)
-interface.launch(share=False)  # 'share=True' for public link in Spaces
+# Launch the Gradio UI
+if __name__ == "__main__":
+    GradioUI(agent).launch()