-
Notifications
You must be signed in to change notification settings - Fork 2
/
api_get_weatherapi.py
186 lines (160 loc) · 6.92 KB
/
api_get_weatherapi.py
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
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
# api_get_weatherapi.py
# ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
# github.com/FlyingFathead/TelegramBot-OpenAI-API/
# ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
#
# >>> weather fetcher module version: v0.708
# >>> (Updated May 25 2024)
#
# This API functionality requires WeatherAPI key.
# You can get the API key from the corresponding service provider.
# Once you have the API key, add it to your environment variables:
# export WEATHERAPI_KEY="<your API key>"
# (or on i.e. Linux, add to your `~/.bashrc`: export WEATHERAPI_KEY="<your API key>" )
#
# (Updated May 25, 2024)
import httpx
import os
import logging
# Function to check for WeatherAPI key
def get_weatherapi_key():
api_key = os.getenv('WEATHERAPI_KEY')
if not api_key:
logging.error("[WARNING] WeatherAPI key not set. You need to set the 'WEATHERAPI_KEY' environment variable to use WeatherAPI functionalities!")
return None
return api_key
# Dictionary to translate moon phases from English to Finnish
moon_phase_translation = {
"New Moon": "uusikuu",
"Waxing Crescent": "kasvava sirppi",
"First Quarter": "ensimmäinen neljännes",
"Waxing Gibbous": "kasvava puolikuu",
"Full Moon": "täysikuu",
"Waning Gibbous": "vähenevä puolikuu",
"Last Quarter": "viimeinen neljännes",
"Waning Crescent": "vähenevä sirppi"
}
# get moon phase data
async def get_moon_phase(lat, lon):
api_key = get_weatherapi_key()
if not api_key:
return None
logging.info(f"Fetching moon phase data for coordinates: Latitude: {lat}, Longitude: {lon}")
base_url = 'http://api.weatherapi.com/v1/astronomy.json'
url = f"{base_url}?key={api_key}&q={lat},{lon}"
async with httpx.AsyncClient() as client:
response = await client.get(url)
logging.info(f"Moon phase response status: {response.status_code}")
if response.status_code == 200:
data = response.json()
logging.info(f"Moon phase data: {data}")
moon_phase = data['astronomy']['astro']['moon_phase']
translated_moon_phase = moon_phase_translation.get(moon_phase, moon_phase)
return translated_moon_phase
else:
logging.error(f"Failed to fetch moon phase data: {response.text}")
return None
# get timezone for the coordinates
async def get_timezone(lat, lon):
api_key = get_weatherapi_key()
if not api_key:
return None
logging.info(f"Fetching timezone data for coordinates: Latitude: {lat}, Longitude: {lon}")
base_url = 'http://api.weatherapi.com/v1/timezone.json'
url = f"{base_url}?key={api_key}&q={lat},{lon}"
async with httpx.AsyncClient() as client:
response = await client.get(url)
logging.info(f"Timezone response status: {response.status_code}")
if response.status_code == 200:
data = response.json()
logging.info(f"Timezone data: {data}")
timezone = data['location']['tz_id']
return timezone
else:
logging.error(f"Failed to fetch timezone data: {response.text}")
return None
# get daily forecast, safety alerts, and air quality index
async def get_daily_forecast(location):
api_key = get_weatherapi_key()
if not api_key:
return None
logging.info(f"Fetching daily forecast data for location: {location}")
base_url = 'http://api.weatherapi.com/v1/forecast.json'
url = f"{base_url}?key={api_key}&q={location}&days=1&alerts=yes&aqi=yes"
async with httpx.AsyncClient() as client:
response = await client.get(url)
logging.info(f"Daily forecast response status: {response.status_code}")
if response.status_code == 200:
data = response.json()
logging.info(f"Daily forecast data: {data}")
forecast = data['forecast']['forecastday'][0]
current = data['current']
alerts = data.get('alerts', {})
air_quality = current['air_quality']
return {
'date': forecast['date'],
'temperature': forecast['day']['avgtemp_c'],
'condition': forecast['day']['condition']['text'],
'wind': forecast['day']['maxwind_kph'],
'precipitation': forecast['day']['totalprecip_mm'],
'uv_index': forecast['day']['uv'],
'air_quality': air_quality,
'alerts': alerts
}
else:
logging.error(f"Failed to fetch daily forecast data: {response.text}")
return None
# get current weather including UV index
async def get_current_weather_via_weatherapi(location):
api_key = get_weatherapi_key()
if not api_key:
return None
logging.info(f"Fetching current weather data for location: {location}")
base_url = 'http://api.weatherapi.com/v1/current.json'
url = f"{base_url}?key={api_key}&q={location}"
async with httpx.AsyncClient() as client:
response = await client.get(url)
logging.info(f"Current weather response status: {response.status_code}")
if response.status_code == 200:
data = response.json()
logging.info(f"Current weather data: {data}")
current = data['current']
return {
'temperature': current['temp_c'],
'condition': current['condition']['text'],
'wind': current['wind_kph'],
'precipitation': current['precip_mm'],
'uv_index': current['uv'],
'visibility': current['vis_km'], # Added visibility data
'air_quality': current.get('air_quality', {})
}
else:
logging.error(f"Failed to fetch current weather data: {response.text}")
return None
# get astronomy data including moonrise, moonset, and moon illumination
async def get_astronomy_data(lat, lon):
api_key = get_weatherapi_key()
if not api_key:
return None
logging.info(f"Fetching astronomy data for coordinates: Latitude: {lat}, Longitude: {lon}")
base_url = 'http://api.weatherapi.com/v1/astronomy.json'
url = f"{base_url}?key={api_key}&q={lat},{lon}"
async with httpx.AsyncClient() as client:
response = await client.get(url)
logging.info(f"Astronomy response status: {response.status_code}")
if response.status_code == 200:
data = response.json()
logging.info(f"Astronomy data: {data}")
astro = data['astronomy']['astro']
moonrise = astro['moonrise']
moonset = astro['moonset']
moon_illumination = astro['moon_illumination']
return {
'moonrise': moonrise,
'moonset': moonset,
'moon_illumination': moon_illumination
}
else:
logging.error(f"Failed to fetch astronomy data: {response.text}")
return None
# Additional WeatherAPI-related functions can be added here