summaryrefslogtreecommitdiff
path: root/searx/engines/wttr.py
blob: 9c7f69b43513a91ee0ca1ebcf1ffc10fcd636ce5 (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
# SPDX-License-Identifier: AGPL-3.0-or-later
"""wttr.in (weather forecast service)"""

import typing as t

from urllib.parse import quote
from datetime import datetime

from searx.result_types import EngineResults, WeatherAnswer
from searx import weather

about = {
    "website": "https://wttr.in",
    "wikidata_id": "Q107586666",
    "official_api_documentation": "https://github.com/chubin/wttr.in#json-output",
    "use_official_api": True,
    "require_api_key": False,
    "results": "JSON",
}

categories = ["weather"]

url = "https://wttr.in/{query}?format=j1&lang={lang}"

# adapted from https://github.com/chubin/wttr.in/blob/master/lib/constants.py
WWO_TO_CONDITION: dict[str, weather.WeatherConditionType] = {
    "113": "clear sky",
    "116": "partly cloudy",
    "119": "cloudy",
    "122": "fair",
    "143": "fair",
    "176": "light rain showers",
    "179": "light snow showers",
    "182": "light sleet showers",
    "185": "light sleet",
    "200": "rain and thunder",
    "227": "light snow",
    "230": "heavy snow",
    "248": "fog",
    "260": "fog",
    "263": "light rain showers",
    "266": "light rain showers",
    "281": "light sleet showers",
    "284": "light snow showers",
    "293": "light rain showers",
    "296": "light rain",
    "299": "rain showers",
    "302": "rain",
    "305": "heavy rain showers",
    "308": "heavy rain",
    "311": "light sleet",
    "314": "sleet",
    "317": "light sleet",
    "320": "heavy sleet",
    "323": "light snow showers",
    "326": "light snow showers",
    "329": "heavy snow showers",
    "332": "heavy snow",
    "335": "heavy snow showers",
    "338": "heavy snow",
    "350": "light sleet",
    "353": "light rain showers",
    "356": "heavy rain showers",
    "359": "heavy rain",
    "362": "light sleet showers",
    "365": "sleet showers",
    "368": "light snow showers",
    "371": "heavy snow showers",
    "374": "light sleet showers",
    "377": "heavy sleet",
    "386": "rain showers and thunder",
    "389": "heavy rain showers and thunder",
    "392": "snow showers and thunder",
    "395": "heavy snow showers",
}


def request(query, params):
    params["url"] = url.format(query=quote(query), lang=params["language"])
    params["raise_for_httperror"] = False

    return params


def _weather_data(location: weather.GeoLocation, data: dict[str, t.Any]):
    # the naming between different data objects is inconsitent, thus temp_C and
    # tempC are possible
    tempC: float = data.get("temp_C") or data.get("tempC")  # type: ignore

    return WeatherAnswer.Item(
        location=location,
        temperature=weather.Temperature(val=tempC, unit="°C"),
        condition=WWO_TO_CONDITION[data["weatherCode"]],
        feels_like=weather.Temperature(val=data["FeelsLikeC"], unit="°C"),
        wind_from=weather.Compass(int(data["winddirDegree"])),
        wind_speed=weather.WindSpeed(val=data["windspeedKmph"], unit="km/h"),
        pressure=weather.Pressure(val=data["pressure"], unit="hPa"),
        humidity=weather.RelativeHumidity(data["humidity"]),
        cloud_cover=data["cloudcover"],
    )


def response(resp):
    res = EngineResults()

    if resp.status_code == 404:
        return res

    json_data = resp.json()
    geoloc = weather.GeoLocation.by_query(resp.search_params["query"])

    weather_answer = WeatherAnswer(
        current=_weather_data(geoloc, json_data["current_condition"][0]),
        service="wttr.in",
    )

    for day in json_data["weather"]:
        date = datetime.fromisoformat(day["date"])
        time_slot_len = 24 // len(day["hourly"])
        for index, forecast in enumerate(day["hourly"]):
            forecast_data = _weather_data(geoloc, forecast)
            forecast_data.datetime = weather.DateTime(date.replace(hour=index * time_slot_len + 1))
            weather_answer.forecasts.append(forecast_data)

    res.add(weather_answer)
    return res