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.
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.
Here’s some references to the new API that I’m sure will be helpful to someone who knows what they’re doing with the HA code to update it.
https://tesla-api.timdorr.com/api-basics/authentication
Adding @estebanp as he may not know his HA PW integration is broken and he seems to be code knowledgable. 
Interesting… my PWs have been switching fine, including this morning (I do automated operation switch at least twice a day). I restarted HA and now see the failure…
Tesla changed their API again… https://tesla-api.timdorr.com/api-basics/authentication
The Tesla component that HomeAssistant has uses the python TeslaApi which is still using the old method as well: https://github.com/mlowijs/tesla_api/blob/de199527c6723f89502d53d92ef8ec87718cf912/tesla_api/__init__.py, so that is also broken.
Here is a working implementation of the new method: https://github.com/enode-engineering/tesla-oauth2/blob/bb579721726cc00f4d1c480a5ef47d9340bb2d06/tesla.py
I am cleaning it up and integrating it into my component, once I have a working version I will post it. For the time being, using a fixed token will work for 45 days at a time.
Here is a working version. I will take a deeper read in the following weeks to check I am not missing something.
I definitely have to improve the login/revoke so we keep the tokens alive, but this should get you back to a working state.
"""
Monitors and controls the Tesla gateway.
"""
import logging
import aiohttp
import asyncio
import async_timeout
import base64
import hashlib
import json
import os
import re
import time
from urllib.parse import parse_qs
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'
tesla_auth_url = 'https://auth.tesla.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():
# Code extracted from https://github.com/enode-engineering/tesla-oauth2/blob/bb579721726cc00f4d1c480a5ef47d9340bb2d06/tesla.py
# Login process explained at https://tesla-api.timdorr.com/api-basics/authentication
authorize_url = tesla_auth_url + '/oauth2/v3/authorize'
callback_url = tesla_auth_url + '/void/callback'
headers = {
"User-Agent": "curl",
"x-tesla-user-agent": "TeslaApp/3.10.9-433/adff2e065/android/10",
"X-Requested-With": "com.teslamotors.tesla",
}
verifier_bytes = os.urandom(86)
code_verifier = base64.urlsafe_b64encode(verifier_bytes).rstrip(b"=")
code_challenge = base64.urlsafe_b64encode(hashlib.sha256(code_verifier).digest()).rstrip(b"=").decode("utf-8")
state = base64.urlsafe_b64encode(os.urandom(16)).rstrip(b"=").decode("utf-8")
params = (
("client_id", "ownerapi"),
("code_challenge", code_challenge),
("code_challenge_method", "S256"),
("redirect_uri", callback_url),
("response_type", "code"),
("scope", "openid email offline_access"),
("state", state),
)
try:
# Step 1: Obtain the login page
_LOGGER.debug('Step 1: GET %s\nparams %s', authorize_url, params)
with async_timeout.timeout(DEFAULT_TIMEOUT, loop=hass.loop):
response = yield from websession.get(authorize_url,
headers=headers,
params=params,
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)
return None
returned_text = yield from response.text()
if not "<title>" in returned_text:
_LOGGER.warning('Error %d on call %s:\n%s', response.status, response.url, returned_text)
return None
# Step 2: Obtain an authorization code
csrf = re.search(r'name="_csrf".+value="([^"]+)"', returned_text).group(1)
transaction_id = re.search(r'name="transaction_id".+value="([^"]+)"', returned_text).group(1)
body = {
"_csrf": csrf,
"_phase": "authenticate",
"_process": "1",
"transaction_id": transaction_id,
"cancel": "",
"identity": conf_user,
"credential": conf_password,
}
_LOGGER.debug('Step 2: POST %s\nparams: %s\nbody: %s', authorize_url, params, body)
with async_timeout.timeout(DEFAULT_TIMEOUT, loop=hass.loop):
response = yield from websession.post(authorize_url,
headers=headers,
params=params,
data=body,
raise_for_status=False,
allow_redirects=False)
returned_text = yield from response.text()
if response.status != 302 or "<title>" in returned_text:
_LOGGER.warning('Error %d on call %s:\n%s', response.status, response.url, returned_text)
return None
is_mfa = True if response.status == 200 and "/mfa/verify" in returned_text else False
if is_mfa:
_LOGGER.warning('Multi-factor authentication enabled for the account and not supported')
return None
# Step 3: Exchange authorization code for bearer token
code = parse_qs(response.headers["location"])[callback_url + '?code']
token_url = tesla_auth_url + '/oauth2/v3/token'
body = {
"grant_type": "authorization_code",
"client_id": "ownerapi",
"code_verifier": code_verifier.decode("utf-8"),
"code": code,
"redirect_uri": callback_url
}
_LOGGER.debug('Step 3: POST %s', token_url)
with async_timeout.timeout(DEFAULT_TIMEOUT, loop=hass.loop):
response = yield from websession.post(token_url,
headers=headers,
data=body,
raise_for_status=False)
returned_json = yield from response.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
I also had to restart the system locally (turn off and on a PW). The app was showing the change, but the change was not reflected in the powerflow. Changing it from the app was also not having any effect. Restarting the powerwall did the trick.
I’ve copied in this code and restarted.
The set_reserve service seems to have vanished?
is that intentional?
Has it been combined into the set_operation service?
don’t worry, I’ve added it back in.
I guess it was a tweak at some point and you’ve reverted it in your update.
Thanks
Yeah… I always used it combined so I have just one method
Note the above code wont work if you enable multi-factor authentication.
Ok. Thanks. I guess I’ll stick with putting the token in for now then.
This replaces the content of init.py, right?
I got an error.
Logger: homeassistant.components.websocket_api.http.connection
Source: custom_components/tesla_gateway/__init__.py:102
Integration: Home Assistant WebSocket API (documentation, issues)
First occurred: 10:52:15 AM (4 occurrences)
Last logged: 11:00:01 AM
[2844986360] 'NoneType' object has no attribute 'group'
Traceback (most recent call last):
File "/usr/src/homeassistant/homeassistant/components/websocket_api/commands.py", line 136, in handle_call_service
await hass.services.async_call(
File "/usr/src/homeassistant/homeassistant/core.py", line 1455, in async_call
task.result()
File "/usr/src/homeassistant/homeassistant/core.py", line 1490, in _execute_service
await handler.job.target(service_call)
File "/config/custom_components/tesla_gateway/__init__.py", line 268, in async_set_operation
access_token = yield from login()
File "/config/custom_components/tesla_gateway/__init__.py", line 102, in login
csrf = re.search(r'name="_csrf".+value="([^"]+)"', returned_text).group(1)
AttributeError: 'NoneType' object has no attribute 'group'
I did cycle a PW but it persists. Guessing I did something wrong.