qweather.py

Overview

The qweather.py file implements a component named QWeather that fetches weather-related data from the QWeather API. It is designed as part of a larger system (likely InfiniFlow) that integrates external weather information services for use in an agent or workflow environment.

This component supports retrieving three types of information:

• Current or forecasted weather conditions,

• Various weather indices (e.g., comfort, UV index),

• Current air quality data.

The component manages API parameters, handles HTTP requests to QWeather's REST API endpoints, processes responses, handles errors gracefully, and outputs results either as strings or Pandas DataFrames depending on the query type.

Classes and Methods

QWeatherParam (inherits ComponentParamBase)

Defines and validates parameters for the QWeather component.

Properties

• web_apikey (str): API key for accessing the QWeather service. Default "xxx".

• lang (str): Language code for response data. Default "zh". Supports many languages including English ("en"), German ("de"), Japanese ("ja"), etc.

• type (str): Type of data to retrieve. One of "weather", "indices", or "airquality". Default "weather".

• user_type (str): Subscription type, either "free" (uses dev API endpoint) or "paid" (uses production API endpoint). Default "free".

• time_period (str): Time period for weather data when type is "weather". Options: "now", "3d", "7d", "10d", "15d", "30d". Default "now".

• error_code (dict): Maps QWeather API error codes to human-readable error messages.

Methods

• init(self): Initializes all parameters with default values.

• check(self): Validates parameter values to ensure they are non-empty and within allowed options.

  • Checks if web_apikey is non-empty.

  • Validates type is one of "weather", "indices", or "airquality".

  • Validates user_type is "free" or "paid".

  • Validates lang against supported language codes.

  • Validates time_period for weather queries.

Usage example

params = QWeatherParam()
params.web_apikey = "your_actual_api_key"
params.type = "weather"
params.time_period = "7d"
params.check()  # Raises error if invalid

QWeather (inherits ComponentBase and ABC)

Main component class responsible for querying the QWeather API and formatting output.

Class Attribute

• component_name (str): Set to "QWeather".

Methods

• _run(self, history, **kwargs)

  Executes the component logic:

  1. Retrieves input location name from upstream components or workflow input.

  2. Calls QWeather City Lookup API to resolve the location name to a location ID.

  3. Based on QWeatherParam settings (type, user_type, time_period), calls the appropriate QWeather API endpoint:

     • Weather: /weather/{time_period} endpoint returns current or forecasted weather.

     • Indices: /indices/1d endpoint returns daily weather indices.

     • Airquality: /air/now endpoint returns current air quality.

  4. Parses the JSON response, checks for errors, and formats output.

     • For "weather" with time_period=="now", returns a string of the current weather.

     • For "weather" with forecast periods, returns a Pandas DataFrame with daily forecast data.

     • For "indices" and "airquality", returns formatted strings.

  5. Handles API errors or exceptions by returning an error message wrapped with "Error" prefix.

  Parameters:

  • history: Unused in this method but typically represents execution context or previous states.

  • **kwargs: Additional keyword arguments passed to the method (not used explicitly).

  Returns:

  • Output data wrapped with the ComponentBase.be_output() method, which presumably formats or packages it for the rest of the system.

Usage example

component = QWeather()
component._param = QWeatherParam()
component._param.web_apikey = "your_api_key"
component._param.type = "weather"
component._param.time_period = "3d"
component._param.user_type = "free"
component._param.lang = "en"

# Simulate input fetching mechanism inside get_input()
def mock_get_input():
    return {"content": ["Shanghai"]}

component.get_input = mock_get_input

output = component._run(history=None)
print(output)

Important Implementation Details

• API Endpoints:

  • City lookup: https://geoapi.qweather.com/v2/city/lookup

  • Weather: https://api.qweather.com/v7/weather/{time_period} (paid) or https://devapi.qweather.com/v7/weather/{time_period} (free)

  • Indices: .../indices/1d

  • Air quality: .../air/now

• The component distinguishes between free and paid API access by switching the base URL accordingly.

• When requesting weather data for forecast periods (not "now"), it converts the daily forecast list into a Pandas DataFrame where each row corresponds to a forecast item wrapped as a dictionary with "content" keys.

• Error handling includes mapping known API error codes to descriptive messages and catching exceptions during HTTP requests.

• The class uses requests for HTTP GET requests and pandas for data handling.

Interaction with Other Parts of the System

• Inherits from ComponentBase and ComponentParamBase from agent.component.base, indicating it is a modular component in a larger agent framework.

• Uses a method get_input() from the base class or environment to fetch input data (presumably user input or prior workflow output).

• Uses ComponentBase.be_output() to format output consistently for downstream consumption.

• Depends on external libraries:

  • requests for HTTP calls to QWeather APIs.

  • pandas for data structuring and returning tabular data.

• The component is likely used in workflows requiring weather data integration, possibly invoked dynamically with parameters set per user or task.

Diagram: Class Structure of qweather.py

classDiagram
    class ComponentParamBase {
        <<abstract>>
        +check_empty(value, name)
        +check_valid_value(value, name, valid_options)
    }

    class ComponentBase {
        <<abstract>>
        +get_input()
        +be_output(content)
    }

    class QWeatherParam {
        -web_apikey: str
        -lang: str
        -type: str
        -user_type: str
        -error_code: dict
        -time_period: str
        +__init__()
        +check()
    }

    class QWeather {
        -component_name: str = "QWeather"
        +_run(history, **kwargs)
    }

    QWeatherParam --|> ComponentParamBase
    QWeather --|> ComponentBase
    QWeather --|> ABC

Summary

The qweather.py file provides a specialized component for integrating QWeather API data into an agent-based system. It encapsulates parameter management, API interaction, error handling, and output formatting for weather, indices, and air quality data. Its modular design allows it to fit cleanly into larger workflows that require dynamic, localized weather information.