I can access fine through the TEG-xxx SSID at 192.168.91.1, but not from my home LAN. It gets an IP address from my home DHCP server and Tx/Rx to the cloud is working. Tesla app works. But the controller seems to be ignoring home LAN subnet traffic, even pings.
Does this exists as code checked in somewhere?
I think @fnord123 's code from Sept 27 is what I want, but Iām not sure what other files youāre talking about.
Iām trying to setup automation to maximize PW usage using Solcast forecast data.
Something seems to have changed at the Tesla end of the API.
I understand they are widening the use of 2FA.
This code seems to fail now and the error behind the scenes isā¦
2021-01-30 14:02:57 WARNING (MainThread) [custom_components.tesla_gateway] Error 400 on call https://owner-api.teslamotors.com/oauth/token:
{"response":"endpoint_deprecated:_please_update_your_app."}
Iām hoping someone better than me can re-write this code to use the new authentication methods.
But in the meantime, Iāve fudged it by passing in my Tesla Auth instead of my password.
I then commented out all the login code and just have that return the token that I supplied.
Iāll need to manually refresh the token every 45 days.
I tried to use a refresh token to do the login and get a token, but this seems to be broken too.
My code now looks like thisā¦
"""
Monitors and controls the Tesla gateway.
"""
import logging
import aiohttp
import asyncio
import async_timeout
import json
import time
import voluptuous as vol
from homeassistant.const import (
CONF_USERNAME,
CONF_PASSWORD
)
from homeassistant.helpers.aiohttp_client import async_get_clientsession
import homeassistant.helpers.config_validation as cv
DOMAIN = 'tesla_gateway'
_LOGGER = logging.getLogger(__name__)
DEFAULT_TIMEOUT = 100
CONFIG_SCHEMA = vol.Schema({
DOMAIN: vol.Schema({
vol.Required(CONF_USERNAME): cv.string,
vol.Required(CONF_PASSWORD): cv.string,
}),
}, extra=vol.ALLOW_EXTRA)
tesla_base_url = 'https://owner-api.teslamotors.com'
@asyncio.coroutine
def async_setup(hass, config):
# Tesla gateway is SSL but has no valid certificates
websession = async_get_clientsession(hass, verify_ssl=False)
domain_config = config[DOMAIN]
conf_user = domain_config[CONF_USERNAME]
conf_password = domain_config[CONF_PASSWORD] # I've hijacked this field to pass in the Token instead of the password
access_token = None
@asyncio.coroutine
def login():
# login_url = tesla_base_url + '/oauth/token'
# headers = {'Content-type': 'application/json'}
# body = {
# 'email': conf_user,
# 'password': conf_password,
# 'client_secret': 'c7257eb71a564034f9419ee651c7d0e5f7aa6bfbd18bafb5c5c033b093bb2fa3',
# 'client_id': '81527cff06843c8634fdc09e8ac0abefb46ac849f38fe1e431c2ef2106796384',
# 'grant_type': 'password'
# }
#
# try:
# with async_timeout.timeout(DEFAULT_TIMEOUT, loop=hass.loop):
# response = yield from websession.post(login_url,
# headers=headers,
# json=body,
# raise_for_status=False)
#
# if response.status != 200:
# returned_text = yield from response.text()
# _LOGGER.warning('Error %d on call %s:\n%s', response.status, response.url, returned_text)
# else:
# returned_json = yield from response.json()
# _LOGGER.debug(returned_json)
# access_token = returned_json['access_token']
# return access_token
#
# except asyncio.TimeoutError:
# _LOGGER.warning('Timeout call %s.', response.url)
#
# except aiohttp.ClientError:
# _LOGGER.error('Client error %s.', response.url)
#
# return None
return conf_password
@asyncio.coroutine
def revoke(access_token):
revoke_url = tesla_base_url + '/oauth/revoke'
headers = {'Content-type': 'application/json'}
body = {
'token': access_token
}
try:
with async_timeout.timeout(DEFAULT_TIMEOUT, loop=hass.loop):
response = yield from websession.post(revoke_url,
headers=headers,
json=body,
raise_for_status=False)
if response.status != 200:
returned_text = yield from response.text()
_LOGGER.warning('Error %d on call %s:\n%s', response.status, response.url, returned_text)
else:
_LOGGER.debug('revoke completed')
return True
except asyncio.TimeoutError:
_LOGGER.warning('Timeout call %s.', response.url)
except aiohttp.ClientError:
_LOGGER.error('Client error %s.', response.url)
return False
@asyncio.coroutine
def get_energy_site_id(access_token):
list_url = tesla_base_url + '/api/1/products'
headers = {
'Authorization': 'Bearer ' + access_token
}
body = {}
try:
with async_timeout.timeout(DEFAULT_TIMEOUT, loop=hass.loop):
response = yield from websession.get(list_url,
headers=headers,
json=body,
raise_for_status=False)
if response.status != 200:
returned_text = yield from response.text()
_LOGGER.warning('Error %d on call %s:\n%s', response.status, response.url, returned_text)
else:
returned_json = yield from response.json()
_LOGGER.debug(returned_json)
for r in returned_json['response']:
if 'energy_site_id' in r:
return r['energy_site_id']
return None
except asyncio.TimeoutError:
_LOGGER.warning('Timeout call %s.', response.url)
except aiohttp.ClientError:
_LOGGER.error('Client error %s.', response.url)
return None
@asyncio.coroutine
def set_operation(access_token,energy_site_id,service_data):
operation_url = tesla_base_url + '/api/1/energy_sites/{}/operation'.format(energy_site_id)
headers = {
'Content-type': 'application/json',
'Authorization': 'Bearer ' + access_token
}
body = {
'default_real_mode': service_data['real_mode']
# 'backup_reserve_percent':int(service_data['backup_reserve_percent'])
}
_LOGGER.debug(body)
try:
with async_timeout.timeout(DEFAULT_TIMEOUT, loop=hass.loop):
response = yield from websession.post(operation_url,
json=body,
headers=headers,
raise_for_status=False)
if response.status != 200:
returned_text = yield from response.text()
_LOGGER.warning('Error %d on call %s:\n%s', response.status, response.url, returned_text)
else:
returned_json = yield from response.json()
_LOGGER.debug('set operation successful, response: %s', returned_json)
except asyncio.TimeoutError:
_LOGGER.warning('Timeout call %s.', response.url)
except aiohttp.ClientError:
_LOGGER.error('Client error %s.', response.url)
@asyncio.coroutine
def async_set_operation(service):
access_token = yield from login()
if access_token:
energy_site_id = yield from get_energy_site_id(access_token)
if energy_site_id:
yield from set_operation(access_token, energy_site_id, service.data)
yield from revoke(access_token)
hass.services.async_register(DOMAIN, 'set_operation', async_set_operation)
@asyncio.coroutine
def set_reserve(access_token,energy_site_id,service_data):
operation_url = tesla_base_url + '/api/1/energy_sites/{}/backup'.format(energy_site_id)
headers = {
'Content-type': 'application/json',
'Authorization': 'Bearer ' + access_token
}
body = {
'backup_reserve_percent': int(service_data['reserve_percent'])
}
_LOGGER.debug(body)
try:
with async_timeout.timeout(DEFAULT_TIMEOUT, loop=hass.loop):
response = yield from websession.post(operation_url,
json=body,
headers=headers,
raise_for_status=False)
if response.status != 200:
returned_text = yield from response.text()
_LOGGER.warning('Error %d on call %s:\n%s', response.status, response.url, returned_text)
else:
returned_json = yield from response.json()
_LOGGER.debug('set reserve successful, response: %s', returned_json)
except asyncio.TimeoutError:
_LOGGER.warning('Timeout call %s.', response.url)
except aiohttp.ClientError:
_LOGGER.error('Client error %s.', response.url)
@asyncio.coroutine
def async_set_reserve(service):
access_token = yield from login()
if access_token:
energy_site_id = yield from get_energy_site_id(access_token)
if energy_site_id:
yield from set_reserve(access_token, energy_site_id, service.data)
yield from revoke(access_token)
hass.services.async_register(DOMAIN, 'set_operation', async_set_operation)
hass.services.async_register(DOMAIN, 'set_reserve', async_set_reserve)
return True
Hope this helps someone as a workaround until this can be fixed properly !
You can access my version of the files on github
I noticed my PW did not switch modes when my car finished charging. Came here to see if something broke as I get the same error. Apparently so. Keeping an eye out for updatesā¦
Yep, same here, can no longer switch modes. Back to using tesla dumb logic for charging as Iām not smart enough to fix this code.
Hi Bruce, Iām keen to get this working again and until a proper fix can be enabled the workaround of having to update a token every 45 days is something I can live with,
However Iām a bit of a novice at this, so would be really grateful if you could offer any low brow instructions on how I go about implementing your fix, and which areas of the files I update with my token/username/password etc. And where i go about getting a token.
Appreciate you probably have better things to do with your weekend though.
I used TeslaFi to generate the token.
Iām not sure how else you can do it at the moment as most of the API services are broken due to this change.
As far as the other changes go.
You put the token into your config file instead of your Tesla password and you change the contents of the custom component file to the version I put in the post or obtain it from my github that is linked above.
1 Like
Thanks I shall give it a try later. I have teslafi so can do that bit.
Brilliant, worked a charm. I notice that teslafi appears to autoupdate the access token, so copying and pasting into HA every month or so should be fine, unless of course as you say this code can be updated to work with the api.
Thanks again, much appreciated.
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?
- edit
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.
@nathanielh, yes, I only changed the login function. If you have other changes, you can just copy the contents of the login function.
Apparently the first request can fail, the https://github.com/enode-engineering/tesla-oauth2/blob/e6922c6e7805d9f65c109d90c5eaf806c5b71938/tesla.py script has a retry (will attempt 7 times). I will add that retry since I have had one of those failures.
Here is with the retries:
"""
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'
step_max_attempts = 7
step_attempt_sleep = 3
@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/main/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_test)
# 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 response.status == 302 and not "<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:
_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
Thanks, still no go. I replaced the login function in my file, then added in the required imports at the top and added the auth url and step statements below it. Once that was all added, I got the same result whenever I attempt to call the service.
2021-02-07 14:29:19 ERROR (MainThread) [homeassistant.components.websocket_api.http.connection] [2842641528] '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 325, in async_set_reserve
access_token = yield from login()
File "/config/custom_components/tesla_gateway/__init__.py", line 107, in login
csrf = re.search(r'name="_csrf".+value="([^"]+)"', returned_text).group(1)
AttributeError: 'NoneType' object has no attribute 'group'
Iām not a coder, but from some googling it seems the regex isnāt finding a match and reporting back ānoneā which is causing the error. Unsure why I am getting a different experience than you. I power cycled both Powerwalls again but that didnāt help.
My config:
Version core-2021.2.1
Installation Type Home Assistant OS
Development false
Supervisor true
Docker true
Virtual Environment false
Python Version 3.8.7
Operating System Family Linux
Operating System Version 5.4.79-v7l
CPU Architecture armv7l
Host Operating System Home Assistant OS 5.10
Update Channel stable
Supervisor Version supervisor-2021.01.7
Docker Version 19.03.13
Updated my previous post to consider the missing hidden field and retry (thats what the regex is looking for).
If you donāt get the above to find the field after the retries, it will output the returned page. Can you filter out sensitive data and post it?