I have a very rough implementation you can start with, just to change the operation mode.
I didnt like the time-based control that was provided, so I made this component to change the operation mode:
- when superoffpeak, I set the batteries to backup-only
- when offpeak or onpeak, I set the batteries to self-consumption
I also have a rest sensor to get the other values: instant power, imported/exported energy
Rest sensors:
- platform: rest
name: Energy Solar
resource: !secret solar_url
method: GET
verify_ssl: false
json_attributes:
- site
- battery
- load
- solar
value_template: '{{ value_json.load.instant_power / 1000 }}'
unit_of_measurement: kW
- platform: template
sensors:
energy_solar_instant_power:
friendly_name: Instant power
value_template: '{{ (states.sensor.energy_solar.attributes.solar.instant_power / 1000 | float) | round(3) }}'
unit_of_measurement: kW
energy_solar_imported:
friendly_name: Imported
value_template: '{{ (states.sensor.energy_solar.attributes.solar.energy_imported / 1000 | float) | round(3) }}'
unit_of_measurement: kWh
energy_solar_exported:
friendly_name: Exported
value_template: '{{ (states.sensor.energy_solar.attributes.solar.energy_exported / 1000 | float) | round(3) }}'
unit_of_measurement: kWh
And here is the component to set the operation mode:
"""
Monitors and controls the Tesla gateway.
"""
import logging
import aiohttp
import asyncio
import async_timeout
import json
import voluptuous as vol
from homeassistant.const import (
CONF_HOST,
CONF_PASSWORD
)
from homeassistant.helpers.aiohttp_client import async_get_clientsession
import homeassistant.helpers.config_validation as cv
from homeassistant.const import CONF_HOST
DOMAIN = 'tesla_gateway'
# Tesla gateway is SSL but has no valid certificates
import urllib3
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
_LOGGER = logging.getLogger(__name__)
DEFAULT_TIMEOUT = 100
CONF_INSTALLER_PASSWORD = 'installer_password'
CONFIG_SCHEMA = vol.Schema({
DOMAIN: vol.Schema({
vol.Required(CONF_HOST): cv.template,
vol.Required(CONF_INSTALLER_PASSWORD): cv.string,
}),
}, extra=vol.ALLOW_EXTRA)
@asyncio.coroutine
def async_setup(hass, config):
# Tesla gateway is SSL but has no valid certificates
websession = async_get_clientsession(hass, verify_ssl=False)
domain_config = config[DOMAIN]
conf_host_template = domain_config[CONF_HOST]
conf_host_template.hass = hass
conf_host = conf_host_template.async_render()
conf_installer_password = domain_config[CONF_INSTALLER_PASSWORD]
status_token = None
try:
sitemaster_url = 'https://' + conf_host + '/api/sitemaster'
with async_timeout.timeout(DEFAULT_TIMEOUT, loop=hass.loop):
response = yield from websession.get(sitemaster_url, raise_for_status=False)
if response.status != 200:
returned_text = yield from response.text()
_LOGGER.warning('Error %d on call %s:\n%s', response.status, response.url, returned_text)
else:
returned_json = yield from response.json()
status_running = returned_json['running']
status_connected_to_tesla = returned_json['connected_to_tesla']
except asyncio.TimeoutError:
_LOGGER.warning('Timeout call %s.', response.url)
except aiohttp.ClientError:
_LOGGER.error('Client error %s.', response.url)
@asyncio.coroutine
def login():
login_url = 'https://' + conf_host + '/api/login/Basic'
headers = {'Content-type':'application/json'}
payload = {'username':'installer','password':conf_installer_password,'force_sm_off':True}
try:
with async_timeout.timeout(DEFAULT_TIMEOUT, loop=hass.loop):
response = yield from websession.post(login_url,
json=payload,
headers=headers,
raise_for_status=False)
if response.status != 200:
returned_text = yield from response.text()
_LOGGER.warning('Error %d on call %s:\n%s', response.status, response.url, returned_text)
else:
returned_json = yield from response.json()
_LOGGER.debug(returned_json)
status_token = returned_json['token']
return status_token
except asyncio.TimeoutError:
_LOGGER.warning('Timeout call %s.', response.url)
except aiohttp.ClientError:
_LOGGER.error('Client error %s.', response.url)
return None
@asyncio.coroutine
def complete(status_token):
complete_url = 'https://' + conf_host + '/api/config/completed'
headers = {'Authorization':'Bearer ' + status_token}
try:
with async_timeout.timeout(DEFAULT_TIMEOUT, loop=hass.loop):
response = yield from websession.get(complete_url,
headers=headers,
raise_for_status=False)
if response.status != 202: # Completed returns 202
returned_text = yield from response.text()
_LOGGER.warning('Error %d on call %s:\n%s', response.status, response.url, returned_text)
else:
_LOGGER.debug('Operation completed')
return True
except asyncio.TimeoutError:
_LOGGER.warning('Timeout call %s.', response.url)
except aiohttp.ClientError:
_LOGGER.error('Client error %s.', response.url)
return False
@asyncio.coroutine
def async_set_operation(service):
status_token = yield from login()
if status_token:
operation_url = 'https://' + conf_host + '/api/operation'
headers = {'Authorization':'Bearer ' + status_token}
payload = {'real_mode':service.data['real_mode'],'backup_reserve_percent':int(service.data['backup_reserve_percent'])}
_LOGGER.debug(payload)
#{"real_mode":"self_consumption","backup_reserve_percent":5}
#{"real_mode":"backup","backup_reserve_percent":100}
try:
with async_timeout.timeout(DEFAULT_TIMEOUT, loop=hass.loop):
response = yield from websession.post(operation_url,
json=payload,
headers=headers,
raise_for_status=False)
if response.status != 200:
returned_text = yield from response.text()
_LOGGER.warning('Error %d on call %s:\n%s', response.status, response.url, returned_text)
else:
returned_json = yield from response.json()
_LOGGER.debug('set operation successful, response: %s', returned_json)
except asyncio.TimeoutError:
_LOGGER.warning('Timeout call %s.', response.url)
except aiohttp.ClientError:
_LOGGER.error('Client error %s.', response.url)
yield from complete(status_token)
hass.services.async_register(DOMAIN, 'set_operation', async_set_operation)
return True
The UI is not great, but does what I need:
Some code would be much appreciated.
