Add support for Tesla Powerwall

Please add support for Tesla Powerwall

Buy me one, and I’ll create the integration straight away :wink:

4 Likes

I have a very rough implementation you can start with, just to change the operation mode.
I didnt like the time-based control that was provided, so I made this component to change the operation mode:

  • when superoffpeak, I set the batteries to backup-only
  • when offpeak or onpeak, I set the batteries to self-consumption

I also have a rest sensor to get the other values: instant power, imported/exported energy
Rest sensors:

- platform: rest
  name: Energy Solar
  resource: !secret solar_url
  method: GET
  verify_ssl: false
  json_attributes:
    - site
    - battery
    - load
    - solar
  value_template: '{{ value_json.load.instant_power / 1000 }}'
  unit_of_measurement: kW

- platform: template
  sensors:
    energy_solar_instant_power:
      friendly_name: Instant power
      value_template: '{{ (states.sensor.energy_solar.attributes.solar.instant_power / 1000 | float) | round(3) }}'
      unit_of_measurement: kW
    energy_solar_imported:
      friendly_name: Imported
      value_template: '{{ (states.sensor.energy_solar.attributes.solar.energy_imported / 1000 | float) | round(3) }}'
      unit_of_measurement: kWh
    energy_solar_exported:
      friendly_name: Exported
      value_template: '{{ (states.sensor.energy_solar.attributes.solar.energy_exported / 1000 | float) | round(3) }}'
      unit_of_measurement: kWh

And here is the component to set the operation mode:

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

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

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

DOMAIN = 'tesla_gateway'

# Tesla gateway is SSL but has no valid certificates
import urllib3
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)


_LOGGER = logging.getLogger(__name__)

DEFAULT_TIMEOUT = 100
CONF_INSTALLER_PASSWORD = 'installer_password'

CONFIG_SCHEMA = vol.Schema({
    DOMAIN: vol.Schema({
        vol.Required(CONF_HOST): cv.template,
        vol.Required(CONF_INSTALLER_PASSWORD): cv.string,
    }),
}, extra=vol.ALLOW_EXTRA)

@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_host_template = domain_config[CONF_HOST]
    conf_host_template.hass = hass
    conf_host = conf_host_template.async_render()
    conf_installer_password = domain_config[CONF_INSTALLER_PASSWORD]
    status_token = None

    try:
        sitemaster_url = 'https://' + conf_host + '/api/sitemaster'
        
        with async_timeout.timeout(DEFAULT_TIMEOUT, loop=hass.loop):
            response = yield from websession.get(sitemaster_url, 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()
            status_running = returned_json['running']
            status_connected_to_tesla = returned_json['connected_to_tesla']

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

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

    @asyncio.coroutine
    def login():

        login_url = 'https://' + conf_host + '/api/login/Basic'
        headers = {'Content-type':'application/json'}
        payload = {'username':'installer','password':conf_installer_password,'force_sm_off':True}

        try:        
            with async_timeout.timeout(DEFAULT_TIMEOUT, loop=hass.loop):
                response = yield from websession.post(login_url,
                    json=payload,
                    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(returned_json)
                status_token = returned_json['token']
                return status_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 complete(status_token):

        complete_url = 'https://' + conf_host + '/api/config/completed'
        headers = {'Authorization':'Bearer ' + status_token}

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

            if response.status != 202: # Completed returns 202
                returned_text = yield from response.text()
                _LOGGER.warning('Error %d on call %s:\n%s', response.status, response.url, returned_text)
            else:
                _LOGGER.debug('Operation 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 async_set_operation(service):
        
        status_token = yield from login()
        if status_token:
            
            operation_url = 'https://' + conf_host + '/api/operation'
            headers = {'Authorization':'Bearer ' + status_token}
            payload = {'real_mode':service.data['real_mode'],'backup_reserve_percent':int(service.data['backup_reserve_percent'])}
            _LOGGER.debug(payload)
            #{"real_mode":"self_consumption","backup_reserve_percent":5}
            #{"real_mode":"backup","backup_reserve_percent":100}

            try:        
                with async_timeout.timeout(DEFAULT_TIMEOUT, loop=hass.loop):
                    response = yield from websession.post(operation_url,
                        json=payload,
                        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)

            yield from complete(status_token)

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

    return True

The UI is not great, but does what I need:

3 Likes

With Tesla’s system Version 1.43.3, the above became highly unstable. I do not recommend using it.

The above component was using a “local API”, “/api/operation” seems to work sometimes, but most of the times it doesnt actually change the mode.
Myself and other people tried different things (even setting the mode twice), but I personally cannot get it stable enough for everyday use. Here is a link with more information: https://github.com/vloschiavo/powerwall2/issues/26

I personally switched to use the “owner-api” API (which is the one the Tesla app uses). This is also a non-official non-documented API. You can find more information here: https://www.teslaapi.io/

This approach is more stable for me, I had only one time the gateway didn’t change the mode. At that time I could also not change the mode with the app. I reset the gateway and was back in business. My hunch is that there is some bug in Tesla’s servers and/or firmware.

If there is some interest, I can share the code, I just need to clean it up a bit.

1 Like

Definitely interested :slight_smile: Some code would be much appreciated.

Or if I only need the SoC etc, so readonly… does this still work?

Here is the component code with the public api.
Would be great if someone takes the time to integrate it to the tesla component (it uses the same API).
Also note that this only handles one site/gateway

"""
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)

    return True
1 Like

Hi @estebanp do you have an example of how you set the batteries mode?

@james_hiscott sure, this is my automation, the triggers are connected to another automation that tracks TOU, but if you take a look at the “action” part should be easy enough:

alias: tesla_gateway_operation
trigger:
- platform: state
  entity_id: utility_meter.energy_meter_daily_imported
  to: 'superoffpeak'
- platform: state
  entity_id: utility_meter.energy_meter_daily_imported
  from: 'superoffpeak'
- platform: homeassistant
  event: start
action:
- service: tesla_gateway.set_operation
  data_template:
    real_mode: >
      {% if is_state('utility_meter.energy_meter_daily_imported', 'superoffpeak') %}
          backup
      {% else %}
          self_consumption
      {% endif %}
    backup_reserve_percent: >
      {% if is_state('utility_meter.energy_meter_daily_imported', 'superoffpeak') %}
          100
      {% else %}
          5
      {% endif %}

I posted the TOU scripts already here: Rainforest EAGLE-200 Energy Gateway - Does you have one?

Awesome, thanks @estebanp so to confirm (sorry quite new to this level of HA development):

  1. You add the public api code as a component in the HA folder (post 6), do I need to change client_secret and client_id?

  2. You set the required CONF_USERNAME and CONF_PASSWORD in configuration.yaml

  3. you add the rest sensor (in post 3) and set the solar_url to the local IP in secrets

  4. Then run the automation every time you switch tariff to set the battery to the below:

You check if you are on the super off peak tariff, if so you set the gateway to “backup” mode and the reserve percent = 100%

I am assuming this means it will start to pull from the grid and charge the battery to 100%?

@james_hiscott, no worries, it takes some time to wire all these things together…

  1. In your homeassistant/custom_components, create a folder “tesla_gateway”. Inside that folder, create a file __init__.py with the component code from the post above.
  2. In the same folder, create a manifest.json file with the following:
{
  "domain": "testla_gateway",
  "name": "Tesla Gateway",
  "documentation": "",
  "dependencies": [],
  "codeowners": [],
  "requirements": []
}
  1. In the same folder, create a file “services.yaml” with the following:
set_operation:
  description: >
    Changes operation
  fields:
    real_mode:
      description: Mode to set to the Tesla gateway.
      example: 'self_consumption, backup'
    backup_reserve_percent:
      description: Percentage of battery reserve
  1. In your configuration.yaml, add:
tesla_gateway:
  username: <username of your tesla account, can be through a secret>
  password: <password of your tesla account, can be through a secret>
  1. Restart HA
  2. After restarting, you can go to the developer section and switch the operation mode (developer tools > Services, then service: “tesla_gateway.set_operation”, and an example should show up). You can try it out there switching from backup to self_consumption.
  3. The rest sensor from the other post is to monitor the power consumption. Is not needed to change the operation mode. Moreover, changing the operation mode is done through the same API that the phone app uses, whereas monitor consumption / battery level is done by poking the gateway directly. At some point I was changing the operation mode directly to the gateway, but that became unstable.
    The other post also has the part to set TOU (time of use) based on your energy provider, which is not needed to change Tesla’s gateway mode, is just how I use it (I change to backup on super-off-peak and self-consumption for off-peak and on-peak). You can drive the automation with time or something simpler if you wanted to.
1 Like

@estebanp Thank you for bringing your explanation down to our level.

In your manifest.json code, should domain be “testla_gateway” with the extra t?

I tried it with and without the t and I’m still getting a notification saying:

The following integrations and platforms could not be set up:
- tesla_gateway
Please check your config.

Is there anything else we might be missing?

Sorry @james_hiscott , thats a typo, should be “tesla_gateway” (I had it wrong locally and didnt affect it, I think its just used for documentation)
It should be loading by having the custom_component/tesla_gateway/init.py. Depending on your HA version, you should be getting a warning at the beginning of the log statis “You are using a custom itnegration for tesla_gateway”…
Is this the first custom_components you have? the “custom_components” folder should be sibling to configuration.yaml

@estebanp I have other custom_components which are working.

Here’s what I’m seeing in the log:

Log Details (ERROR)

Logger: homeassistant.setup
First occured: 1:31:42 PM (1 occurences)
Last logged: 1:31:42 PM

Setup failed for tesla_gateway: No setup function defined.

What version of HA are you running?
There is a “setup” function, is async_setup, I dunno at what version async was added, here is the documentation: https://developers.home-assistant.io/docs/asyncio_working_with_async/

Core 0.106.2 and OS version 3.11.

In any case, I do at least have your power and energy sensors collecting data locally, so that’s nice.

How are you displaying your battery %?

Mine is even worse (106.6)…

However everything looks correct to me (and the same as other custom components:

An configuration.yaml looks good as well. Very strange

Do you think a car will come with it ?
Well, if freebies are on offer … :rofl:

1 Like

it may be that it needs the non-async versions of those functions… did a quick search and couldn’t find if they are actually required or if there is some setting that drives it.

@Matthew_Budraitis for the battery % I use a custom integration: “mini-graph-card” https://github.com/kalkih/mini-graph-card

@james_hiscott I have not tried this in hassio, dunno if there is any difference, are the same setup/update methods in your other custom components?

From which entity are you getting the battery % though? I’m not seeing that attribute in what I have coming in locally.

Is coming from “sensor.energy_battery_percentage” which is a rest sensor:

- platform: rest
  name: Energy battery percentage
  resource: !secret solar_battery_url
  method: GET
  verify_ssl: false
  value_template: '{{ value_json.percentage | float | round(2) }}'
  unit_of_measurement: '%'

solar_battery_url is: http://<gateway ip>/api/system_status/soe

1 Like