Has the firmware on your powerwall upgraded, I think may have been making security changes.
They must have been doing something, firmware hasnt updated, but a restart of the gateway has forced it to reconnect to the local network. Cable was fine, but gateway had lost its local IP address for some reason. Sensors all back up and running again now.
Still doesnt explain why the custom component is failing saying dhcp is not available when i have default config enabled. Seems I may have somethingunderlying wrong with my HA setup.
Some general clarification since there seem to be some confusion. There are two components:
-
tesla_gateway which is a hacky component I put together to change powerwall modes. Some time ago this component was using the local API (talking directly to the powerwall) but the handshake between Tesla and the PW changed and it became unstable to change mode (we would have to change it 3 times and stop/start the PW to actually change it). So instead, we went with thee public API which is the one that e.g. the phone apps use.
This component stopped working recently because the authentication was changed to SSO. SSO is “single sign on” and is a method to sign you across multiple sites. In this method, you need to authenticate through a more complex handshake, then exchange tokens from the old method and continue. The SSO seems to complain if you try to sign on frequently, so we need to keep the token around and renew it (I am testing that change, is going well).
Some of us use this component to optimize the times we have the powerwall on “backup” and “self_consumption” modes. In my case, I have a TOU energy plan with 3 different prices during the day. Times changes on weekends and holidays. To make it “cheaper” I set the powerwall on backup mode during super-offpeak and self_consumption otherwsie. I also calculate if I am not going to have enough energy to go through “on-peak” and reserve a % at off-peak. With this, I manager to consume almost never energy from the grid on “on-peak” and little on “off-peak”. This may depend on your system size and consumption profile.
This component is just doing that, changing modes and setting the reserve %. Is NOT getting the consumption values. -
powerwall is a component that was added some months ago to the official HA release. Me and others were using a rest sensor and talking directly to the powerwall. Seems you were doing the same. This was possible because the powerwall let you locally (using the local API) query these values without any authentication.
Recently, Tesla updated the firmware and now that data is no longer available without authentication. The REST sensor doesn’t have things like templates for the headers, so you need to use a component to authenticate and get that data.
Luckily, someone already wrote a component: powerwall. So, me and others have switched from using that rest sensor to this powerwall component. The powerwall component that is in the HA release still doesnt have this authentication. It will be there in the next release. To get it earlier, you can download it and place it on custom_components. HA then will override the one from HA with the one you have there. To configure this now you have to pass the local IP and the customer password.
For both of these authentications to work, the version of your PW has to be > 20.49.0
The dhcp issue seems to be related to some other setting/component in your HA configuration. There is a dhcp component: https://www.home-assistant.io/integrations/dhcp/
However, I dont have that enabled and dont hit that issue, but it could be related to my HA and network setup. I did not inspect the component to understand what that dhcp is used for. According to @jimmyleet, he updated his hass and that worked.
If that doesnt work, enable debug logging for that component and inspect the logs.
[quote=“Hodor, post:75, topic:142280”]
interestingly I have two IP addresses for my powerwall gateway, both are fixed.[/quote]
I also have two IP addresses, but one is for the LAN connection, and the other one is for the WLAN connection.
This worked fine for me. Copy the files into the newly created folder config/custom_components and restart HA. Then you can install the new integration powerwall. In the input form, enter the IP (upper field) and the password of the powerwall.
Is that true for all cases? When using the integration that came with my Home assistant image or the one suggestes by estebanp (GitHub - bdraco/powerwall: Updated powerwall integration for HASS), I had to enter the IP of my Powerwall.
Some smart people chiming in over on TMC about this change. Sharing in case some of the discussion there helps @estebanp.
1 Like
Sorry, I meant to do this sooner, didnt want to keep posting without testing it a bit more. This updated version has been working for a couple of days without any issues. Calling it frequently seems to work and so does calling it twice a day.
The local api is now working in the ‘Tesla Powerwall’ component from the public release of HA (https://www.home-assistant.io/integrations/powerwall/)
The owners API from the Tesla component is also working: https://www.home-assistant.io/integrations/tesla/
And here is the updated version of the Tesla gateway to modify the operation:
"""
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,
CONF_ACCESS_TOKEN
)
from homeassistant.helpers.aiohttp_client import async_get_clientsession
import homeassistant.helpers.config_validation as cv
DOMAIN = 'tesla_gateway'
CONF_REFRESH_TOKEN = "refresh_token"
_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,
vol.Optional(CONF_ACCESS_TOKEN, default=''): cv.string,
vol.Optional(CONF_REFRESH_TOKEN, default=''): cv.string,
}),
}, extra=vol.ALLOW_EXTRA)
tesla_base_url = 'https://owner-api.teslamotors.com'
tesla_auth_url = 'https://auth.tesla.com'
step_max_attempts = 7
step_attempt_sleep = 3
TESLA_CLIENT_ID = '81527cff06843c8634fdc09e8ac0abefb46ac849f38fe1e431c2ef2106796384'
@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]
@asyncio.coroutine
def SSO_login():
# Code extracted from https://github.com/enode-engineering/tesla-oauth2/blob/e6922c6e7805d9f65c109d90c5eaf806c5b71938/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)
for attempt in range(step_max_attempts):
with async_timeout.timeout(DEFAULT_TIMEOUT, loop=hass.loop):
response = yield from websession.get(authorize_url,
headers=headers,
params=params,
raise_for_status=False)
returned_text = yield from response.text()
if response.status == 200 and "<title>" in returned_text:
crsf_regex_result = re.search(r'name="_csrf".+value="([^"]+)"', returned_text)
if crsf_regex_result:
_LOGGER.debug('Step 1: Success on attempt %d', attempt)
break
_LOGGER.warning('Step 1: Error %d on attempt %d, call %s:\n%s', response.status, attempt, response.url, returned_text)
time.sleep(step_attempt_sleep)
else:
raise ValueError('Step 1: failed after %d attempts, last response %s:\n%s', step_max_attempts, response.status, returned_text)
# Step 2: Obtain an authorization code
csrf = crsf_regex_result.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)
for attempt in range(step_max_attempts):
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 "We could not sign you in" in returned_text and response.status == 401:
raise ValueError('Step 2: Invalid credentials. Error %d on call %s:\n%s', response.status, response.url, returned_text)
if response.status == 302 or "<title>" in returned_text:
_LOGGER.debug('Step 2: Success on attempt %d', attempt)
break
_LOGGER.warning('Step 2: Error %d on call %s:\n%s', response.status, response.url, returned_text)
time.sleep(step_attempt_sleep)
else:
raise ValueError('Step 2: failed after %d attempts, last response %s:\n%s', step_max_attempts, response.status, returned_text)
is_mfa = True if response.status == 200 and "/mfa/verify" in returned_text else False
if is_mfa:
raise ValueError('Multi-factor authentication enabled for the account and not supported')
# 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']
domain_config[CONF_ACCESS_TOKEN] = access_token
domain_config[CONF_REFRESH_TOKEN] = returned_json['refresh_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 SSO_refresh_token():
token_oauth2_url = tesla_auth_url + '/oauth2/v3/token'
headers = {
"User-Agent": "curl",
"x-tesla-user-agent": "TeslaApp/3.10.9-433/adff2e065/android/10",
"X-Requested-With": "com.teslamotors.tesla",
}
body = {
"grant_type": "refresh_token",
"refresh_token": domain_config[CONF_REFRESH_TOKEN],
"client_id": "ownerapi",
"scope": "openid email offline_access",
}
with async_timeout.timeout(DEFAULT_TIMEOUT, loop=hass.loop):
response = yield from websession.post(token_oauth2_url,
headers=headers,
data=body,
raise_for_status=False)
returned_json = yield from response.json()
access_token = returned_json['access_token']
domain_config[CONF_ACCESS_TOKEN] = access_token
domain_config[CONF_REFRESH_TOKEN] = returned_json['refresh_token']
return access_token
@asyncio.coroutine
def OWNER_get_token(access_token):
try:
token_oauth_url = tesla_base_url + '/oauth/token'
headers = {
"User-Agent": "curl",
"authorization": "bearer " + access_token,
}
body = {
"grant_type": "urn:ietf:params:oauth:grant-type:jwt-bearer",
"client_id": TESLA_CLIENT_ID,
}
with async_timeout.timeout(DEFAULT_TIMEOUT, loop=hass.loop):
response = yield from websession.post(token_oauth_url,
headers=headers,
data=body,
raise_for_status=False)
returned_json = yield from response.json()
owner_access_token = returned_json["access_token"]
return owner_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 OWNER_revoke(owner_token):
revoke_url = tesla_base_url + '/oauth/revoke'
headers = {'Content-type': 'application/json'}
body = {
'token': owner_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(owner_token):
list_url = tesla_base_url + '/api/1/products'
headers = {
'Authorization': 'Bearer ' + owner_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()
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(owner_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 ' + owner_token
}
body = {
'default_real_mode': service_data['real_mode'],
'backup_reserve_percent':int(service_data['backup_reserve_percent'])
}
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 get_owner_api_token():
access_token = domain_config[CONF_ACCESS_TOKEN]
if not access_token:
access_token = yield from SSO_login()
else:
access_token = yield from SSO_refresh_token()
if not access_token:
return None
owner_token = yield from OWNER_get_token(access_token)
return owner_token
@asyncio.coroutine
def async_set_operation(service):
owner_token = yield from get_owner_api_token()
if owner_token:
energy_site_id = yield from get_energy_site_id(owner_token)
if energy_site_id:
yield from set_operation(owner_token, energy_site_id, service.data)
yield from OWNER_revoke(owner_token)
hass.services.async_register(DOMAIN, 'set_operation', async_set_operation)
return True
Hi @estebanp,
I updated the Tesla Powerwall component from HA integrations and got that set up. I now have the PW data back in Lovelace.
I updated my __init__.py with the code from your last post. It changed the mode one time but then fails on any subsequent attempt. HA log shows:
2021-02-17 19:25:44 ERROR (MainThread) [homeassistant.core] Error executing service: <ServiceCall tesla_gateway.set_operation (c:0946cb13e5ebb5c883491eda478b757f): real_mode=backup>
Traceback (most recent call last):
File "/usr/src/homeassistant/homeassistant/core.py", line 1471, in catch_exceptions
await coro_or_task
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 281, in async_set_operation
headers=headers,
File "/config/custom_components/tesla_gateway/__init__.py", line 106, in login
_LOGGER.warning('Step 1: Error %d on attempt %d, call %s:\n%s', response.status, attempt, response.url, returned_text)
NameError: name 'returned_test' is not defined
Any suggestions?
Thanks!
Looks like I had a typo on that line, it should be ‘returned_text’, just fixed the last post. However, this is showing that you are getting an issue on the first step… if you fix that, can you post the log? check that it doesnt have anything sensitive
Hi,
Which log are you referring to? The same one? After I fixed that and restarted core, I get:
2021-02-17 20:56:36 ERROR (MainThread) [homeassistant.components.websocket_api.http.connection] [2850035768] 'backup_reserve_percent'
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 352, in async_set_operation
yield from set_operation(owner_token, energy_site_id, service.data)
File "/config/custom_components/tesla_gateway/__init__.py", line 312, in set_operation
'backup_reserve_percent':int(service_data['backup_reserve_percent'])
KeyError: 'backup_reserve_percent'
I’m calling the service via the developer tools area if that means anything.
that output means that you didn’t pass the backup_reserve_percent entry in the call. If you had a different code that allowed you to not pass that, use that code…
The log I am referring to is the log from homeassistant, but enable debug level for this component:
logger:
logs:
custom_components.tesla_gateway: debug
A ha.
So when I was calling it via developer tools, I was only trying to change the mode, not also set the reserve percentage. I didn’t have to set both before as I believe the code I had prior was someone else’s addition to control the reserve percent and it worked a bit differently (two different services in the drop down). I think your original code from a year or so ago only changed mode if memory serves.
When I put both parameters in the field, it worked. Thank you for spending your time to fix the code and help me get it working. Much appreciated!
real_mode: 'self_consumption'
backup_reserve_percent: '20'
Success!!! glad its working now!
1 Like
All working for me again after the latest update apart from one item.
I had made some cards using the sensor “sensor.powerwall_soc” which was pulled from the API which was the raw state of charge for the powerwall to around 10 decimal places. I used this as a base to work out the real charge (accounting for the power that is reserved in system etc)
This sensor now does not seem to work anymore and was pulled from the local api using api/system_status/soe
can anyone point me in the right direction of obtaining this information again?
From what I can see, the sensor.powerwall_charge has been altered to now round, and does not update very often. My local api call to soe i updated every 20 seconds, as does my solar level monitoring. Im hoping there is a way to still access these values.
My gateway serial is not work on the Tesla Powerwall integration 
How can I use it please ?
Thanks all
Denis
Tesla Gateway script seems to be broken again:
401 error on 
https://owner-api.teslamotors.com/api/1/products
Co-incidentally my Tesla app had logged out this morning and my Teslafi had lost it’s token.
I refresh teslafi token, but it into the gateway script, but I’m getting a 401 ??
Oh dear !
edit: hang on - might have been a simple typo !
This worked great for me, thank you!
One nit - the new version removed the set_reserve function. That function is my favorite because I can leave the operation mode on Advanced/time-based and just push the reserve up (in non-peak hours) and down (in peak) and it takes care of itself otherwise.
Here’s what I added back in (with tweaks to work in with the new auth code):
@asyncio.coroutine
def set_reserve(owner_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 ' + owner_token
}
body = {
'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 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):
owner_token = yield from get_owner_api_token()
if owner_token:
energy_site_id = yield from get_energy_site_id(owner_token)
if energy_site_id:
yield from set_reserve(owner_token, energy_site_id, service.data)
yield from OWNER_revoke(owner_token)
hass.services.async_register(DOMAIN, 'set_reserve', async_set_reserve)
Edit: There was a bug in the body that sent reserve_percent instead of backup_reserve_percent. It is now fixed.
Many thanks for all your hard work here, my first post so please be gentle!!
Am using developer tools to set the tesla_gateway.set_reserve to say 10
backup_reserve_percent: 10
However I’m getting this error:
Logger: homeassistant.components.websocket_api.http.connection
Source: components/websocket_api/connection.py:129
Integration: Home Assistant WebSocket API (documentation, issues)
First occurred: 12:51:35 (1 occurrences)
Last logged: 12:51:35
[547834177184] Error handling message: expected dict for dictionary value @ data['service_data']. Got 10
Any help would be appreciated.
I’d have to see more details, but here’s what I am doing that works fine:
When called from the developer/services interface, I call it as follows:
When called from an automation, I call it as follows:
action:
- service: tesla_gateway.set_reserve
data_template:
backup_reserve_percent: 50
Does that help?
Also, make sure to get the updated code I put in my addition to change reserve_percent to backup_reserve_percent, I updated it on Saturday Feb 27th I think.
Many thanks for the reply.
Am running ‘core-2021.2.3’.
I had created ‘init.py’ from your post 84 (17-Feb) above and included the change in post 93 (23-Feb). Since your post I’ve re-created ‘init.py’ with just the code from post 84, however this still fails with the same error when I run it from developer tools.
I’ve now reverted back to the combined code from both posts and still the same error.
I’ve listed my various files below, and will continue to investigate, if I get anywhere I’ll post again.
init.py
"""
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,
CONF_ACCESS_TOKEN
)
from homeassistant.helpers.aiohttp_client import async_get_clientsession
import homeassistant.helpers.config_validation as cv
DOMAIN = 'tesla_gateway'
CONF_REFRESH_TOKEN = "refresh_token"
_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,
vol.Optional(CONF_ACCESS_TOKEN, default=''): cv.string,
vol.Optional(CONF_REFRESH_TOKEN, default=''): cv.string,
}),
}, extra=vol.ALLOW_EXTRA)
tesla_base_url = 'https://owner-api.teslamotors.com'
tesla_auth_url = 'https://auth.tesla.com'
step_max_attempts = 7
step_attempt_sleep = 3
TESLA_CLIENT_ID = '81527cff06843c8634fdc09e8ac0abefb46ac849f38fe1e431c2ef2106796384'
@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]
@asyncio.coroutine
def SSO_login():
# Code extracted from https://github.com/enode-engineering/tesla-oauth2/blob/e6922c6e7805d9f65c109d90c5eaf806c5b71938/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)
for attempt in range(step_max_attempts):
with async_timeout.timeout(DEFAULT_TIMEOUT, loop=hass.loop):
response = yield from websession.get(authorize_url,
headers=headers,
params=params,
raise_for_status=False)
returned_text = yield from response.text()
if response.status == 200 and "<title>" in returned_text:
crsf_regex_result = re.search(r'name="_csrf".+value="([^"]+)"', returned_text)
if crsf_regex_result:
_LOGGER.debug('Step 1: Success on attempt %d', attempt)
break
_LOGGER.warning('Step 1: Error %d on attempt %d, call %s:\n%s', response.status, attempt, response.url, returned_text)
time.sleep(step_attempt_sleep)
else:
raise ValueError('Step 1: failed after %d attempts, last response %s:\n%s', step_max_attempts, response.status, returned_text)
# Step 2: Obtain an authorization code
csrf = crsf_regex_result.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)
for attempt in range(step_max_attempts):
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 "We could not sign you in" in returned_text and response.status == 401:
raise ValueError('Step 2: Invalid credentials. Error %d on call %s:\n%s', response.status, response.url, returned_text)
if response.status == 302 or "<title>" in returned_text:
_LOGGER.debug('Step 2: Success on attempt %d', attempt)
break
_LOGGER.warning('Step 2: Error %d on call %s:\n%s', response.status, response.url, returned_text)
time.sleep(step_attempt_sleep)
else:
raise ValueError('Step 2: failed after %d attempts, last response %s:\n%s', step_max_attempts, response.status, returned_text)
is_mfa = True if response.status == 200 and "/mfa/verify" in returned_text else False
if is_mfa:
raise ValueError('Multi-factor authentication enabled for the account and not supported')
# 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']
domain_config[CONF_ACCESS_TOKEN] = access_token
domain_config[CONF_REFRESH_TOKEN] = returned_json['refresh_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 SSO_refresh_token():
token_oauth2_url = tesla_auth_url + '/oauth2/v3/token'
headers = {
"User-Agent": "curl",
"x-tesla-user-agent": "TeslaApp/3.10.9-433/adff2e065/android/10",
"X-Requested-With": "com.teslamotors.tesla",
}
body = {
"grant_type": "refresh_token",
"refresh_token": domain_config[CONF_REFRESH_TOKEN],
"client_id": "ownerapi",
"scope": "openid email offline_access",
}
with async_timeout.timeout(DEFAULT_TIMEOUT, loop=hass.loop):
response = yield from websession.post(token_oauth2_url,
headers=headers,
data=body,
raise_for_status=False)
returned_json = yield from response.json()
access_token = returned_json['access_token']
domain_config[CONF_ACCESS_TOKEN] = access_token
domain_config[CONF_REFRESH_TOKEN] = returned_json['refresh_token']
return access_token
@asyncio.coroutine
def OWNER_get_token(access_token):
try:
token_oauth_url = tesla_base_url + '/oauth/token'
headers = {
"User-Agent": "curl",
"authorization": "bearer " + access_token,
}
body = {
"grant_type": "urn:ietf:params:oauth:grant-type:jwt-bearer",
"client_id": TESLA_CLIENT_ID,
}
with async_timeout.timeout(DEFAULT_TIMEOUT, loop=hass.loop):
response = yield from websession.post(token_oauth_url,
headers=headers,
data=body,
raise_for_status=False)
returned_json = yield from response.json()
owner_access_token = returned_json["access_token"]
return owner_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 OWNER_revoke(owner_token):
revoke_url = tesla_base_url + '/oauth/revoke'
headers = {'Content-type': 'application/json'}
body = {
'token': owner_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(owner_token):
list_url = tesla_base_url + '/api/1/products'
headers = {
'Authorization': 'Bearer ' + owner_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()
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(owner_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 ' + owner_token
}
body = {
'default_real_mode':service_data['real_mode'],
'backup_reserve_percent':int(service_data['backup_reserve_percent'])
}
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 get_owner_api_token():
access_token = domain_config[CONF_ACCESS_TOKEN]
if not access_token:
access_token = yield from SSO_login()
else:
access_token = yield from SSO_refresh_token()
if not access_token:
return None
owner_token = yield from OWNER_get_token(access_token)
return owner_token
@asyncio.coroutine
def async_set_operation(service):
owner_token = yield from get_owner_api_token()
if owner_token:
energy_site_id = yield from get_energy_site_id(owner_token)
if energy_site_id:
yield from set_operation(owner_token, energy_site_id, service.data)
yield from OWNER_revoke(owner_token)
hass.services.async_register(DOMAIN, 'set_operation', async_set_operation)
@asyncio.coroutine
def set_reserve(owner_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 ' + owner_token
}
body = {
'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 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):
owner_token = yield from get_owner_api_token()
if owner_token:
energy_site_id = yield from get_energy_site_id(owner_token)
if energy_site_id:
yield from set_reserve(owner_token, energy_site_id, service.data)
yield from OWNER_revoke(owner_token)
hass.services.async_register(DOMAIN, 'set_reserve', async_set_reserve)
return True
manifest.json
{
"domain": "tesla_gateway",
"name": "Tesla Gateway",
"documentation": "",
"dependencies": [],
"codeowners": [],
"requirements": []
}
services.yaml
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:
backup_reserve_percent:
description: Percentage of battery reserve
example: 70
configuration.yaml
tesla_gateway:
username: !secret tesla_username
password: !secret tesla_password
# Logging
logger:
logs:
custom_components.tesla_gateway: debug
