Please add support for Tesla Powerwall
Buy me one, and I’ll create the integration straight away 
5 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 
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
@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):
-
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?
-
You set the required CONF_USERNAME and CONF_PASSWORD in configuration.yaml
-
you add the rest sensor (in post 3) and set the solar_url to the local IP in secrets
-
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…
- In your homeassistant/custom_components, create a folder “tesla_gateway”. Inside that folder, create a file
__init__.pywith the component code from the post above. - In the same folder, create a
manifest.jsonfile with the following:
{
"domain": "testla_gateway",
"name": "Tesla Gateway",
"documentation": "",
"dependencies": [],
"codeowners": [],
"requirements": []
}
- 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
- 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>
- Restart HA
- 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.
- 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 … 
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