searxng/searx/engines/open_meteo.py

# SPDX-License-Identifier: AGPL-3.0-or-later
"""Open Meteo (weather)"""

from urllib.parse import urlencode, quote_plus
from datetime import datetime
from flask_babel import gettext

from searx.network import get
from searx.exceptions import SearxEngineAPIException
from searx.result_types import EngineResults, Weather

about = {
    "website": 'https://open-meteo.com',
    "wikidata_id": None,
    "official_api_documentation": 'https://open-meteo.com/en/docs',
    "use_official_api": True,
    "require_api_key": False,
    "results": "JSON",
}

categories = ["weather"]

geo_url = "https://geocoding-api.open-meteo.com"
api_url = "https://api.open-meteo.com"

data_of_interest = "temperature_2m,relative_humidity_2m,apparent_temperature,cloud_cover,pressure_msl,wind_speed_10m,wind_direction_10m"  # pylint: disable=line-too-long


def request(query, params):
    location_url = f"{geo_url}/v1/search?name={quote_plus(query)}"

    resp = get(location_url)
    if resp.status_code != 200:
        raise SearxEngineAPIException("invalid geo location response code")

    json_locations = resp.json().get("results", [])
    if len(json_locations) == 0:
        raise SearxEngineAPIException("location not found")

    location = json_locations[0]
    args = {
        'latitude': location['latitude'],
        'longitude': location['longitude'],
        'timeformat': 'unixtime',
        'format': 'json',
        'current': data_of_interest,
        'forecast_days': 7,
        'hourly': data_of_interest,
    }

    params['url'] = f"{api_url}/v1/forecast?{urlencode(args)}"
    params['location'] = location['name']

    return params


def c_to_f(temperature):
    return "%.2f" % ((temperature * 1.8) + 32)


def get_direction(degrees):
    if degrees < 45 or degrees >= 315:
        return "N"

    if 45 <= degrees < 135:
        return "O"

    if 135 <= degrees < 225:
        return "S"

    return "W"


def build_condition_string(data):
    if data['relative_humidity_2m'] > 50:
        return "rainy"

    if data['cloud_cover'] > 30:
        return 'cloudy'

    return 'clear sky'


def generate_weather_data(data):
    return Weather.DataItem(
        condition=build_condition_string(data),
        temperature=f"{data['temperature_2m']}°C / {c_to_f(data['temperature_2m'])}°F",
        feelsLike=f"{data['apparent_temperature']}°C / {c_to_f(data['apparent_temperature'])}°F",
        wind=(
            f"{get_direction(data['wind_direction_10m'])}, "
            f"{data['wind_direction_10m']}° — "
            f"{data['wind_speed_10m']} km/h"
        ),
        pressure=f"{data['pressure_msl']}hPa",
        humidity=f"{data['relative_humidity_2m']}hPa",
        attributes={gettext('Cloud cover'): f"{data['cloud_cover']}%"},
    )


def response(resp):
    res = EngineResults()
    json_data = resp.json()

    current_weather = generate_weather_data(json_data['current'])
    weather_answer = Weather(
        location=resp.search_params['location'],
        current=current_weather,
    )

    for index, time in enumerate(json_data['hourly']['time']):
        hourly_data = {}

        for key in data_of_interest.split(","):
            hourly_data[key] = json_data['hourly'][key][index]

        forecast_data = generate_weather_data(hourly_data)
        forecast_data.time = datetime.fromtimestamp(time).strftime('%Y-%m-%d %H:%M')
        weather_answer.forecasts.append(forecast_data)

    res.add(weather_answer)
    return res