-
-
Notifications
You must be signed in to change notification settings - Fork 0
/
main.py
98 lines (86 loc) · 4.03 KB
/
main.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
import os
import requests
from datetime import datetime, timedelta
from dotenv import load_dotenv
import json # Import json for saving data to files
# Ensure the output directory exists
output_dir = './output'
if not os.path.exists(output_dir):
os.makedirs(output_dir)
# Load the environment variables from the .env file
load_dotenv()
# Load environment variables
WEATHERXMUSERNAME = os.getenv('WEATHERXMUSERNAME')
WEATHERXMPASSWORD = os.getenv('WEATHERXMPASSWORD')
# Base URL for the WeatherXM API
BASE_URL = 'https://api.weatherxm.com'
def login():
"""Authenticate with the WeatherXM API and return the access token."""
url = f'{BASE_URL}/api/v1/auth/login'
credentials = {
"username": WEATHERXMUSERNAME,
"password": WEATHERXMPASSWORD
}
response = requests.post(url, json=credentials)
response.raise_for_status()
return response.json()['token']
def get_devices(token):
"""Retrieve and return the list of devices."""
url = f'{BASE_URL}/api/v1/me/devices'
headers = {'Authorization': f'Bearer {token}'}
response = requests.get(url, headers=headers)
response.raise_for_status()
return response.json()
def get_device_weather_history(token, device_id, from_date, to_date):
"""Retrieve weather data for a specific device."""
url = f'{BASE_URL}/api/v1/me/devices/{device_id}/history'
headers = {'Authorization': f'Bearer {token}'}
params = {
'fromDate': from_date,
'toDate': to_date
}
response = requests.get(url, headers=headers, params=params)
response.raise_for_status()
return response.json()
def save_weather_data(data, date):
"""Save weather data to a JSON file."""
file_path = os.path.join(output_dir, f'{date}.json')
with open(file_path, 'w') as file:
json.dump(data, file, indent=4)
print(f"Weather data saved to {file_path}")
if __name__ == '__main__':
try:
token = login()
print("Authentication successful.")
devices = get_devices(token)
if devices:
for device in devices:
device_id = device['id']
print(f"\nFetching weather history for device ID: {device_id}")
from_date = (datetime.now() - timedelta(days=1)).strftime('%Y-%m-%d')
to_date = datetime.now().strftime('%Y-%m-%d')
weather_data = get_device_weather_history(token, device_id, from_date, to_date)
save_weather_data(weather_data, to_date) # Save data to file
# Print weather data in human-readable format
for day_data in weather_data:
print(f"\nDate: {day_data['date']}, Time Zone: {day_data['tz']}")
for hourly_data in day_data.get('hourly', []):
print(f"Timestamp: {hourly_data['timestamp']}")
print(f" Temperature: {hourly_data['temperature']}°C")
print(f" Humidity: {hourly_data['humidity']}%")
print(f" Feels Like: {hourly_data['feels_like']}°C")
print(f" Wind Speed: {hourly_data['wind_speed']} km/h")
print(f" Wind Direction: {hourly_data['wind_direction']} km/h")
print(f" Wind Gust: {hourly_data['wind_gust']}")
print(f" Precipitation: {hourly_data['precipitation']} mm")
print(f" Precipitation Accumulated: {hourly_data['precipitation_accumulated']} mm")
print(f" Pressure: {hourly_data['pressure']} hPa")
print(f" Dew Point: {hourly_data['dew_point']}°C")
print(f" UV Index: {hourly_data['uv_index']}")
print(f" Illuminance: {hourly_data['illuminance']}")
print(f" Solar Irradiance: {hourly_data['solar_irradiance']}")
print(" ---")
else:
print('No devices found.')
except requests.RequestException as e:
print(f'An error occurred: {e}')