#!/usr/bin/env python """ Python class to interface with local Tesla Powerwall JSON API https://github.com/piersdd/tesla-powerwall-json-py Example: import powerwall_site gateway_host = '192.168.5.6' password = 'ST17I0012345' backup_reserve_percent = float("5.1") tpw = powerwall_site.main('192.168.5.6', 'ST17I0012345') """ import json, time import requests from requests.exceptions import HTTPError, Timeout import urllib3 urllib3.disable_warnings() # For 'verify=False' SSL warning import logging _LOGGER = logging.getLogger(__name__) def main(): gateway_host = '192.168.1.179' password = 'ABCDEFGH' email = 'email@gmail.com' backup_reserve_percent = float("5.0") logging.basicConfig(filename='powerwall_site.log', level=logging.WARNING) ## Instantiate powerwall_site object tpw = powerwall_site(gateway_host, password) ## Query Sitemaster _sitemaster = tpw.sitemaster() ## Query battery state of charge (SoC) _stateofenergy = tpw.stateofenergy() ## Store battery SoC tpw.battery_soc = _stateofenergy['percentage'] ## Query meter aggregates _meters_aggregates = tpw.meters_aggregates() ## Store meter aggregates # Currently populated by site (grid), battery, load, and solar tpw.meters = meters(_meters_aggregates) ## Validate token (Restarts Powerwall) tpw.token = tpw.valid_token() # ## Query Operation mode # _operation = tpw.operation() # ## Set Battery to Charge (Backup) #real_mode = 'backup' #tpw.operation_set(real_mode, backup_reserve_percent) # ## Set Battery to Go back to time control (Self Consumption) real_mode = 'autonomous' tpw.operation_set(real_mode, backup_reserve_percent) ## Pause Battery (Self Consumption) tpw.operation_pause() ## Validate token (Restarts Powerwall) print ("grabbing another token") #tpw.token = 'nWc26nWM7T6ZLH6MQQUYzyabeSn56IBWfj_i0JtFUQS1A1wEZ1ZaPBkFHYFf8LxDJqrU6fgesY2TOFLKm99xnw==' tpw.token = tpw.valid_token() ## Run site master tpw.sitemaster_run() ## Some output print() print(tpw.meters.site.last_communication_time) print() print("battery_soc: " + str(tpw.battery_soc)) print() print("site.instant_power: " + str(tpw.meters.site.instant_power)) print("battery.instant_power: " + str(tpw.meters.battery.instant_power)) print("load.instant_power: " + str(tpw.meters.load.instant_power)) print("solar.instant_power: " + str(tpw.meters.solar.instant_power)) print() print("site.energy_imported: " + str(tpw.meters.site.energy_imported)) print("site.energy_exported: " + str(tpw.meters.site.energy_exported)) print("solar.energy_exported: " + str(tpw.meters.solar.energy_exported)) class powerwall_site(object): """Tesla Powerwall Sitemaster Attributes: token: for access to Powerwall gateway. running: Boolean uptime: in seconds connected_to_tesla: Boolean gateway_host: fqdn or IP address of gateway password: derivative of serial number battery_soc: percentage charge in battery backup_reserve_percent: backup event reserve limit for discharge """ def __init__(self, gateway_host, password): """Return a new Powerwall_site object.""" self.token = 'nWc26nWM7T6ZLH6MQQUYzyabeSn56IBWfj_i0JtFUQS1A1wEZ1ZaPBkFHYFf8LxDJqrU6fgesY2TOFLKm99xnw==' self.running = False # self.uptime = 0 # self.connected_to_tesla = False self.gateway_host = gateway_host self.password = password self.battery_soc = 0 self.backup_reserve_percent = 13.14159265358979 self.real_mode = '' self.base_path = 'https://' + self.gateway_host self.auth_header = {'Authorization': 'Bearer ' + self.token} ### Returns current valid token or new valid token def valid_token(self): endpoint = '/api/login/Basic' url = self.base_path + endpoint status_endpoint = '/api/status' status_url = self.base_path + status_endpoint ## Assess token validity [with /api/status] result = requests.get(status_url, headers=self.auth_header, verify=False, timeout=2) # Use the built-in JSON function to return parsed data dataobj = result.json() print ("Got token") print (dataobj) print ("Result") print (result) ## Token expired, Retrieve new token [with /api/login/Basic] if result.status_code == 401: print ("Token expired getting new one") headers = { 'Content-Type': 'application/json', } #Data here should match your criteria enter your password and email used for login on TEG data = '{"username":"installer","password":"ABCDEFGH","email":"email@gmail.com","force_sm_off":false}' print(data) result = requests.post(url, headers=headers, data=data, verify=False) print ("Got response") print(result) new_dataobj = result.json() ## Unable to retrieve new token (Unauthorised). Error with password perhaps if result.status_code == 401: # Use the built-in JSON function to return parsed data print("Unable to retreive new token") print(json.dumps(new_dataobj,indent=4)) elif result.status_code == 200: new_token = new_dataobj['token'] print ("Got new token now:") print (new_token) ## Restart Powerwall sitemaster. #print ("restarting site master") self.auth_header = {'Authorization': 'Bearer ' + new_token} #self.sitemaster_run() return new_token else: return self.token # existing invalid token ## Returns Sitemaster status def sitemaster(self): endpoint = '/api/sitemaster' url = self.base_path + endpoint try: result = requests.get(url, verify=False, timeout=5) return result.json() except requests.exceptions.RequestException: print('HTTP Request failed') ## Start the Powerwall(s) & Gateway (usually after getting an authentication token) def sitemaster_run(self): endpoint = '/api/sitemaster/run' url = self.base_path + endpoint try: result = requests.get(url, headers=self.auth_header, verify=False, timeout=5) print("## Debug sitemaster_run()") print("## result.status_code:" + str(result.status_code)) if result.status_code == 202: self.running = True except requests.exceptions.RequestException: print('HTTP Request failed') ## Reads aggregate meter information. def meters_aggregates(self): endpoint = '/api/meters/aggregates' url = self.base_path + endpoint result = requests.get(url, verify=False, timeout=5) return result.json() ## Read State of Charge (in percent) def stateofenergy(self): # When Sitemaster is not running, caught: # Error: 502 Server Error: Bad Gateway for url: https://powerwall.local/api/system_status/soe endpoint = '/api/system_status/soe' url = self.base_path + endpoint try: result = requests.get(url, verify=False, timeout=5) # raise_for_status will throw an exception if an HTTP error # code was returned as part of the response result.raise_for_status() return result.json() except HTTPError as err: print("Error: {0}".format(err)) except Timeout as err: print("Request timed out: {0}".format(err)) ## Read Powerwall Operation Mode (real_mode) def operation(self): endpoint = '/api/operation' url = self.base_path + endpoint try: result = requests.get(url, headers=self.auth_header, verify=False, timeout=5) if result.status_code == 200: self.real_mode = result.json()['real_mode'] print("## Debug valid_token()") print("self.real_mode: " + self.real_mode) return result.json() except HTTPError as err: print("Error: {0}".format(err)) except Timeout as err: print("Request timed out: {0}".format(err))# ## Pause Powerwall Operation # Discharge (self_consumption) w/ Current SoC as backup_reserve_percent def operation_pause(self): #_backup_reserve_percent = self.battery_soc #_payload = json.dumps({"real_mode": "self_consumption", "backup_reserve_percent": _backup_reserve_percent}) _set_endpoint = '/api/operation' _set_url = self.base_path + _set_endpoint _enable_endpoint = '/api/config/completed' _enable_url = self.base_path + _enable_endpoint try: print ('Committing change to server') result = requests.get(_enable_url, headers=self.auth_header, verify=False, timeout=5) print('Response HTTP Status Code: {status_code}'.format(status_code=result.status_code)) print('Response HTTP Response Body: {content}'.format(content=result.content)) print ("Setting old token again") self.token = 'nWc26nWM7T6ZLH6MQQUYzyabeSn56IBWfj_i0JtFUQS1A1wEZ1ZaPBkFHYFf8LxDJqrU6fgesY2TOFLKm99xnw==' self.auth_header = {'Authorization': 'Bearer ' + self.token} except HTTPError as err: print("Error: {0}".format(err)) except Timeout as err: print("Request timed out: {0}".format(err))# ## Set Powerwall Operation to Charge (Backup) or Discharge (self_consumption) # Pause PERHAPS WITH (self_consumption) w/ Current SoC as backup_reserve_percent def operation_set(self, real_mode, backup_reserve_percent): # auth_header = {'Authorization': 'Bearer ' + self.token} payload = json.dumps({"real_mode": real_mode, "backup_reserve_percent": backup_reserve_percent}) set_endpoint = '/api/operation' set_url = self.base_path + set_endpoint enable_endpoint = '/api/config/completed' enable_url = self.base_path + enable_endpoint print ("Setting mode: " + payload) try: result = requests.post(set_url, data=payload, headers=self.auth_header, verify=False, timeout=5) if result.status_code == 200: self.real_mode = result.json()['real_mode'] # print("## Debug valid_token()") # print("self.real_mode: " + self.real_mode) # print('Response HTTP Status Code: {status_code}'.format( # status_code=result.status_code)) # print('Response HTTP Response Body: {content}'.format( # content=result.content)) # Enable Powerwall operation (after set operation) try: result = requests.get(enable_url, headers=self.auth_header, verify=False, timeout=5) # print('Response HTTP Status Code: {status_code}'.format( # status_code=result.status_code)) # print('Response HTTP Response Body: {content}'.format( # content=result.content)) except HTTPError as err: print("Error: {0}".format(err)) except Timeout as err: print("Request timed out: {0}".format(err))# except HTTPError as err: print("Error: {0}".format(err)) except Timeout as err: print("Request timed out: {0}".format(err))# ## Meter class to store meter_aggregates JSON class meters(object): def __init__(self, d): if type(d) is str: d = json.loads(d) self.convert_json(d) def convert_json(self, d): self.__dict__ = {} for key, value in d.items(): if type(value) is dict: value = meters(value) self.__dict__[key] = value def __setitem__(self, key, value): self.__dict__[key] = value def __getitem__(self, key): return self.__dict__[key] if __name__ == "__main__": main()