-
Notifications
You must be signed in to change notification settings - Fork 2
/
api_get_openweathermap.py
359 lines (299 loc) · 17.8 KB
/
api_get_openweathermap.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
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
# api_get_openweathermap.py
# ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
# github.com/FlyingFathead/TelegramBot-OpenAI-API/
# ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
#
# >>> weather fetcher module version: v0.708
# >>> (Updated May 25 2024)
#
# This API functionality requires both OpenWeatherMap and MapTiler API keys.
# You can get both from the corresponding service providers.
# Once you have the API keys, add them to your environment variables:
# export OPENWEATHERMAP_API_KEY="<your API key>"
# export MAPTILER_API_KEY="<your API key>"
# date & time utils
import datetime as dt
from dateutil import parser
from timezonefinder import TimezoneFinder
import pytz
import json
import httpx
import os
import logging
import openai
# Stuff we want to get via WeatherAPI:
from api_get_weatherapi import get_moon_phase, get_timezone, get_daily_forecast, get_current_weather_via_weatherapi, get_astronomy_data
# Import the additional data fetching function for Finland
from api_get_additional_weather_data import get_additional_data_dump
# get the combined weather
async def get_weather(city_name, country, exclude='', units='metric', lang='fi'):
api_key = os.getenv('OPENWEATHERMAP_API_KEY')
if not api_key:
logging.error("[WARNING] OpenWeatherMap API key not set. You need to set the 'OPENWEATHERMAP_API_KEY' environment variable to use OpenWeatherMap API functionalities!")
return "OpenWeatherMap API key not set."
logging.info(f"Fetching weather data for city: {city_name}, Country: {country}")
if not city_name or not country or city_name.lower() in ["defaultcity", ""]:
return "Please provide a valid city name and country."
base_url = 'http:https://api.openweathermap.org/data/2.5/'
lat, lon, resolved_country = await get_coordinates(city_name, country=country)
if lat is None or lon is None or resolved_country is None:
logging.info("Failed to retrieve coordinates or country.")
return "[Unable to retrieve coordinates or country for the specified location. Ask the user for clarification.]"
current_weather_url = f"{base_url}weather?lat={lat}&lon={lon}&appid={api_key}&units={units}&lang={lang}"
forecast_url = f"{base_url}forecast?lat={lat}&lon={lon}&appid={api_key}&units={units}&lang={lang}"
async with httpx.AsyncClient() as client:
current_weather_response = await client.get(current_weather_url)
forecast_response = await client.get(forecast_url)
logging.info(f"Current weather response status: {current_weather_response.status_code}")
logging.info(f"Forecast response status: {forecast_response.status_code}")
if current_weather_response.status_code == 200 and forecast_response.status_code == 200:
current_weather_data = current_weather_response.json()
forecast_data = forecast_response.json()
moon_phase_data = await get_moon_phase(lat, lon)
daily_forecast_data = await get_daily_forecast(f"{lat},{lon}")
current_weather_data_from_weatherapi = await get_current_weather_via_weatherapi(f"{lat},{lon}")
astronomy_data = await get_astronomy_data(lat, lon) # Fetch astronomy data
# Fetch additional data for Finland
additional_data = ""
if resolved_country.lower() == "fi": # Case-insensitive check
logging.info("Fetching additional weather data for Finland.")
additional_data = await get_additional_data_dump()
logging.info(f"Additional data fetched: {additional_data}")
combined_data = await combine_weather_data(city_name, resolved_country, lat, lon, current_weather_data, forecast_data, moon_phase_data, daily_forecast_data, current_weather_data_from_weatherapi, astronomy_data, additional_data)
return combined_data
else:
logging.error(f"Failed to fetch weather data: {current_weather_response.text} / {forecast_response.text}")
return "[Inform the user that data fetching the weather data failed, current information could not be fetched. Reply in the user's language.]"
# get coordinates
async def get_coordinates(city_name, country=None):
lat = lon = None
resolved_country = None
logging.info(f"Coordinates for {city_name}, {country}: Latitude: {lat}, Longitude: {lon}")
api_key = os.getenv('MAPTILER_API_KEY')
if not api_key:
logging.info("[WARNING] MapTiler API key not set. You need to set the 'MAPTILER_API_KEY' environment variable in order to be able to use coordinate lookups, i.e. for weather data!")
return None, None, None
query = f"{city_name}"
if country:
query += f", {country}"
geocode_url = f"https://api.maptiler.com/geocoding/{query}.json?key={api_key}"
logging.info(f"Making API request to URL: {geocode_url}")
async with httpx.AsyncClient() as client:
response = await client.get(geocode_url)
logging.info(f"Received response with status code: {response.status_code}")
if response.status_code == 200:
data = response.json()
logging.info(f"Response data: {data}")
if data['features']:
feature = data['features'][0]
lat = feature['geometry']['coordinates'][1]
lon = feature['geometry']['coordinates'][0]
resolved_country = feature['properties'].get('country_code', 'Country not available')
logging.info(f"Coordinates for {city_name}, {resolved_country}: Latitude: {lat}, Longitude: {lon}")
return lat, lon, resolved_country
else:
logging.error("No features found in the geocoding response.")
return None, None, None
else:
logging.error(f"Failed to fetch coordinates: {response.text}")
return None, None, None
# the function below can be implemented to use for POI lookups
async def get_location_info_from_coordinates(latitude, longitude):
logging.info(f"Fetching location information for coordinates: Latitude: {latitude}, Longitude: {longitude}")
# Retrieve MapTiler API key from environment variables
api_key = os.getenv('MAPTILER_API_KEY')
if not api_key:
logging.info("[WARNING] MapTiler API key not set. You need to set the 'MAPTILER_API_KEY' environment variable for this function to work!")
return "MapTiler API key not set."
# Construct the API request URL for reverse geocoding
reverse_geocode_url = f"https://api.maptiler.com/geocoding/{longitude},{latitude}.json?key={api_key}"
logging.info(f"Making API request to URL: {reverse_geocode_url}")
async with httpx.AsyncClient() as client:
response = await client.get(reverse_geocode_url)
logging.info(f"Received response with status code: {response.status_code}")
if response.status_code == 200:
data = response.json()
logging.info(f"Response data: {data}")
# Process the response data to extract useful information
# For example, you might extract the nearest city name, points of interest, etc.
# Return this information
return data
else:
logging.info(f"Failed to fetch location information: {response.text}")
return "Failed to fetch location information."
# Format and return detailed weather information along with location data
def format_weather_response(city_name, country, weather_info):
# Example of how you might construct the message with location and weather data
location_info = f"[{city_name}, {country}]\n\n"
return f"{location_info} {weather_info}"
# wind direction from degrees to cardinal
def degrees_to_cardinal(d):
dirs = ['N', 'NNE', 'NE', 'ENE', 'E', 'ESE', 'SE', 'SSE', 'S', 'SSW', 'SW', 'WSW', 'W', 'WNW', 'NW', 'NNW']
ix = int((d + 11.25)/22.5 - 0.02) # Subtract a small epsilon to correct edge case at North (360 degrees)
return dirs[ix % 16]
# Function to convert 12-hour AM/PM time to 24-hour time
def convert_to_24_hour(time_str, timezone_str):
try:
dt_time = dt.datetime.strptime(time_str, '%I:%M %p')
local_timezone = pytz.timezone(timezone_str)
local_time = local_timezone.localize(dt_time)
formatted_time = local_time.strftime('%H:%M')
logging.info(f"Converted time {time_str} to {formatted_time} in timezone {timezone_str}")
return formatted_time
except Exception as e:
logging.error(f"Error converting time string {time_str}: {e}")
return "Invalid time"
# combined weather data
async def combine_weather_data(city_name, country, lat, lon, current_weather_data, forecast_data, moon_phase_data, daily_forecast_data, current_weather_data_from_weatherapi, astronomy_data, additional_data):
tf = TimezoneFinder()
timezone_str = tf.timezone_at(lat=lat, lng=lon) # get timezone using the coordinates
local_timezone = pytz.timezone(timezone_str)
# Current weather details from OpenWeatherMap
weather_description = current_weather_data['weather'][0]['description']
temperature = current_weather_data['main']['temp']
feels_like = current_weather_data['main']['feels_like']
temp_min = current_weather_data['main']['temp_min']
temp_max = current_weather_data['main']['temp_max']
pressure = current_weather_data['main']['pressure']
humidity = current_weather_data['main']['humidity']
wind_speed = current_weather_data['wind']['speed']
wind_direction = current_weather_data['wind']['deg']
wind_direction_cardinal = degrees_to_cardinal(wind_direction)
visibility = current_weather_data.get('visibility', 'N/A')
snow_1h = current_weather_data.get('snow', {}).get('1h', 'N/A')
# Data to get from WeatherAPI
uv_index = current_weather_data_from_weatherapi['uv_index']
visibility_wapi = current_weather_data_from_weatherapi['visibility']
condition_wapi = current_weather_data_from_weatherapi['condition']
# Astronomy data
moonrise_time = convert_to_24_hour(astronomy_data['moonrise'], timezone_str)
moonset_time = convert_to_24_hour(astronomy_data['moonset'], timezone_str)
moon_illumination = astronomy_data['moon_illumination']
sunrise_time_utc = dt.datetime.utcfromtimestamp(current_weather_data['sys']['sunrise'])
sunset_time_utc = dt.datetime.utcfromtimestamp(current_weather_data['sys']['sunset'])
sunrise_time_local = sunrise_time_utc.replace(tzinfo=pytz.utc).astimezone(local_timezone)
sunset_time_local = sunset_time_utc.replace(tzinfo=pytz.utc).astimezone(local_timezone)
sunrise_time_local_str = sunrise_time_local.strftime('%H:%M')
sunset_time_local_str = sunset_time_local.strftime('%H:%M')
temp_fahrenheit = (temperature * 9/5) + 32
feels_like_fahrenheit = (feels_like * 9/5) + 32
temp_min_fahrenheit = (temp_min * 9/5) + 32
temp_max_fahrenheit = (temp_max * 9/5) + 32
country_code = current_weather_data['sys']['country']
country_info = f"Country: {country_code}"
coordinates_info = f"lat: {lat}, lon: {lon}"
# Get current UTC and local times
current_time_utc = dt.datetime.utcnow()
current_time_local = current_time_utc.replace(tzinfo=pytz.utc).astimezone(local_timezone)
current_time_utc_str = current_time_utc.strftime('%Y-%m-%d %H:%M:%S')
current_time_local_str = current_time_local.strftime('%Y-%m-%d %H:%M:%S')
detailed_weather_info = (
f"Sää paikassa {city_name}, {country_code} (UTC: {current_time_utc_str}, Paikallinen aika: {current_time_local_str}): {weather_description}, "
f"Lämpötila: {temperature}°C / {temp_fahrenheit:.1f}°F (Tuntuu kuin: {feels_like}°C / {feels_like_fahrenheit:.1f}°F), "
f"Minimi: {temp_min}°C / {temp_min_fahrenheit:.1f}°F, Maksimi: {temp_max}°C / {temp_max_fahrenheit:.1f}°F, "
f"Ilmanpaine: {pressure} hPa, Ilmankosteus: {humidity}%, "
f"Tuulen nopeus: {wind_speed} m/s, Tuulen suunta: {wind_direction} astetta ({wind_direction_cardinal}), "
f"Näkyvyys: {visibility} metriä [OpenWeatherMap] | {visibility_wapi} km [WeatherAPI], "
f"Lumisade (viimeisen 1h aikana): {snow_1h} mm, "
f"Auringonnousu: {sunrise_time_local_str}, "
f"Auringonlasku: {sunset_time_local_str}, "
f"Koordinaatit: {coordinates_info} (Maa: {country_info}), "
f"Kuun vaihe: {moon_phase_data}, "
f"UV-indeksi [WeatherAPI]: {uv_index}, " # Include UV index in the detailed weather info
f"Sääolosuhteet [WeatherAPI]: {condition_wapi}, " # Include 'condition' from WeatherAPI
f"Kuu nousee klo (paikallista aikaa): {moonrise_time}, "
f"Kuu laskee klo (paikallista aikaa): {moonset_time}, "
f"Kuun valaistus: {moon_illumination}%"
)
# Include additional WeatherAPI data (daily forecast, air quality, and alerts)
if daily_forecast_data:
air_quality_data = daily_forecast_data['air_quality']
alerts = daily_forecast_data['alerts']
air_quality_info = "\n(Ilmanlaatu: " + " / ".join(
[f"{key}: {value}" for key, value in air_quality_data.items()]
) + ")"
# air_quality_info = "\nIlmanlaatu:\n" + "\n".join(
# [f"{key}: {value}" for key, value in air_quality_data.items()]
# )
alerts_info = "\nSäävaroitukset:\n" + (
"\n".join(
[f"Alert: {alert['headline']}\nDescription: {alert['desc']}\nInstructions: {alert['instruction']}\n"
for alert in alerts['alert']]
) if 'alert' in alerts and alerts['alert'] else "No weather alerts."
)
detailed_weather_info += f"\n{air_quality_info}\n{alerts_info}"
# 3-hour forecast details
forecasts = forecast_data['list']
formatted_forecasts = []
for forecast_data in forecasts[:5]: # Adjust the range as needed
utc_time = dt.datetime.utcfromtimestamp(forecast_data['dt'])
local_time = utc_time.replace(tzinfo=pytz.utc).astimezone(local_timezone)
local_time_str = local_time.strftime('%Y-%m-%d %H:%M:%S')
utc_time_str = utc_time.strftime('%Y-%m-%d %H:%M:%S')
temp = forecast_data['main']['temp']
temp_fahrenheit = (temp * 9/5) + 32
description = forecast_data['weather'][0]['description']
wind_speed = forecast_data['wind']['speed']
humidity = forecast_data['main']['humidity']
pressure = forecast_data['main']['pressure']
clouds = forecast_data['clouds']['all']
rain = forecast_data.get('rain', {}).get('3h', 'N/A')
formatted_forecasts.append(
f"- {city_name}: {local_time_str} (Local time, 24hr format) / {utc_time_str} (UTC, 24hr format): Lämpötila: {temp}°C / {temp_fahrenheit:.1f}°F, {description.capitalize()}, Tuuli: {wind_speed} m/s, "
f"Ilmanpaine: {pressure} hPa, Ilmankosteus: {humidity}%, Pilvisyys: {clouds}%, "
f"Sade (viimeisen 3h aikana): {rain} mm"
)
final_forecast = f"Kolmen tunnin sääennuste, {city_name}:\n" + "\n".join(formatted_forecasts)
additional_info_to_add = (
"NOTE: TRANSLATE AND FORMAT THIS DATA FOR THE USER AS APPROPRIATE. Use emojis where suitable to enhance the readability and engagement of the weather report. "
"For example, use 🌞 for sunny, 🌧️ for rain, ⛅ for partly cloudy, etc and include a relevant and concise overview of what was asked."
)
combined_info = f"{detailed_weather_info}\n\n{final_forecast}"
# Append additional data for Finland if available
if additional_data:
combined_info += f"\n\n[ Lisätiedot Suomeen (lähde: foreca.fi -- MAINITSE LÄHDE) ]\n{additional_data}"
# Append the additional info at the end
combined_info += f"\n\n{additional_info_to_add}"
logging.info(f"Formatted combined weather data being sent: {combined_info}")
return combined_info
# Format the weather information and translate it if necessary.
async def format_and_translate_weather(bot, user_request, weather_info):
# System message to instruct the model
format_translate_system_message = {
"role": "system",
"content": "Translate if needed (depending on user's language) and format the data into a digestable Telegram message with emoji symbols and html parsemode tags. Use i.e. <b>type</b> etc. Respond in user's original language, DO NOT OMIT DETAILS! INCLUDE THE COUNTRY INFO IF AVAILABLE."
}
# Prepare chat history with the user's request, system message, and weather info
chat_history = [
{"role": "user", "content": user_request},
format_translate_system_message,
{"role": "assistant", "content": weather_info}
]
# Prepare the payload for the OpenAI API
payload = {
"model": bot.model,
"messages": chat_history,
"temperature": 0.5
}
headers = {
"Content-Type": "application/json",
"Authorization": f"Bearer {openai.api_key}"
}
# Make the API request
async with httpx.AsyncClient() as client:
response = await client.post("https://api.openai.com/v1/chat/completions",
data=json.dumps(payload),
headers=headers,
timeout=bot.timeout)
response_json = response.json()
# Extract the formatted and potentially translated response
if response.status_code == 200 and 'choices' in response_json:
translated_reply = response_json['choices'][0]['message']['content'].strip()
bot_token_count = bot.count_tokens(translated_reply) # Count the tokens in the translated reply
bot.total_token_usage += bot_token_count # Add to the total token usage
bot.write_total_token_usage(bot.total_token_usage) # Update the total token usage file
logging.info(f"Sent this weather report to user: {translated_reply}")
return translated_reply
else:
logging.error("Error in formatting and translating weather data.")
return weather_info # Return the original weather info in case of error