The following two files are updated to support both operation mode changes and battery reserve percent changes.
All the other files are identical to the excellent post of @estebanp from above.
The first, services.yaml, needs to be put into /custom_components/tesla_gateway:
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:
reserve_percent:
description: Percentage of battery reserve
example: 70
The second, __init__.py needs to go in the same directory:
"""
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)
@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
Thatâs it! One thing to note is that if you use these and the Tesla app doesnât show any changes, try killing and restarting the app. The Tesla app caches old values and so even if the changes happen you wonât see them unless you exit and restart the app.
