1
+ import aiohttp
2
+ import asyncio
3
+ import logging
4
+ import argparse
5
+ from win10toast import ToastNotifier
6
+
7
+ logging .basicConfig (level = logging .INFO )
8
+ notifier = ToastNotifier ()
9
+
10
+ async def fetch_data (url ):
11
+ API_KEY = "apikey=0zN2ZgXItrVsy4GfQQlz6HEcgMY55t8U"
12
+ headers = {"accept" : "application/json" }
13
+ async with aiohttp .ClientSession () as session :
14
+ try :
15
+ async with session .get (url + API_KEY , headers = headers ) as response :
16
+ response .raise_for_status ()
17
+ return await response .json ()
18
+ except aiohttp .ClientError as e :
19
+ logging .error (f'Failed to fetch data { e } ' )
20
+ return
21
+
22
+ async def formate_data (data ):
23
+ if "timelines" in data and 'minutely' in data ['timelines' ]:
24
+ first_entry = data ['timelines' ]['minutely' ][0 ]
25
+ time = first_entry ['time' ]
26
+ values = first_entry ['values' ]
27
+
28
+ data_Summary = (
29
+ f"Time: { time } \n "
30
+ f"Temperature: { values ['temperature' ]} °C\n "
31
+ f"Apparent Temperature: { values ['temperatureApparent' ]} °C\n "
32
+ f"Humidity: { values ['humidity' ]} %\n "
33
+ f"Precipitation Probability: { values ['precipitationProbability' ]} %\n "
34
+ f"Rain Intensity: { values ['rainIntensity' ]} mm/hr\n "
35
+ f"Wind Speed: { values ['windSpeed' ]} m/s\n "
36
+ f"Wind Direction: { values ['windDirection' ]} °\n "
37
+ f"Visibility: { values ['visibility' ]} km\n "
38
+ f"Cloud Cover: { values ['cloudCover' ]} %\n "
39
+ )
40
+
41
+ return data_Summary
42
+ else :
43
+ return 'No weather data available at this movement.'
44
+
45
+ async def main (location , update_interval , notification_duration ):
46
+ while True :
47
+ URL = f'https://api.tomorrow.io/v4/weather/forecast?location={ location } &'
48
+ weather_data = await fetch_data (URL )
49
+ if weather_data :
50
+ weather_Summary = await formate_data (weather_data )
51
+ logging .info (f'Weather data fetched { weather_Summary } ' )
52
+ notifier .show_toast (f'Live Weather Update { location } ' , weather_Summary , duration = notification_duration )
53
+ else :
54
+ logging .error ("failed to fetch weather data" )
55
+
56
+ logging .info (f'Waiting for { update_interval } seconds before next update.' )
57
+ await asyncio .sleep (update_interval )
58
+
59
+ if __name__ == '__main__' :
60
+ parser = argparse .ArgumentParser (description = "Live Weather Notification" )
61
+ parser .add_argument ('location' , help = 'Location for weather (e.g. Pakistan, PK)' )
62
+ parser .add_argument ('-u' , '--update-interval' , type = int , default = 3600 , help = "Update interval in seconds (Default : 3600)" )
63
+ parser .add_argument ('-d' , '--notification-duration' , type = int , default = 3 , help = "Duration of Notification in seconds (Default : 3)" )
64
+
65
+ args = parser .parse_args ()
66
+
67
+ asyncio .run (main (args .location , args .update_interval , args .notification_duration ))
0 commit comments