1+ import urllib .parse
2+ from typing import Dict , Union
3+
4+ import requests
5+ from requests import RequestException
6+
7+ from ._errors import RequestError
8+ from ._forecast import (
9+ HomeAssistantForecast , HomeAssistantCurrentForecast , HomeAssistantHourlyForecast , HomeAssistantDailyForecast
10+ )
11+ from ._sun import HomeAssistantSunInfo , HomeAssistantSunState
12+
13+
14+ class HomeAssistantAdapter :
15+ @staticmethod
16+ def __make_headers_from_token (token : str ) -> Dict [str , str ]:
17+ return {"Authorization" : f"Bearer { token } , "Content-Type" : "application/json" }
18+
19+ @staticmethod
20+ def __request (url : str , token : str , post : bool = False ,
21+ data : Union [Dict [str , str ], None ] = None , check_ssl = True ) -> requests .Response :
22+ try :
23+ if post :
24+ r = requests .post (
25+ url = url , headers = HomeAssistantAdapter .__make_headers_from_token (token = token ), json = data ,
26+ params = {"return_response" : True }, verify = check_ssl
27+ )
28+ else :
29+ r = requests .get (
30+ url = url , headers = HomeAssistantAdapter .__make_headers_from_token (token = token ), params = data , verify = check_ssl
31+ )
32+ except RequestException :
33+ raise RequestError (error_code = - 1 , url = url , method = "POST" if post else "GET" , body = "" )
34+ if not r .ok :
35+ raise RequestError (error_code = r .status_code , url = url , method = "POST" if post else "GET" , body = r .text )
36+ return r
37+
38+ @staticmethod
39+ def filter_attributes (attributes_received ,forecast_type = 'current' ):
40+ output_attributes = {}
41+ allowed_keys = [
42+ 'temperature' ,
43+ 'dew_point' ,
44+ 'temperature_unit' ,
45+ 'humidity' ,
46+ 'uv_index' ,
47+ 'pressure' ,
48+ 'pressure_unit' ,
49+ 'wind_bearing' ,
50+ 'wind_speed' ,
51+ 'wind_speed_unit' ,
52+ 'precipitation_unit' ,
53+ 'supported_features' ,
54+ 'visibility_unit' ,
55+ 'attribution' ,
56+ 'friendly_name' ,
57+ 'cloud_coverage'
58+ ]
59+ if forecast_type == 'hourly' :
60+ allowed_keys = [
61+ 'temperature' ,
62+ 'humidity' ,
63+ 'uv_index' ,
64+ 'wind_bearing' ,
65+ 'wind_speed' ,
66+ 'cloud_coverage' ,
67+ 'condition' ,
68+ 'datetime' ,
69+ 'precipitation' ,
70+ ]
71+ elif forecast_type == 'daily' :
72+ allowed_keys = [
73+ 'temperature' ,
74+ 'humidity' ,
75+ 'uv_index' ,
76+ 'wind_bearing' ,
77+ 'wind_speed' ,
78+ 'condition' ,
79+ 'datetime' ,
80+ 'precipitation' ,
81+ 'templow' ,
82+ ]
83+ for key in attributes_received :
84+ if key in allowed_keys :
85+ output_attributes [key ] = attributes_received [key ]
86+ for key in allowed_keys :
87+ if not key in output_attributes :
88+ output_attributes [key ] = None
89+ return output_attributes
90+
91+ @staticmethod
92+ def get_forecast (server_url : str , entity_id : str , token : str , check_ssl : bool ) -> HomeAssistantForecast :
93+ # Based on Home Assistant's WeatherEntityFeature IntFlag
94+ FORECAST_DAILY = 1
95+ FORECAST_HOURLY = 2
96+
97+ current_url = urllib .parse .urljoin (base = server_url , url = f"/api/states/{ entity_id } )
98+ forecast_url = urllib .parse .urljoin (base = server_url , url = "/api/services/weather/get_forecasts" )
99+
100+ current = HomeAssistantAdapter .__request (url = current_url , token = token , check_ssl = check_ssl )
101+ current_json = current .json ()
102+ current_forecast_attributes = HomeAssistantAdapter .filter_attributes (current_json ["attributes" ], 'current' )
103+ current_forecast_attributes ['condition' ] = current_json ['state' ]
104+ supported_features = current_json .get ("attributes" , {}).get ("supported_features" , 0 )
105+ if supported_features is None :
106+ supported_features = 0
107+
108+ hourly_forecast_attributes = []
109+ if supported_features & FORECAST_HOURLY :
110+ try :
111+ hourly = HomeAssistantAdapter .__request (
112+ url = forecast_url , token = token , post = True , data = {"entity_id" : entity_id , "type" : "hourly" }, check_ssl = check_ssl
113+ )
114+ hourly_forecast_attributes = [HomeAssistantAdapter .filter_attributes (hourly_forecast , 'hourly' ) for hourly_forecast in hourly .json ()["service_response" ][entity_id ]["forecast" ]]
115+ except RequestError :
116+ hourly_forecast_attributes = []
117+
118+ daily_forecast_attributes = []
119+ if supported_features & FORECAST_DAILY :
120+ try :
121+ daily = HomeAssistantAdapter .__request (
122+ url = forecast_url , token = token , post = True , data = {"entity_id" : entity_id , "type" : "daily" }, check_ssl = check_ssl
123+ )
124+ daily_forecast_attributes = [HomeAssistantAdapter .filter_attributes (daily_forecast , 'daily' ) for daily_forecast in daily .json ()["service_response" ][entity_id ]["forecast" ]]
125+ except RequestError :
126+ daily_forecast_attributes = []
127+
128+ return HomeAssistantForecast (
129+ current = HomeAssistantCurrentForecast (** current_forecast_attributes ),
130+ hourly = [
131+ HomeAssistantHourlyForecast (** hourly_forecast )
132+ for hourly_forecast in hourly_forecast_attributes
133+ ],
134+ daily = [
135+ HomeAssistantDailyForecast (** daily_forecast )
136+ for daily_forecast in daily_forecast_attributes
137+ ],
138+ )
139+
140+ @staticmethod
141+ def get_sun_info (server_url : str , entity_id : str , token : str , check_ssl : bool ) -> HomeAssistantSunInfo :
142+ sun_url = urllib .parse .urljoin (base = server_url , url = f"/api/states/{ entity_id } )
143+ sun = HomeAssistantAdapter .__request (url = sun_url , token = token , check_ssl = check_ssl )
144+ sun_data = sun .json ()
145+ return HomeAssistantSunInfo (** sun_data ["attributes" ], state = HomeAssistantSunState (sun_data ["state" ]))
0 commit comments