sascha/badnest
1
0
Fork 0
mirror of https://github.com/USA-RedDragon/badnest.git synced 2025-01-18 18:20:44 +00:00
Code Issues Releases Wiki Activity

Add preliminary error handling support

Browse Source
This commit is contained in:
Jacob McSwain 2019-10-22 13:34:00 -05:00
parent c4f702b859
commit 98daabc5d7
No known key found for this signature in database
GPG Key ID: 59957AD09B94EDF5
1 changed files with 85 additions and 60 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

145
custom_components/badnest/api.py
View File

@ -25,10 +25,18 @@ class NestAPI:
self._session = requests.Session() self._session = requests.Session()
self._session.headers.update({"Referer": "https://home.nest.com/"}) self._session.headers.update({"Referer": "https://home.nest.com/"})
self._device_id = device_id self._device_id = device_id
if not email and not password: self._email = email
self._login_google(issue_token, cookie, api_key) self._password = password
self._issue_token = issue_token
self._cookie = cookie
self._api_key = api_key
self.login()
def login(self):
if not self._email and not self._password:
self._login_google(self._issue_token, self._cookie, self._api_key)
else: else:
self._login_nest(email, password) self._login_nest(self._email, self._password)
def _login_nest(self, email, password): def _login_nest(self, email, password):
r = self._session.post( r = self._session.post(
@ -100,21 +108,27 @@ class NestThermostatAPI(NestAPI):
self.update() self.update()
def get_devices(self): def get_devices(self):
r = self._session.post( try:
f"{API_URL}/api/0.1/user/{self._user_id}/app_launch", r = self._session.post(
json={ f"{API_URL}/api/0.1/user/{self._user_id}/app_launch",
"known_bucket_types": ["buckets"], json={
"known_bucket_versions": [], "known_bucket_types": ["buckets"],
}, "known_bucket_versions": [],
headers={"Authorization": f"Basic {self._access_token}"}, },
) headers={"Authorization": f"Basic {self._access_token}"},
devices = [] )
buckets = r.json()['updated_buckets'][0]['value']['buckets'] devices = []
for bucket in buckets: buckets = r.json()['updated_buckets'][0]['value']['buckets']
if bucket.startswith('device.'): for bucket in buckets:
devices.append(bucket.replace('device.', '')) if bucket.startswith('device.'):
devices.append(bucket.replace('device.', ''))
return devices return devices
except requests.exceptions.RequestException as e:
_LOGGER.error(e)
_LOGGER.error('Failed to get devices, trying to log in again')
self.login()
return self.get_devices()
def get_action(self): def get_action(self):
if self._hvac_ac_state: if self._hvac_ac_state:
@ -125,51 +139,62 @@ class NestThermostatAPI(NestAPI):
return "off" return "off"
def update(self): def update(self):
r = self._session.post( try:
f"{API_URL}/api/0.1/user/{self._user_id}/app_launch", r = self._session.post(
json={ f"{API_URL}/api/0.1/user/{self._user_id}/app_launch",
"known_bucket_types": ["shared", "device"], json={
"known_bucket_versions": [], "known_bucket_types": ["shared", "device"],
}, "known_bucket_versions": [],
headers={"Authorization": f"Basic {self._access_token}"}, },
) headers={"Authorization": f"Basic {self._access_token}"},
)
self._czfe_url = r.json()["service_urls"]["urls"]["czfe_url"] self._czfe_url = r.json()["service_urls"]["urls"]["czfe_url"]
temp_mode = None temp_mode = None
for bucket in r.json()["updated_buckets"]: for bucket in r.json()["updated_buckets"]:
if bucket["object_key"].startswith(f"shared.{self._device_id}"): if bucket["object_key"] \
thermostat_data = bucket["value"] .startswith(f"shared.{self._device_id}"):
self.current_temperature = \ thermostat_data = bucket["value"]
thermostat_data["current_temperature"] self.current_temperature = \
self.target_temperature = thermostat_data["target_temperature"] thermostat_data["current_temperature"]
self._compressor_lockout_enabled = thermostat_data[ self.target_temperature = \
"compressor_lockout_enabled" thermostat_data["target_temperature"]
] self._compressor_lockout_enabled = thermostat_data[
self._compressor_lockout_time = thermostat_data[ "compressor_lockout_enabled"
"compressor_lockout_timeout" ]
] self._compressor_lockout_time = thermostat_data[
self._hvac_ac_state = thermostat_data["hvac_ac_state"] "compressor_lockout_timeout"
self._hvac_heater_state = thermostat_data["hvac_heater_state"] ]
temp_mode = thermostat_data["target_temperature_type"] self._hvac_ac_state = thermostat_data["hvac_ac_state"]
self.target_temperature_high = thermostat_data[ self._hvac_heater_state = \
"target_temperature_high" thermostat_data["hvac_heater_state"]
] temp_mode = thermostat_data["target_temperature_type"]
self.target_temperature_low = \ self.target_temperature_high = thermostat_data[
thermostat_data["target_temperature_low"] "target_temperature_high"
self.can_heat = thermostat_data["can_heat"] ]
self.can_cool = thermostat_data["can_cool"] self.target_temperature_low = \
elif bucket["object_key"].startswith(f"device.{self._device_id}"): thermostat_data["target_temperature_low"]
thermostat_data = bucket["value"] self.can_heat = thermostat_data["can_heat"]
self._time_to_target = thermostat_data["time_to_target"] self.can_cool = thermostat_data["can_cool"]
self._fan_timer_timeout = thermostat_data["fan_timer_timeout"] elif bucket["object_key"] \
self.has_fan = thermostat_data["has_fan"] .startswith(f"device.{self._device_id}"):
self.fan = thermostat_data["fan_timer_timeout"] > 0 thermostat_data = bucket["value"]
self.current_humidity = thermostat_data["current_humidity"] self._time_to_target = thermostat_data["time_to_target"]
if thermostat_data["eco"]["mode"] == 'manual-eco' or \ self._fan_timer_timeout = \
thermostat_data["eco"]["mode"] == 'auto-eco': thermostat_data["fan_timer_timeout"]
temp_mode = 'eco' self.has_fan = thermostat_data["has_fan"]
self.mode = temp_mode self.fan = thermostat_data["fan_timer_timeout"] > 0
self.current_humidity = thermostat_data["current_humidity"]
if thermostat_data["eco"]["mode"] == 'manual-eco' or \
thermostat_data["eco"]["mode"] == 'auto-eco':
temp_mode = 'eco'
self.mode = temp_mode
except requests.exceptions.RequestException as e:
_LOGGER.error(e)
_LOGGER.error('Failed to update, trying to log in again')
self.login()
self.update()
def set_temp(self, temp, temp_high=None): def set_temp(self, temp, temp_high=None):
if temp_high is None: if temp_high is None: