powerwallBackup.py

#!/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():
    #Change these items
    gateway_host = '192.168.1.179'
    password = 'ABCDEFGH'
    email = '[email protected]'
    backup_reserve_percent = float("100.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")
            #result = requests.post(url, data=payload, verify=False, timeout=5)
            #new_dataobj = result.json()
            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":"ABCDEFG","email":"[email protected]","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()

powerwallTimeOfUse.py

#!/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 protected]'
    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 protected]","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()