app.py

from smolagents import CodeAgent,DuckDuckGoSearchTool, HfApiModel,load_tool,tool
import datetime
from dateutil import parser # for flexible date parsing
import requests
import pytz
import yaml
from tools.final_answer import FinalAnswerTool

from Gradio_UI import GradioUI
import gradio as gr

# Below is an example of a tool that does nothing. Amaze us with your creativity !


def parse_date(date_str: str) -> datetime.datetime: # Note the type hint change
    try:
        return parser.parse(date_str)
    except:
        return datetime.datetime.now() # Changed here


@tool
def my_custom_tool(arg1:str, arg2:int)-> str: #it's import to specify the return type
    #Keep this format for the description / args / args description but feel free to modify the tool
    """A tool that does nothing yet 
    Args:
        arg1: the first argument
        arg2: the second argument
    """
    return "What magic will you build ?"

@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:
        # Create timezone object
        tz = pytz.timezone(timezone)
        # Get current time in that 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"Error fetching time for timezone '{timezone}': {str(e)}"

@tool
def get_horoscope(sign: str, date: str = None, language: str = "EN") -> str:
    """Fetches the horoscope for a given zodiac sign and date.
    Uses the ExaWeb API. Defaults to today if no date is provided.
    
    Args:
        sign: Zodiac sign (e.g., Aries, Taurus, Gemini)
        date: Date in any format (optional)
        language: Language code ('EN' or 'HI')
    """
    try:
        params = {
            "rashi": sign.upper(),
            "language": language
        }

        if date:
            date_obj = parse_date(date)
            formatted_date = date_obj.strftime("%d-%m-%Y")
            params["day"] = formatted_date  # Only include if date is provided

        url = "https://api.exaweb.in:3004/api/rashi"
        response = requests.get(url, params=params)
        response.raise_for_status()
        data = response.json()

        sign_upper = sign.upper()
        sign_data = data.get(sign_upper)

        if sign_data:
            prediction = sign_data.get("Prediction", "No prediction found.")
            return f"Horoscope for {sign.capitalize()} on {formatted_date}:\n{prediction}"

        return f"No horoscope found for sign: {sign}"

    except Exception as e:
        return f"Error fetching horoscope: {str(e)}"

@tool
def get_date_panchang(date: str = None, data_language: str = "EN") -> str:
    """Fetches the Panchang data for a given date.
    If the user does not provide a date, use today's real date.
    Args:
        date: Date in any format (optional)
        data_language: Language of the Panchang data (e.g., 'EN' for English, 'HI' for Hindi).
    """


    if not date:
        now = datetime.datetime.now()
    else:
        try:
            now = datetime.datetime.strptime(date, "%Y-%m-%d")
        except ValueError:
            return "❌ Invalid date format. Use YYYY-MM-DD."

    api_date = now.strftime("%d/%m/%y")  # Format as DD/MM/YY for API

    try:
        url = (
            f"https://api.exaweb.in:3004/api/panchang/daily?"
            f"date={api_date}&app_language=EN&data_language={data_language}"
        )

        headers = {
            "api_key": "anvl_bharat_cal123",  # Replace with your actual token
            "Content-Type": "application/json"
        }

        response = requests.get(url, headers=headers)
        response.raise_for_status()
        data = response.json()

        if not data or not isinstance(data, dict):
            return "⚠️ Empty or unexpected response."

        # Format the entire Panchang data nicely
        lines = [f"📅 {data.get('date', api_date)} का पंचांग ({data.get('location', 'Unknown')}):\n"]

        for key, value in data.items():
            if key in ["date", "location", "app_language", "data_language", "universalDate"]:
                continue  # Skip metadata

            if isinstance(value, list):
                if all(isinstance(item, str) for item in value):
                    lines.append(f"🔹 {key}: {', '.join(value)}")
                elif all(isinstance(item, dict) for item in value):
                    lines.append(f"🔹 {key}:")
                    for item in value:
                        vals = item.get("values", [])
                        upto = item.get("upto", "")
                        lines.append(f"    • {', '.join(vals)} {'— ' + upto if upto else ''}")
            elif isinstance(value, str):
                lines.append(f"🔹 {key}: {value}")
            elif isinstance(value, int):
                lines.append(f"🔹 {key}: {value}")

        return "\n".join(lines)

    except Exception as e:
        return f"❌ Failed to fetch Panchang for {api_date}: {str(e)}"

        

@tool
def get_holidays(year: int = None, app_language: str = "EN", data_language: str = "HI") -> str:
    """
    Fetches holidays from all categories (Hindu, Islamic, Christian) for a given year from ExaWeb API.

    Args:
        year: Year (e.g., 2025). Optional — defaults to current year.
        app_language: App language for API (default: "EN").
        data_language: Data language for holidays (default: "HI").
    """
    if not year:
        year = datetime.datetime.now().year

    params = {
        "app_language": app_language,
        "data_language": data_language,
        "year": year
    }
    headers = {
        "api_key": "anvl_bharat_cal123"
    }

    try:
        response = requests.get(
            "https://api.exaweb.in:3004/api/panchang/holiday",
            params=params,
            headers=headers
        )
        response.raise_for_status()
        data = response.json()

        formatted = []

        for category, groups in data.items():
            if category in ["year", "app_language", "data_language"]:
                continue  # Skip non-holiday keys

            formatted.append(f"\n=== {category} Holidays ===")

            for group in groups:
                for holiday in group:
                    title = holiday.get("title", "")
                    date = holiday.get("date", "")
                    formatted.append(f"{date}: {title}")

        return "\n".join(formatted) if formatted else f"No holidays found for {year}."

    except requests.RequestException as e:
        return f"Error fetching holidays: {e}"


@tool
def get_panchang_field(field: str, date: str = None) -> str:
    """Fetches specific Panchang field like Tithi, Nakshatra, Yoga, etc.
    Args:
        field: The Panchang field to fetch (e.g., Tithi, Nakshatra)
        date: Date in any format (optional — defaults to today)
    """
    if not date:
        date = datetime.now().strftime("%Y-%m-%d")
    return f"{field} on {date}: [Sample value]"

# Optional tool if needed in the future:
@tool
def get_festivals_today(date: str = None) -> str:
    """Fetches the festivals for a given date.
    Args:
        date: Date in any format (e.g., '2025-04-08'). Optional — defaults to today.
    """
    if not date:
        date = datetime.now().strftime("%Y-%m-%d")
    return f"Festivals on {date}: [Sample festival list]"


final_answer = FinalAnswerTool()

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)
# If the agent does not answer, the model is overloaded, please use another model or the following Hugging Face Endpoint that also contains qwen2.5 coder:
# model_id='https://pflgm2locj2t89co.us-east-1.aws.endpoints.huggingface.cloud' 


# Define Models
primary_model = HfApiModel(
    max_tokens=2096,
    temperature=0.5,
    model_id='sarvamai/sarvam-m',  # Primary model, possibly overloaded
)

backup_model = HfApiModel(
    max_tokens=2096,
    temperature=0.5,
    model_id='https://pflgm2locj2t89co.us-east-1.aws.endpoints.huggingface.cloud',  # Backup model
)

# Function to Build an Agent with Any Model
def build_agent(model_to_use):
    # Assuming the tools are defined elsewhere
    tools = [
        # final_answer, get_horoscope, get_date_panchang,
        # get_holidays, get_panchang_field, get_festivals_today,
        # get_current_time_in_timezone, my_custom_tool, image_generation_tool
    ]
    return CodeAgent(
        model=model_to_use,
        tools=tools,
        max_steps=6,
        verbosity_level=1,
        grammar=None,
        planning_interval=None,
        name=None,
        description=None,
        prompt_templates=prompt_templates
    )

# Instantiate Primary Agent
primary_agent = build_agent(primary_model)

# Fallback-Handled Runner Function
def agent_runner(user_input):
    """
    This function takes user input, tries the primary agent,
    and switches to the backup agent on failure.
    """
    try:
        print("Attempting to run with the primary model...")
        result = primary_agent.run(user_input)
        if result is None or (isinstance(result, str) and result.strip() == ""):
            raise ValueError("Primary model returned an empty or null response.")
        return result
    except Exception as e:
        print(f"Primary model failed with error: {e}")
        print("Switching to the backup model...")
        try:
            backup_agent = build_agent(backup_model)
            result = backup_agent.run(user_input)
            if result is None or (isinstance(result, str) and result.strip() == ""):
                return "Backup model also returned an empty response."
            return result
        except Exception as e2:
            print(f"Backup model also failed with error: {e2}")
            return f"The backup model failed to generate a response: {e2}"

# Launch Gradio with the runner function
# We use the standard gr.Interface for this logic.
iface = gr.Interface(
    fn=agent_runner,
    inputs="text",
    outputs="text",
    title="Agent with Fallback Logic",
    description="Enter your query. The system will use a primary model and switch to a backup if the primary fails."
)

iface.launch()