Add support for Tesla Powerwall

@james_hiscott from the server the best bet is to use the phone app, you can also try https://www.teslaapi.io/powerwalls/state-and-settings.
Sorry, I dont change the percentage, so if there is some error there I am not aware. You can also try changing the format, maybe is expecting a float number (e.g. 0.05 for 5%) or it could be expecting a string with a percentage. Since is so hard to sniff the communication, we are pretty much guessing here what to send.

@Hodor glad you were able to figure it out! thanks for leaving the update.

I followed all the instructions for the custom_components directory, the component is being found (along with the usual custom component warning) but I am getting the same setup error:

2020-09-24 18:47:47 ERROR (MainThread) [homeassistant.setup] Setup failed for tesla_gateway: No setup function defined.

I verified the async_setup is present in the init.py code and I am running a very recent (0.115.1) version of HA.

@Matthew_Budraitis or anyone else, how did you go get past this?

Ok, the fix was renaming init.py to __init__.py. Hopefully that will help the next person :slight_smile:

More learnings: I can’t get the reserve percent to change, similar to @james_hiscott saw above. I am able to switch modes from self powered to backup-only and back, so things are basically wired up correctly. It could be a formatting issue as @estebanp mentioned on Jun 28.

I note that there is an implementation over at the SmartThings side of the world for controlling Powerwalls via the Tesla server APIs, but it uses the /api/1/energy_sites/:energySiteId/operation API only to set operation mode, and uses a separate /api/1/energy_sites/:energySiteId/backup API to set the backup reserve percent.

I’m going to muck around with @estebanp’s the code a bit to see if I can use the dedicated API call to make it work.

Ok I got this to work too, except the battery percentage piece but not too concerned with that.

I want to switch my PW to TOU at 10p and back to self at 6a during the summer when the AC is on. I set up two automations to do so, but when I hit the slider to turn them off, the slider turns back on within a second. What do?

The following two files are updated to support both operation mode changes and battery reserve percent changes.

All the other files are identical to the excellent post of @estebanp from above.

The first, services.yaml, needs to be put into /custom_components/tesla_gateway:

set_operation:
  description: >
    Changes operation mode
  fields:
    real_mode:
      description: Mode to set to the Tesla gateway.
      example: 'self_consumption, backup'
set_reserve:
  description: >
    Changes battery reserve percent in self_consumption mode
  fields:
    reserve_percent:
      description: Percentage of battery reserve
      example: 70

The second, __init__.py needs to go in the same directory:

"""
Monitors and controls the Tesla gateway.
"""
import logging

import aiohttp
import asyncio
import async_timeout
import json
import time
import voluptuous as vol

from homeassistant.const import (
    CONF_USERNAME,
    CONF_PASSWORD
    )
from homeassistant.helpers.aiohttp_client import async_get_clientsession
import homeassistant.helpers.config_validation as cv

DOMAIN = 'tesla_gateway'

_LOGGER = logging.getLogger(__name__)

DEFAULT_TIMEOUT = 100

CONFIG_SCHEMA = vol.Schema({
    DOMAIN: vol.Schema({
        vol.Required(CONF_USERNAME): cv.string,
        vol.Required(CONF_PASSWORD): cv.string,
    }),
}, extra=vol.ALLOW_EXTRA)

tesla_base_url = 'https://owner-api.teslamotors.com'

@asyncio.coroutine
def async_setup(hass, config):

    # Tesla gateway is SSL but has no valid certificates
    websession = async_get_clientsession(hass, verify_ssl=False)

    domain_config = config[DOMAIN]
    conf_user = domain_config[CONF_USERNAME]
    conf_password = domain_config[CONF_PASSWORD]
    access_token = None

    @asyncio.coroutine
    def login():

        login_url = tesla_base_url + '/oauth/token'
        headers = {'Content-type': 'application/json'}
        body = {
            'email': conf_user,
            'password': conf_password,
            'client_secret': 'c7257eb71a564034f9419ee651c7d0e5f7aa6bfbd18bafb5c5c033b093bb2fa3',
            'client_id': '81527cff06843c8634fdc09e8ac0abefb46ac849f38fe1e431c2ef2106796384',
            'grant_type': 'password'
        }

        try:
            with async_timeout.timeout(DEFAULT_TIMEOUT, loop=hass.loop):
                response = yield from websession.post(login_url,
                    headers=headers,
                    json=body,
                    raise_for_status=False)

            if response.status != 200:
                returned_text = yield from response.text()
                _LOGGER.warning('Error %d on call %s:\n%s', response.status, response.url, returned_text)
            else:
                returned_json = yield from response.json()
                _LOGGER.debug(returned_json)
                access_token = returned_json['access_token']
                return access_token

        except asyncio.TimeoutError:
            _LOGGER.warning('Timeout call %s.', response.url)

        except aiohttp.ClientError:
            _LOGGER.error('Client error %s.', response.url)

        return None

    @asyncio.coroutine
    def revoke(access_token):

        revoke_url = tesla_base_url + '/oauth/revoke'
        headers = {'Content-type': 'application/json'}
        body = {
            'token': access_token
        }

        try:
            with async_timeout.timeout(DEFAULT_TIMEOUT, loop=hass.loop):
                response = yield from websession.post(revoke_url,
                    headers=headers,
                    json=body,
                    raise_for_status=False)

            if response.status != 200:
                returned_text = yield from response.text()
                _LOGGER.warning('Error %d on call %s:\n%s', response.status, response.url, returned_text)
            else:
                _LOGGER.debug('revoke completed')
                return True

        except asyncio.TimeoutError:
            _LOGGER.warning('Timeout call %s.', response.url)

        except aiohttp.ClientError:
            _LOGGER.error('Client error %s.', response.url)

        return False

    @asyncio.coroutine
    def get_energy_site_id(access_token):

        list_url = tesla_base_url + '/api/1/products'
        headers = {
            'Authorization': 'Bearer ' + access_token
            }
        body = {}

        try:
            with async_timeout.timeout(DEFAULT_TIMEOUT, loop=hass.loop):
                response = yield from websession.get(list_url,
                    headers=headers,
                    json=body,
                    raise_for_status=False)

            if response.status != 200:
                returned_text = yield from response.text()
                _LOGGER.warning('Error %d on call %s:\n%s', response.status, response.url, returned_text)
            else:
                returned_json = yield from response.json()
                _LOGGER.debug(returned_json)
                for r in returned_json['response']:
                    if 'energy_site_id' in r:
                        return r['energy_site_id']
                return None

        except asyncio.TimeoutError:
            _LOGGER.warning('Timeout call %s.', response.url)

        except aiohttp.ClientError:
            _LOGGER.error('Client error %s.', response.url)

        return None

    @asyncio.coroutine
    def set_operation(access_token,energy_site_id,service_data):

        operation_url = tesla_base_url + '/api/1/energy_sites/{}/operation'.format(energy_site_id)
        headers = {
            'Content-type': 'application/json',
            'Authorization': 'Bearer ' + access_token
            }
        body = {
            'default_real_mode': service_data['real_mode']
            # 'backup_reserve_percent':int(service_data['backup_reserve_percent'])
            }
        _LOGGER.debug(body)

        try:
            with async_timeout.timeout(DEFAULT_TIMEOUT, loop=hass.loop):
                response = yield from websession.post(operation_url,
                    json=body,
                    headers=headers,
                    raise_for_status=False)

            if response.status != 200:
                returned_text = yield from response.text()
                _LOGGER.warning('Error %d on call %s:\n%s', response.status, response.url, returned_text)
            else:
                returned_json = yield from response.json()
                _LOGGER.debug('set operation successful, response: %s', returned_json)

        except asyncio.TimeoutError:
            _LOGGER.warning('Timeout call %s.', response.url)

        except aiohttp.ClientError:
            _LOGGER.error('Client error %s.', response.url)

    @asyncio.coroutine
    def async_set_operation(service):

        access_token = yield from login()
        if access_token:
            energy_site_id = yield from get_energy_site_id(access_token)
            if energy_site_id:
                yield from set_operation(access_token, energy_site_id, service.data)
            yield from revoke(access_token)

    hass.services.async_register(DOMAIN, 'set_operation', async_set_operation)

    @asyncio.coroutine
    def set_reserve(access_token,energy_site_id,service_data):

        operation_url = tesla_base_url + '/api/1/energy_sites/{}/backup'.format(energy_site_id)
        headers = {
            'Content-type': 'application/json',
            'Authorization': 'Bearer ' + access_token
            }
        body = {
            'backup_reserve_percent': int(service_data['reserve_percent'])
            }
        _LOGGER.debug(body)

        try:
            with async_timeout.timeout(DEFAULT_TIMEOUT, loop=hass.loop):
                response = yield from websession.post(operation_url,
                    json=body,
                    headers=headers,
                    raise_for_status=False)

            if response.status != 200:
                returned_text = yield from response.text()
                _LOGGER.warning('Error %d on call %s:\n%s', response.status, response.url, returned_text)
            else:
                returned_json = yield from response.json()
                _LOGGER.debug('set reserve successful, response: %s', returned_json)

        except asyncio.TimeoutError:
            _LOGGER.warning('Timeout call %s.', response.url)

        except aiohttp.ClientError:
            _LOGGER.error('Client error %s.', response.url)

    @asyncio.coroutine
    def async_set_reserve(service):

        access_token = yield from login()
        if access_token:
            energy_site_id = yield from get_energy_site_id(access_token)
            if energy_site_id:
                yield from set_reserve(access_token, energy_site_id, service.data)
            yield from revoke(access_token)

    hass.services.async_register(DOMAIN, 'set_operation', async_set_operation)
    hass.services.async_register(DOMAIN, 'set_reserve', async_set_reserve)

    return True

That’s it! One thing to note is that if you use these and the Tesla app doesn’t show any changes, try killing and restarting the app. The Tesla app caches old values and so even if the changes happen you won’t see them unless you exit and restart the app.

2 Likes

This is awesome and has helped me make some really good automations for the very dynamic time of use tariff we’re on here. However one thing that would be really good is if the powerwall could be queried for it’s current mode, backup percentage and remaining percentage using this same custom component. The latter is available direct from the gateway and I use that at the moment but it can be up to 5% out from what the app and web API report as the app displays remaining percentage hiding a buffer, but the gateway does not. This currently causes a response lag and sometimes small amounts of unwanted charging when I set the backup reserve to the current remaining value in order to pause discharge during cheap periods.

I noticed some folks said they could not find energy site ID, this is actually embedded in to your product list , so you have to get products first, filter for “battery” then pull out the attribute.

Some helpful code we wrote to assist:

This repo has the code we use to automate powerwall on rasberry pi.

Is it just me, or did the Powerwall firmware update 1.50.2 totally disable local API access?

Mine is on 1.50.1 and the official powerwall integration is still giving me live energy use stats. I don’t use local control though, preferring the app API calls in the custom integration above so i can’t say if local control is now completely dead.

Can you still get to the local page to view the status?

I am on 1.50.1 still so can’t confirm yet, but no doubt it will be alone soon.

I can access fine through the TEG-xxx SSID at 192.168.91.1, but not from my home LAN. It gets an IP address from my home DHCP server and Tx/Rx to the cloud is working. Tesla app works. But the controller seems to be ignoring home LAN subnet traffic, even pings.

Does this exists as code checked in somewhere?
I think @fnord123 's code from Sept 27 is what I want, but I’m not sure what other files you’re talking about.
I’m trying to setup automation to maximize PW usage using Solcast forecast data.

Something seems to have changed at the Tesla end of the API.
I understand they are widening the use of 2FA.
This code seems to fail now and the error behind the scenes is…

2021-01-30 14:02:57 WARNING (MainThread) [custom_components.tesla_gateway] Error 400 on call https://owner-api.teslamotors.com/oauth/token:
{"response":"endpoint_deprecated:_please_update_your_app."}

I’m hoping someone better than me can re-write this code to use the new authentication methods.
But in the meantime, I’ve fudged it by passing in my Tesla Auth instead of my password.
I then commented out all the login code and just have that return the token that I supplied.
I’ll need to manually refresh the token every 45 days.
I tried to use a refresh token to do the login and get a token, but this seems to be broken too.

My code now looks like this…

"""
Monitors and controls the Tesla gateway.
"""
import logging

import aiohttp
import asyncio
import async_timeout
import json
import time
import voluptuous as vol

from homeassistant.const import (
    CONF_USERNAME,
    CONF_PASSWORD
    )
from homeassistant.helpers.aiohttp_client import async_get_clientsession
import homeassistant.helpers.config_validation as cv

DOMAIN = 'tesla_gateway'

_LOGGER = logging.getLogger(__name__)

DEFAULT_TIMEOUT = 100

CONFIG_SCHEMA = vol.Schema({
    DOMAIN: vol.Schema({
        vol.Required(CONF_USERNAME): cv.string,
        vol.Required(CONF_PASSWORD): cv.string,
    }),
}, extra=vol.ALLOW_EXTRA)

tesla_base_url = 'https://owner-api.teslamotors.com'

@asyncio.coroutine
def async_setup(hass, config):

    # Tesla gateway is SSL but has no valid certificates
    websession = async_get_clientsession(hass, verify_ssl=False)

    domain_config = config[DOMAIN]
    conf_user = domain_config[CONF_USERNAME]
    conf_password = domain_config[CONF_PASSWORD]  # I've hijacked this field to pass in the Token instead of the password
    access_token = None

    @asyncio.coroutine
    def login():

#        login_url = tesla_base_url + '/oauth/token'
#        headers = {'Content-type': 'application/json'}
#        body = {
#            'email': conf_user,
#            'password': conf_password,
#            'client_secret': 'c7257eb71a564034f9419ee651c7d0e5f7aa6bfbd18bafb5c5c033b093bb2fa3',
#            'client_id': '81527cff06843c8634fdc09e8ac0abefb46ac849f38fe1e431c2ef2106796384',
#            'grant_type': 'password'
#        }
#
#        try:
#            with async_timeout.timeout(DEFAULT_TIMEOUT, loop=hass.loop):
#                response = yield from websession.post(login_url,
#                    headers=headers,
#                    json=body,
#                    raise_for_status=False)
#
#            if response.status != 200:
#                returned_text = yield from response.text()
#                _LOGGER.warning('Error %d on call %s:\n%s', response.status, response.url, returned_text)
#            else:
#                returned_json = yield from response.json()
#                _LOGGER.debug(returned_json)
#                access_token = returned_json['access_token']
#                return access_token
#
#        except asyncio.TimeoutError:
#            _LOGGER.warning('Timeout call %s.', response.url)
#
#        except aiohttp.ClientError:
#            _LOGGER.error('Client error %s.', response.url)
#
#        return None
        return conf_password

    @asyncio.coroutine
    def revoke(access_token):

        revoke_url = tesla_base_url + '/oauth/revoke'
        headers = {'Content-type': 'application/json'}
        body = {
            'token': access_token
        }

        try:
            with async_timeout.timeout(DEFAULT_TIMEOUT, loop=hass.loop):
                response = yield from websession.post(revoke_url,
                    headers=headers,
                    json=body,
                    raise_for_status=False)

            if response.status != 200:
                returned_text = yield from response.text()
                _LOGGER.warning('Error %d on call %s:\n%s', response.status, response.url, returned_text)
            else:
                _LOGGER.debug('revoke completed')
                return True

        except asyncio.TimeoutError:
            _LOGGER.warning('Timeout call %s.', response.url)

        except aiohttp.ClientError:
            _LOGGER.error('Client error %s.', response.url)

        return False

    @asyncio.coroutine
    def get_energy_site_id(access_token):

        list_url = tesla_base_url + '/api/1/products'
        headers = {
            'Authorization': 'Bearer ' + access_token
            }
        body = {}

        try:
            with async_timeout.timeout(DEFAULT_TIMEOUT, loop=hass.loop):
                response = yield from websession.get(list_url,
                    headers=headers,
                    json=body,
                    raise_for_status=False)

            if response.status != 200:
                returned_text = yield from response.text()
                _LOGGER.warning('Error %d on call %s:\n%s', response.status, response.url, returned_text)
            else:
                returned_json = yield from response.json()
                _LOGGER.debug(returned_json)
                for r in returned_json['response']:
                    if 'energy_site_id' in r:
                        return r['energy_site_id']
                return None

        except asyncio.TimeoutError:
            _LOGGER.warning('Timeout call %s.', response.url)

        except aiohttp.ClientError:
            _LOGGER.error('Client error %s.', response.url)

        return None

    @asyncio.coroutine
    def set_operation(access_token,energy_site_id,service_data):

        operation_url = tesla_base_url + '/api/1/energy_sites/{}/operation'.format(energy_site_id)
        headers = {
            'Content-type': 'application/json',
            'Authorization': 'Bearer ' + access_token
            }
        body = {
            'default_real_mode': service_data['real_mode']
            # 'backup_reserve_percent':int(service_data['backup_reserve_percent'])
            }
        _LOGGER.debug(body)

        try:
            with async_timeout.timeout(DEFAULT_TIMEOUT, loop=hass.loop):
                response = yield from websession.post(operation_url,
                    json=body,
                    headers=headers,
                    raise_for_status=False)

            if response.status != 200:
                returned_text = yield from response.text()
                _LOGGER.warning('Error %d on call %s:\n%s', response.status, response.url, returned_text)
            else:
                returned_json = yield from response.json()
                _LOGGER.debug('set operation successful, response: %s', returned_json)

        except asyncio.TimeoutError:
            _LOGGER.warning('Timeout call %s.', response.url)

        except aiohttp.ClientError:
            _LOGGER.error('Client error %s.', response.url)

    @asyncio.coroutine
    def async_set_operation(service):

        access_token = yield from login()
        if access_token:
            energy_site_id = yield from get_energy_site_id(access_token)
            if energy_site_id:
                yield from set_operation(access_token, energy_site_id, service.data)
            yield from revoke(access_token)

    hass.services.async_register(DOMAIN, 'set_operation', async_set_operation)

    @asyncio.coroutine
    def set_reserve(access_token,energy_site_id,service_data):

        operation_url = tesla_base_url + '/api/1/energy_sites/{}/backup'.format(energy_site_id)
        headers = {
            'Content-type': 'application/json',
            'Authorization': 'Bearer ' + access_token
            }
        body = {
            'backup_reserve_percent': int(service_data['reserve_percent'])
            }
        _LOGGER.debug(body)

        try:
            with async_timeout.timeout(DEFAULT_TIMEOUT, loop=hass.loop):
                response = yield from websession.post(operation_url,
                    json=body,
                    headers=headers,
                    raise_for_status=False)

            if response.status != 200:
                returned_text = yield from response.text()
                _LOGGER.warning('Error %d on call %s:\n%s', response.status, response.url, returned_text)
            else:
                returned_json = yield from response.json()
                _LOGGER.debug('set reserve successful, response: %s', returned_json)

        except asyncio.TimeoutError:
            _LOGGER.warning('Timeout call %s.', response.url)

        except aiohttp.ClientError:
            _LOGGER.error('Client error %s.', response.url)

    @asyncio.coroutine
    def async_set_reserve(service):

        access_token = yield from login()
        if access_token:
            energy_site_id = yield from get_energy_site_id(access_token)
            if energy_site_id:
                yield from set_reserve(access_token, energy_site_id, service.data)
            yield from revoke(access_token)

    hass.services.async_register(DOMAIN, 'set_operation', async_set_operation)
    hass.services.async_register(DOMAIN, 'set_reserve', async_set_reserve)

    return True

Hope this helps someone as a workaround until this can be fixed properly !
You can access my version of the files on github

I noticed my PW did not switch modes when my car finished charging. Came here to see if something broke as I get the same error. Apparently so. Keeping an eye out for updates…

Yep, same here, can no longer switch modes. Back to using tesla dumb logic for charging as I’m not smart enough to fix this code.

Hi Bruce, I’m keen to get this working again and until a proper fix can be enabled the workaround of having to update a token every 45 days is something I can live with,

However I’m a bit of a novice at this, so would be really grateful if you could offer any low brow instructions on how I go about implementing your fix, and which areas of the files I update with my token/username/password etc. And where i go about getting a token.

Appreciate you probably have better things to do with your weekend though.

I used TeslaFi to generate the token.
I’m not sure how else you can do it at the moment as most of the API services are broken due to this change.

As far as the other changes go.
You put the token into your config file instead of your Tesla password and you change the contents of the custom component file to the version I put in the post or obtain it from my github that is linked above.

1 Like

Thanks I shall give it a try later. I have teslafi so can do that bit.

Brilliant, worked a charm. I notice that teslafi appears to autoupdate the access token, so copying and pasting into HA every month or so should be fine, unless of course as you say this code can be updated to work with the api.

Thanks again, much appreciated.