tools/weather_info_tool.py

import os
from typing import List
from dotenv import load_dotenv
from langchain.tools import tool
from logger.decorators import log_entry
from utils.weather_info import WeatherForecastTool

class WeatherInfoTool:
    """
    A wrapper class for retrieving weather information using the OpenWeatherMap API.

    This class provides LangChain-compatible tools to get current weather and weather forecasts
    for a given city.
    """

    def __init__(self):
        """
        Initialize the WeatherInfoTool with API credentials and set up weather-related tools.
        """
        load_dotenv()
        self.api_key = os.environ.get("OPENWEATHERMAP_API_KEY")
        self.weather_service = WeatherForecastTool(self.api_key)
        self.weather_tool_list = self._setup_tools()

    def _setup_tools(self) -> List:
        """
        Define and register weather-related LangChain tool functions.

        Returns:
            List: A list of LangChain-compatible weather tool functions.
        """

        @tool
        @log_entry
        def get_current_weather(city: str) -> str:
            """
            Get the current weather information for a specified city.

            Args:
                city (str): The name of the city to get current weather data for.

            Returns:
                str: A string describing the current temperature and weather condition.
            """
            print('Entered into get_current_weather().')
            weather_data = self.weather_service.get_current_weather(city)
            if weather_data:
                temp = weather_data.get('main', {}).get('temp', 'N/A')
                desc = weather_data.get('weather', [{}])[0].get('description', 'N/A')
                return f"Current weather in {city}: {temp}°C, {desc}"
            return f"Could not fetch weather for {city}"

        @tool
        @log_entry
        def get_weather_forecast(city: str) -> str:
            """
            Get the multi-day weather forecast for a specified city.

            Args:
                city (str): The name of the city to get the weather forecast for.

            Returns:
                str: A string listing daily temperatures and descriptions for upcoming forecasts.
            """
            print('Entered into get_weather_forecast().')
            forecast_data = self.weather_service.get_forecast_weather(city)
            if forecast_data and 'list' in forecast_data:
                forecast_summary = []
                for item in forecast_data['list']:
                    date = item['dt_txt'].split(' ')[0]
                    temp = item['main']['temp']
                    desc = item['weather'][0]['description']
                    forecast_summary.append(f"{date}: {temp}°C, {desc}")
                return f"Weather forecast for {city}:\n" + "\n".join(forecast_summary)
            return f"Could not fetch forecast for {city}"

        return [get_current_weather, get_weather_forecast]