import json import ssl from datetime import timezone from time import time import requests from requests.exceptions import ReadTimeout from websocket import create_connection from websocket._exceptions import WebSocketTimeoutException from .sense_api import * from .sense_exceptions import * class Senseable(SenseableBase): def __init__( self, username=None, password=None, api_timeout=API_TIMEOUT, wss_timeout=WSS_TIMEOUT, ssl_verify=True, ssl_cafile="", device_id=None, ): """Init the Senseable object.""" # Create session self.s = requests.session() self.set_ssl_context(ssl_verify, ssl_cafile) SenseableBase.__init__( self, username=username, password=password, api_timeout=api_timeout, wss_timeout=wss_timeout, ssl_verify=ssl_verify, ssl_cafile=ssl_cafile, device_id=device_id, ) def set_ssl_context(self, ssl_verify, ssl_cafile): """Create or set the SSL context. Use custom ssl verification, if specified.""" if not ssl_verify: self.s.verify = False elif ssl_cafile: self.s.verify = ssl_cafile def authenticate(self, username, password, ssl_verify=True, ssl_cafile=""): """Authenticate with username (email) and password. Optionally set SSL context as well. This or `load_auth` must be called once at the start of the session.""" auth_data = {"email": username, "password": password} # Get auth token try: resp = self.s.post(API_URL + "authenticate", auth_data, headers=self.headers, timeout=self.api_timeout) except Exception as e: raise Exception(f"Connection failure: {e}") # check MFA code required if resp.status_code == 401: data = resp.json() if "mfa_token" in data: self._mfa_token = data["mfa_token"] raise SenseMFARequiredException(data["error_reason"]) # check for 200 return if resp.status_code != 200: raise SenseAuthenticationException( f"Please check username and password. API Return Code: {resp.status_code}" ) data = resp.json() self._set_auth_data(data) self.set_monitor_id(data["monitors"][0]["id"]) def validate_mfa(self, code): """Validate a multi-factor authentication code after authenticate raised SenseMFARequiredException. Authentication process is completed if code is valid.""" mfa_data = { "totp": code, "mfa_token": self._mfa_token, "client_time:": datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"), } # Get auth token try: resp = self.s.post(API_URL + "authenticate/mfa", mfa_data, headers=self.headers, timeout=self.api_timeout) except Exception as e: raise Exception(f"Connection failure: {e}") # check for 200 return if resp.status_code != 200: raise SenseAuthenticationException( f"Please check username and password. API Return Code: {resp.status_code}" ) data = resp.json() self._set_auth_data(data) self.set_monitor_id(data["monitors"][0]["id"]) def renew_auth(self): renew_data = { "user_id": self.sense_user_id, "refresh_token": self.refresh_token, } # Get auth token try: resp = self.s.post(API_URL + "renew", renew_data, headers=self.headers, timeout=self.api_timeout) except Exception as e: raise Exception(f"Connection failure: {e}") # check for 200 return if resp.status_code != 200: raise SenseAuthenticationException( f"Please check username and password. API Return Code: {resp.status_code}" ) self._set_auth_data(resp.json()) def logout(self): try: resp = self.s.get(API_URL + "logout", timeout=self.api_timeout) except Exception as e: raise Exception(f"Connection failure: {e}") # check for 200 return if resp.status_code != 200: raise SenseAPIException(f"API Return Code: {resp.status_code}") def update_realtime(self, retry=True): """Update the realtime data (device status and current power).""" # rate limit API calls now = time() if self._realtime and self.rate_limit and self.last_realtime_call + self.rate_limit > now: return self._realtime self.last_realtime_call = now try: if self.supports_realtime_update_api(): data = self.get_realtime_update() self._set_realtime(data) else: next(self.get_realtime_stream()) except SenseAuthenticationException as e: if retry: self.renew_auth() self.update_realtime(False) else: raise e def get_realtime_stream(self): """Reads realtime data from websocket. Realtime data variable is set and data is returned through generator. Continues until loop broken.""" ws = 0 url = WS_URL % (self.sense_monitor_id, self.sense_access_token) try: ws = create_connection(url, timeout=self.wss_timeout, sslopt={"cert_reqs": ssl.CERT_NONE}) while True: # hello, features, [updates,] data result = json.loads(ws.recv()) if result.get("type") == "realtime_update": data = result["payload"] self._set_realtime(data) yield data elif result.get("type") == "error": data = result["payload"] if not data["authorized"]: raise SenseAuthenticationException("Web Socket Unauthorized") raise SenseWebsocketException(data["error_reason"]) except WebSocketTimeoutException: raise SenseAPITimeoutException("API websocket timed out") finally: if ws: ws.close() def _api_call(self, url, payload={}, retry=False): """Make a call to the Sense API directly and return the json results.""" try: resp = self.s.get( API_URL + url, headers=self.headers, timeout=self.api_timeout, params=payload, ) if not retry and resp.status_code == 401: self.renew_auth() return self._api_call(url, payload, True) # 4xx represents unauthenticated if resp.status_code == 401 or resp.status_code == 403 or resp.status_code == 404: raise SenseAuthenticationException(f"API Return Code: {resp.status_code}") return resp.json() except ReadTimeout: raise SenseAPITimeoutException("API call timed out") def get_trend_data(self, scale: Scale, dt=None): """Update trend data for specified scale from API. Optionally set a date to fetch data from.""" if not dt: dt = datetime.now(timezone.utc) # Ensure monitor data is available to check solar configuration if not self._monitor: self.get_monitor_data() usage_data = self._api_call( f"app/monitors/{self.sense_monitor_id}/history/usage" + f"?scale={scale.name}&start={dt.strftime('%Y-%m-%dT%H:%M:%S')}" ) # Get solar data if solar is configured solar_data = None if self._monitor.get("solar_configured"): solar_data = self._api_call( f"app/monitors/{self.sense_monitor_id}/history/usage/solar" + f"?scale={scale.name}&start={dt.strftime('%Y-%m-%dT%H:%M:%S')}" ) # Transform to legacy format for backward compatibility self._trend_data[scale] = self._transform_usage_response(usage_data, solar_data) self._update_device_trends(scale) def update_trend_data(self, dt=None): """Update trend data of all scales from API. Optionally set a date to fetch data from.""" for scale in Scale: self.get_trend_data(scale, dt) def get_monitor_data(self): """Get monitor overview info from API.""" json = self._api_call(f"app/monitors/{self.sense_monitor_id}/overview") if "monitor_overview" in json and "monitor" in json["monitor_overview"]: self._monitor = json["monitor_overview"]["monitor"] return self._monitor def get_discovered_device_names(self): """Get list of device names from API.""" json = self._api_call(f"app/monitors/{self.sense_monitor_id}/devices") self._devices = [entry["name"] for entry in json] return self._devices def get_discovered_device_data(self): """Get list of raw device data from API.""" return self._api_call(f"monitors/{self.sense_monitor_id}/devices") def always_on_info(self): """Always on info from API - pretty generic similar to the web page.""" return self._api_call(f"app/monitors/{self.sense_monitor_id}/devices/always_on") def get_monitor_info(self): """View info on monitor & device detection status from API.""" return self._api_call(f"app/monitors/{self.sense_monitor_id}/status") def get_device_info(self, device_id): """Get specific informaton about a device from API.""" return self._api_call(f"app/monitors/{self.sense_monitor_id}/devices/{device_id}") def get_all_usage_data(self, payload={"n_items": 30}): """Gets usage data by device from API Args: payload (dict, optional): known params are: n_items: the number of items to return device_id: limit results to a specific device_id prior_to_item:. date in format YYYY-MM-DDTHH:MM:SS.mmmZ rollup: ? Defaults to {'n_items': 30}. Returns: dict: usage data """ # lots of info in here to be parsed out return self._api_call(f"users/{self.sense_user_id}/timeline", payload) def get_realtime_update(self): """Get a realtime update for a device from API.""" return self._api_call(f"app/{self.sense_monitor_id}/realtime_update") def get_sw_version(self): """Get the currently installed software version on a device from API.""" now = time() if not self._sw_version or now > self.last_sw_version_check + SW_VERSION_CHECK_INTERVAL: data = self.get_monitor_info() self._sw_version = data["monitor_info"]["version"] self.last_sw_version_check = now return self._sw_version def supports_realtime_update_api(self) -> bool: """Returns whether a device supports the /app//realtime_update API.""" # Assume that software version will not downgrade; if set, don't check again if self._sw_version >= MIN_VERSION_REALTIME_UPDATE_API: return True else: self.get_sw_version() return self._sw_version >= MIN_VERSION_REALTIME_UPDATE_API