tools.py

"""

tools.py

--------

Tool registry and executor for EcoAgent's agentic mode.



Provides:

  - TOOLS           : List of tool definitions (JSON Schema format)

  - execute_tool()  : Routes tool calls to appropriate functions

  - SCHEMES_DB      : Static lookup for Indian government eco schemes

"""

import os
import logging
from datetime import datetime
from typing import Any

# Fix SSL certificate path before importing watsonx_client
if "SSL_CERT_FILE" not in os.environ or not os.path.isfile(os.environ.get("SSL_CERT_FILE", "")):
    try:
        import certifi
        os.environ["SSL_CERT_FILE"] = certifi.where()
        os.environ["REQUESTS_CA_BUNDLE"] = certifi.where()
    except ImportError:
        pass

from watsonx_client import (
    IMPACT_TABLE,
    INDIAN_CITIES,
    compute_session_impact,
    get_recycling_guide,
    _get_model,
    AGENT_INSTRUCTIONS,
)

logger = logging.getLogger(__name__)

# ===========================================================================
#  TOOL DEFINITIONS
#  Format compatible with IBM Granite function calling / tool use.
# ===========================================================================
TOOLS: list[dict[str, Any]] = [
    {
        "name": "calculate_impact",
        "description": (
            "Calculate CO2 (kg/year), water (L/day), and waste (kg/year) savings "
            "for one or more eco actions. Use when the user asks about environmental "
            "impact, wants numbers, or compares actions."
        ),
        "parameters": {
            "type": "object",
            "properties": {
                "action_slug": {
                    "type": "string",
                    "description": (
                        "Action identifier from the impact table. "
                        "Valid slugs: cloth_bags, led_bulbs, solar_panels, composting, "
                        "public_transport, short_shower, rainwater_harvesting, "
                        "vegetarian_diet, no_plastic_bottles, drip_irrigation, "
                        "smart_powerstrip, electric_two_wheeler, reusable_bags_produce, "
                        "fix_water_leaks, line_dry_clothes, seasonal_local_produce, "
                        "segregate_waste, pressure_cooker, no_single_use_plastic, "
                        "organic_farming"
                    ),
                },
                "members": {
                    "type": "integer",
                    "description": "Number of household members to scale savings. Default: 1",
                },
            },
            "required": ["action_slug"],
        },
    },
    {
        "name": "get_recycling_guide",
        "description": (
            "Get recycling instructions for a specific material in an Indian city. "
            "Use when the user asks how to recycle something, where to dispose of waste, "
            "or wants recycling guidelines for their city."
        ),
        "parameters": {
            "type": "object",
            "properties": {
                "material": {
                    "type": "string",
                    "description": "Material category",
                    "enum": [
                        "Paper", "Plastic", "Glass", "E-waste",
                        "Metal", "Organic", "Batteries", "Clothing",
                    ],
                },
                "city": {
                    "type": "string",
                    "description": "Indian city name (e.g. Mumbai, Delhi, Bangalore)",
                },
            },
            "required": ["material", "city"],
        },
    },
    {
        "name": "web_search",
        "description": (
            "Search the web for latest eco news, government schemes, local recycling "
            "centers, or any current information. Use when the user asks about recent "
            "events, new policies, or needs up-to-date information not in your training data."
        ),
        "parameters": {
            "type": "object",
            "properties": {
                "query": {
                    "type": "string",
                    "description": "Search query string",
                },
            },
            "required": ["query"],
        },
    },
    {
        "name": "check_scheme",
        "description": (
            "Look up Indian government eco scheme details, eligibility, and benefits. "
            "Use when the user asks about subsidies, government programs, or financial "
            "incentives for eco-friendly actions."
        ),
        "parameters": {
            "type": "object",
            "properties": {
                "scheme_name": {
                    "type": "string",
                    "description": (
                        "Specific scheme name (e.g. 'PM Surya Ghar', 'FAME II', "
                        "'Swachh Bharat', 'Jal Jeevan Mission', 'UJALA')"
                    ),
                },
                "category": {
                    "type": "string",
                    "description": "Scheme category if name not specified",
                    "enum": ["solar", "EV", "water", "waste", "energy", "agriculture"],
                },
            },
        },
    },
    {
        "name": "analyze_household",
        "description": (
            "Analyze a household profile and suggest a personalized eco action plan "
            "based on location, family size, and current habits. Use when the user "
            "wants a comprehensive plan or personalized recommendations."
        ),
        "parameters": {
            "type": "object",
            "properties": {
                "location": {
                    "type": "string",
                    "description": "Indian city name",
                },
                "members": {
                    "type": "integer",
                    "description": "Number of household members",
                },
                "habits": {
                    "type": "array",
                    "items": {"type": "string"},
                    "description": "Current eco habits the household already practices",
                },
            },
            "required": ["location", "members"],
        },
    },
]


# ===========================================================================
#  INDIAN GOVERNMENT ECO SCHEMES DATABASE
#  Static lookup for quick access. Falls back to web search if not found.
# ===========================================================================
SCHEMES_DB: dict[str, dict] = {
    "PM Surya Ghar": {
        "full_name": "PM Surya Ghar Muft Bijli Yojana",
        "category": "solar",
        "description": "Rooftop solar panel installation subsidy for residential homes",
        "benefit": "Up to 300 units of free electricity per month; 40% subsidy on installation cost for systems up to 3 kW",
        "eligibility": "Indian resident with own rooftop; must apply via portal pmsuryaghar.gov.in",
        "how_to_apply": "Register at pmsuryaghar.gov.in → get vendor quote → apply for subsidy → install → inspection → subsidy disbursed",
    },
    "FAME II": {
        "full_name": "Faster Adoption and Manufacturing of Electric Vehicles (FAME II)",
        "category": "EV",
        "description": "Subsidy for electric vehicles, especially 2-wheelers and 3-wheelers",
        "benefit": "Up to Rs 60,000 for electric 2-wheelers; up to Rs 1.5 lakh for electric 3-wheelers; varies by vehicle type",
        "eligibility": "Purchase of notified electric vehicles from registered dealers; vehicle must be registered in India",
        "how_to_apply": "Subsidy applied at point of purchase through registered dealer; no separate application needed",
    },
    "Swachh Bharat Mission": {
        "full_name": "Swachh Bharat Mission (Urban 2.0)",
        "category": "waste",
        "description": "National mission for sanitation and waste management",
        "benefit": "Free waste collection services; community composting support; toilet construction subsidies",
        "eligibility": "All urban households; waste collectors registered with ULB",
        "how_to_apply": "Contact local municipal corporation/ULB for waste collection registration; composting support via ward office",
    },
    "Jal Jeevan Mission": {
        "full_name": "Jal Jeevan Mission",
        "category": "water",
        "description": "Har Ghar Jal — functional tap water connection to every rural household",
        "benefit": "Functional tap water connection; water quality testing; community water supply management",
        "eligibility": "Rural households without functional tap water connection",
        "how_to_apply": "Apply through Gram Panchayat; village water and sanitation committee (VWSC) manages implementation",
    },
    "UJALA": {
        "full_name": "Ujala LED Bulb Distribution Scheme",
        "category": "energy",
        "description": "Subsidized LED bulb distribution across India",
        "benefit": "LED bulbs at Rs 10-15 per bulb (vs Rs 50-80 market price); 9W LED replaces 60W incandescent",
        "eligibility": "All Indian households; exchange old bulbs for LED at distribution centers",
        "how_to_apply": "Visit nearest EESL/Discom distribution center; exchange old incandescent/CFL for LED bulbs",
    },
    "National Biogas Programme": {
        "full_name": "National Biogas and Manure Management Programme (NBMMP)",
        "category": "waste",
        "description": "Subsidy for household biogas plants",
        "benefit": "40-60% capital subsidy on biogas plant installation; varies by category (SC/ST/Others)",
        "eligibility": "Rural households with cattle dung availability; SC/ST families get higher subsidy",
        "how_to_apply": "Apply through District Nodal Agency (DNA) or Block Development Officer (BDO)",
    },
    "PM e-DRIVE": {
        "full_name": "PM Electric Drive Revolution in Innovative Vehicle Enhancement (PM e-DRIVE)",
        "category": "EV",
        "description": "Successor to FAME II; extended EV subsidies and charging infrastructure",
        "benefit": "Demand incentive for EVs; EV charging infrastructure support; extends beyond 2024",
        "eligibility": "Same as FAME II; purchase of notified EVs from registered dealers",
        "how_to_apply": "Subsidy applied at point of purchase through registered dealer",
    },
    "KUSUM": {
        "full_name": "Kisan Urja Suraksha evam Utthaan Mahabhiyan (KUSUM)",
        "category": "solar",
        "description": "Solar pumps and grid-connected solar for farmers",
        "benefit": "60% subsidy on solar water pumps; 30% loan from banks; selling surplus solar power to DISCOM",
        "eligibility": "Farmer with existing grid-connected agriculture pump or need for new pump",
        "how_to_apply": "Apply through state agriculture department or MNRE portal; DISCOM tie-up for grid connectivity",
    },
}


# ===========================================================================
#  TOOL EXECUTOR
# ===========================================================================
def execute_tool(name: str, args: dict) -> str:
    """Execute a tool by name with given arguments and return result as string.



    Args:

        name: Tool name matching one of the TOOLS definitions.

        args: Dictionary of arguments matching the tool's parameter schema.



    Returns:

        Tool execution result as a formatted string.



    Raises:

        ValueError: Unknown tool name.

    """
    logger.info("Executing tool: %s with args: %s", name, args)

    if name == "calculate_impact":
        return _execute_calculate_impact(args)
    elif name == "get_recycling_guide":
        return _execute_recycling_guide(args)
    elif name == "web_search":
        return _execute_web_search(args)
    elif name == "check_scheme":
        return _execute_check_scheme(args)
    elif name == "analyze_household":
        return _execute_analyze_household(args)
    else:
        raise ValueError(f"Unknown tool: {name}")


def _execute_calculate_impact(args: dict) -> str:
    """Calculate impact for a single action or list of actions."""
    action_slug = args.get("action_slug", "")
    members = args.get("members", 1)

    # Support comma-separated slugs for multi-action queries
    slugs = [s.strip() for s in action_slug.split(",") if s.strip()]

    if not slugs:
        return "Error: No action slug provided."

    results = []
    for slug in slugs:
        entry = IMPACT_TABLE.get(slug)
        if not entry:
            results.append(f"- {slug}: Unknown action (not in impact table)")
            continue

        scaled_co2 = entry["co2_kg_year"] * members
        scaled_water = entry["water_L_day"] * members
        scaled_waste = entry["waste_kg_year"] * members

        results.append(
            f"- **{entry['label']}** ({slug}):\n"
            f"  CO2: {scaled_co2:.1f} kg/year | "
            f"Water: {scaled_water:.1f} L/day | "
            f"Waste: {scaled_waste:.1f} kg/year\n"
            f"  Source: [{entry['source']}] {entry['note']}"
        )

    header = f"Impact calculation for **{members}** household member(s):\n\n"
    return header + "\n".join(results)


def _execute_recycling_guide(args: dict) -> str:
    """Get recycling guide for a material and city."""
    material = args.get("material", "")
    city = args.get("city", "")

    if not material or not city:
        return "Error: Both material and city are required."

    if city not in INDIAN_CITIES:
        city_list = ", ".join(INDIAN_CITIES[:5]) + f", and {len(INDIAN_CITIES)-5} more"
        return (
            f"City '{city}' not in our database. "
            f"Supported cities: {city_list}. "
            f"Please try one of these cities."
        )

    try:
        guide = get_recycling_guide(material, city)
        return f"## Recycling Guide: {material} in {city}\n\n{guide}"
    except Exception as e:
        return f"Error getting recycling guide: {e}"


def _execute_web_search(args: dict) -> str:
    """Search the web using DuckDuckGo."""
    query = args.get("query", "")
    if not query:
        return "Error: Search query is required."

    try:
        # Try new ddgs package first, fall back to duckduckgo_search
        try:
            from ddgs import DDGS
        except ImportError:
            from duckduckgo_search import DDGS

        with DDGS() as ddgs:
            results = list(ddgs.text(query, max_results=5))

        if not results:
            return f"No results found for: {query}. Try a different search query."

        current_date = datetime.now().strftime("%B %d, %Y")
        formatted = [f"*Search conducted on: {current_date}*\n"]
        for i, r in enumerate(results, 1):
            title = r.get("title", "No title")
            body = r.get("body", "No description")
            url = r.get("href", "")
            formatted.append(f"{i}. **{title}**\n   {body}\n   {url}")

        return f"## Web Search Results: {query}\n\n" + "\n\n".join(formatted)

    except ImportError:
        return (
            "Web search is not available. "
            "Install ddgs: pip install ddgs"
        )
    except Exception as e:
        return f"Web search failed: {e}. Please try again or use check_scheme for known schemes."


def _execute_check_scheme(args: dict) -> str:
    """Look up Indian government eco scheme details."""
    scheme_name = args.get("scheme_name", "")
    category = args.get("category", "")

    # Direct lookup by name
    if scheme_name:
        # Fuzzy match: try exact, then partial
        scheme = SCHEMES_DB.get(scheme_name)
        if not scheme:
            for key in SCHEMES_DB:
                if scheme_name.lower() in key.lower() or key.lower() in scheme_name.lower():
                    scheme = SCHEMES_DB[key]
                    scheme_name = key
                    break

        if scheme:
            return _format_scheme(scheme_name, scheme)

        return (
            f"Scheme '{scheme_name}' not found in local database. "
            f"Use web_search tool to find current information about this scheme."
        )

    # Lookup by category
    if category:
        matching = [
            (name, s) for name, s in SCHEMES_DB.items()
            if s.get("category") == category
        ]
        if matching:
            formatted = []
            for name, scheme in matching:
                formatted.append(_format_scheme(name, scheme))
            return f"## Government Schemes: {category.title()}\n\n" + "\n\n---\n\n".join(formatted)
        return f"No schemes found for category: {category}. Try web_search for more options."

    # List all schemes
    formatted = []
    for name, scheme in SCHEMES_DB.items():
        formatted.append(f"- **{name}** ({scheme['category']}): {scheme['description']}")
    return "## Available Government Eco Schemes\n\n" + "\n".join(formatted)


def _format_scheme(name: str, scheme: dict) -> str:
    """Format a single scheme for display."""
    return (
        f"### {scheme.get('full_name', name)}\n\n"
        f"**Category:** {scheme['category'].title()}\n\n"
        f"**Description:** {scheme['description']}\n\n"
        f"**Benefits:** {scheme['benefit']}\n\n"
        f"**Eligibility:** {scheme['eligibility']}\n\n"
        f"**How to Apply:** {scheme['how_to_apply']}"
    )


def _execute_analyze_household(args: dict) -> str:
    """Analyze household and suggest personalized action plan using LLM."""
    location = args.get("location", "India")
    members = args.get("members", 1)
    habits = args.get("habits", [])

    # Build a list of actions the household does NOT already do
    all_actions = list(IMPACT_TABLE.keys())
    habit_slugs = set()
    for habit in habits:
        habit_lower = habit.lower()
        for slug, entry in IMPACT_TABLE.items():
            if habit_lower in entry["label"].lower() or slug in habit_lower:
                habit_slugs.add(slug)

    new_actions = [a for a in all_actions if a not in habit_slugs]

    # Use LLM to analyze and recommend
    analysis_prompt = (
        f"Analyze this Indian household and suggest a personalized eco action plan:\n\n"
        f"**Location:** {location}, India\n"
        f"**Household members:** {members}\n"
        f"**Current eco habits:** {', '.join(habits) if habits else 'None specified'}\n\n"
        f"**Available new actions** (not yet practiced):\n"
    )
    for slug in new_actions:
        entry = IMPACT_TABLE.get(slug, {})
        analysis_prompt += (
            f"- {slug}: {entry.get('label', slug)} — "
            f"{entry.get('co2_kg_year', 0) * members:.0f} kg CO2/year, "
            f"{entry.get('water_L_day', 0) * members:.0f} L water/day\n"
        )

    analysis_prompt += (
        f"\nProvide a **prioritized action plan** for this household:\n"
        f"1. Top 3 highest-impact actions they should start with\n"
        f"2. Quick wins (easy, low-cost)\n"
        f"3. Long-term investments (higher cost, higher impact)\n"
        f"4. Location-specific tips for {location}\n"
        f"5. Scale all numbers to {members} person(s)\n"
        f"\nBe specific, practical, and India-focused."
    )

    try:
        model = _get_model()
        response = model.chat(
            messages=[
                {"role": "system", "content": AGENT_INSTRUCTIONS.strip()},
                {"role": "user", "content": analysis_prompt},
            ],
            params={"max_tokens": 800, "temperature": 0.6},
        )
        analysis = response["choices"][0]["message"]["content"].strip()

        # Prepend summary
        impact = compute_session_impact(new_actions[:5], members)
        summary = (
            f"## Household Analysis: {members}-person household in {location}\n\n"
            f"**Current habits:** {', '.join(habits) if habits else 'None'}\n\n"
            f"**Potential impact** (top 5 new actions): "
            f"{impact['co2_kg_year']:.0f} kg CO2/year, "
            f"{impact['water_L_day']:.0f} L water/day\n\n"
            f"---\n\n"
        )
        return summary + analysis

    except Exception as e:
        return f"Error analyzing household: {e}. Falling back to basic recommendations."