app.py

from smolagents import CodeAgent, DuckDuckGoSearchTool, HfApiModel, load_tool, tool
import datetime
import requests
import pytz
import yaml
import os
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")

@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.
    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
    """
    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']

    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

    except Exception as e:
        return f"<p>Error fetching Landsat images: {str(e)}</p>"



final_answer = FinalAnswerTool()

model = HfApiModel(
    max_tokens=2096,
    temperature=0.5,
    model_id='Qwen/Qwen2.5-Coder-32B-Instruct',
    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:
    prompt_templates = yaml.safe_load(stream)

agent = CodeAgent(
    model=model,
    tools=[fetch_landsat_image, final_answer],
    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