Hi all,
I’ve recently put together an automation for my Tesla which I thought others might find interesting. I haven’t tried packaging this up as a blueprint etc as it’s probably a bit niche to be worth the bother, but it might give someone a building block for their own project.
I am not a professional programmer and as far as I’m concerned YAML is the work of the devil so my code may not be the best - but it works.
The elevator pitch is - I wanted the car to pre-heat/cool itself in time for any journeys that I wanted to make, without me having to remember to do it manually, based on information in my Google Calendar.
What I have ended up with looks a bit like this:
The major components are (starting with the simpler end of the process):
- An HA automation that is triggered by a date_time helper. When triggered it launches a script which turns on the climate control in the car, notifies a user group that it’s done so and sets a timeout. If the car is not in use when the timeout elapses it turns the climate back off (the assumption is that we made a mistake on this occasion) and notifies the user group again.
- An automation which detects that a user has gotten in to the car and cancels the script launched above - this proved necessary as some really short journeys can take place during the timeout and lead to ugly spurious messages.
The guts of the operation is:
3) A Python script which:
- Polls Google Calendar to pull the next few upcoming appointments from a list of calendars
- Filters out any for which there is no location set, no start time set (ie, all day events) or where there is a flag text string in the appointment description.
- Filters out any where the car is already within 250m of the location. (We have two kids events back to back at the same location, so there are two calendar entries but obviously no need to drive anywhere.)
For each remaining event it then:
-
Works out if they are a future event (we want to drive there), an event in progress (we want to drive back at the end) or a ‘drop off’ event in-progress (eg - somewhere you take your kids and have to go back at the end. If the event is in progress it means we want to drive back in time for the end of the event).
-
For the ‘future’ and ‘drop off/in progress’ events it uses Google’s Travel Time API to work out how long it’s going to take to drive to the location for the relevant time from the car’s current location (exposed through the Tesla component/HA’s states API)
Once we have done that for all events we work out which is the next relevant event - ie, what is going to cause us to use the car next. Once we have that we: -
Use the Tesla component to work out how big the difference between the car’s temperature and the desirable temperature is, and uses that to estimate how long the climate control needs to be on for.
-
Based on how long it needs to be on for we also calculate a sensible timeout after which we assume we made a mistake and turn the climate back off again.
-
Finally, we use HA’s states API to set the value of some date_time, input_text and input_number helpers that can be picked up by the automations covered in item 1) above.
Deployment wise, it’s 2 HA automations and 1 script, though you need the Tesla Custom integration from HACS for them to work, and then there’s the python component, which runs entirely outside Home Assistant via Cron. You can run this on a totally separate host provided it can access the HA APIs.
The script uses a few libraries that are all installable via pip:
google-auth google-api-python-client python-dateutil requests
The faffiest part of the process is getting the right authentication tokens set up with Google for API access.
Automation 1: (Trigger pre-heating)
- alias: Tesla dynamic preheat start
mode: restart
description: "The car may be about to make a journey"
id: tesla_dynamic_preheat_start
trigger:
platform: time
at: input_datetime.tesla_next_journey_preheat_time
condition:
- condition: state
entity_id: binary_sensor.model_y_user_present
state: 'off'
action:
- alias: "Preheat car immediately"
service: script.turn_on
target:
entity_id: script.tesla_climate_on_and_timeout
- alias: "Notify user"
service: notify.yourgroup
data:
message: "Preconditioning started for journey from {{ states('input_text.tesla_next_journey_desc') }}"
title: "Preheating.."
data:
push:
interruption-level: time-sensitive
group: tesla-preheat
url: /dashboard-tesla/0
actions:
- action: "cancel_preheating"
title: "Abort!"
- alias: "Wait for user response to cancel"
wait_for_trigger:
- platform: event
event_type: mobile_app_notification_action
event_data:
action: "cancel_preheating"
timeout:
minutes: "{{ states('input_number.tesla_next_journey_preheat_timeout_mins') | int }}"
continue_on_timeout: false
- alias: "Stop Preheating"
choose:
- conditions: "{{ wait.trigger.event.data.action == 'cancel_preheating' }}"
sequence:
- service: climate.set_hvac_mode
target:
entity_id: climate.hvac_climate_system
data:
hvac_mode: 'off'
Script 1: (Turn on heating and set timeout)
preheat_tesla_from_schedule:
alias: Tesla climate on and timeout
description: "Turn on tesla climate but turn off if car remains unused"
mode: restart
sequence:
# Step 1: Set HVAC mode to 'heat_cool'
- service: climate.set_hvac_mode
target:
entity_id: climate.hvac_climate_system
data:
hvac_mode: 'heat_cool'
# Step 2: Wait for specified number of minutes
- delay:
minutes: "{{ states('input_number.tesla_next_journey_preheat_timeout_mins') | int }}"
# Step 3: Check if 'binary_sensor.model_y_user_present' is 'Off' and set HVAC mode to 'Off'
- condition: state
entity_id: binary_sensor.model_y_user_present
state: 'off'
- service: climate.set_hvac_mode
target:
entity_id: climate.hvac_climate_system
data:
hvac_mode: 'off'
# Step 4 - notify heating cancelled
- service: notify.yourgroup
data:
message: "Preheating timeout elapsed"
title: "Preheating turned off"
data:
push:
interruption-level: time-sensitive
Automation 2: (Cancel the above script if someone gets in the car)
- alias: "Cancel Tesla Preheat on User Presence"
id: cancel_preheat_start
description: "Stops the preheat script when the user is detected in the car"
trigger:
- platform: state
entity_id: binary_sensor.model_y_user_present
to: 'on'
action:
- service: script.turn_off
target:
entity_id: script.preheat_tesla_from_schedule
mode: single
Python script 1: (Integrate with the Google APIs and set home assistant values)
#!/usr/bin/python3
# sudo apt-get install python3-pip
# pip install google-auth google-api-python-client
# pip install python-dateutil
# pip install requests
from pprint import pprint, pformat
from dateutil import parser, tz
from google.oauth2.credentials import Credentials
from googleapiclient.discovery import build
from urllib.parse import quote_plus
import requests
import math
import datetime
import syslog
import sys
import logging
import logging.handlers
import re
# Sensitive config
#!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!
# Home Assistant configuration
HA_URL = 'http://homeassistant.yours:8123'
HA_TOKEN = 'Your HA token here'
# Google API configuration
SCOPES = ['https://www.googleapis.com/auth/calendar.readonly']
GOOGLE_API_CREDENTIALS = './google_calendar_credentials.json'
GEOCODING_API_KEY = 'Your token'
TRAVEL_TIME_API_KEY = 'Your token'
# Calendars to check
CALENDAR_IDS = [
'Your calendar 1',
'Your calendar 2'
]
#!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!
# Create a logger
logger = logging.getLogger('GCAL2HA')
logger.setLevel(logging.DEBUG)
# Create a handler for writing to syslog
syslog_handler = logging.handlers.SysLogHandler(address='/dev/log')
syslog_handler.setLevel(logging.WARNING) # Set level for syslog (e.g., WARNING)
# Create a handler for writing to the console
console_handler = logging.StreamHandler(sys.stdout)
console_handler.setLevel(logging.DEBUG) # Set level for console output (e.g., DEBUG)
# Create a formatter and add it to the handlers
formatter = logging.Formatter('%(name)s - %(levelname)s - %(message)s')
syslog_handler.setFormatter(formatter)
console_handler.setFormatter(formatter)
# Add the handlers to the logger
logger.addHandler(syslog_handler)
logger.addHandler(console_handler)
# Home Assistant API call function
def call_ha_api(endpoint, method='get', payload=None):
headers = {
'Authorization': f'Bearer {HA_TOKEN}',
'Content-Type': 'application/json',
}
url = f"{HA_URL}/api/{endpoint}"
try:
if method.lower() == 'get':
logger.debug(f"GETing: {url} with headers {pformat(headers)}")
response = requests.get(url, headers=headers)
elif method.lower() == 'post':
logger.debug(f"POSTing: {url} with headers {pformat(headers)} and payload {payload}")
response = requests.post(url, json=payload, headers=headers)
except Exception(e):
logger.error(f"Unable to connect to Home Assistant on {endpoint}{url}")
sys.exit(1)
logger.debug(f"HA said: HTTP {response.status_code}: {response.json()}")
return response.json()
# Function to get current location of the Tesla from Home Assistant
def get_tesla_location():
tesla_state = call_ha_api('states/device_tracker.model_y_location_tracker')
if tesla_state and 'attributes' in tesla_state and 'latitude' in tesla_state['attributes'] and 'longitude' in tesla_state['attributes']:
return {
'latitude': tesla_state['attributes']['latitude'],
'longitude': tesla_state['attributes']['longitude']
}
else:
# Log the error: send a warning to syslog
logger.error("Tesla position received from Home Assistant is not in the expected format. Using home address instead.")
return {
'latitude': "Your house",
'longitude': "Your house"
}
# return None
# Helper to inform an estimate of how much warmup time we need
def determine_tesla_temperature_delta():
try:
tesla_climate_info = call_ha_api('states/climate.hvac_climate_system')
set_temp = tesla_climate_info['attributes']['temperature']
if set_temp is None:
logger.debug(f"No setpoint temperature was returned, so defaulting to 20 degrees")
set_temp = 20
inside_temp = tesla_climate_info['attributes']['current_temperature']
# Check if the data is in the correct format and is a number
temperature_delta = abs(float(inside_temp) - float(set_temp))
logger.debug(f"Delta between inside and outside of car is {temperature_delta} degrees")
return temperature_delta
except Exception as e:
logger.error(f"Error in fetching temperature data: {e}")
return None
# Function to convert address to latitude and longitude using the Geocoding API
def get_lat_lon_from_address(address):
logger.info(f"Trying to geocode: {address}")
address = quote_plus(address)
url = f"https://maps.googleapis.com/maps/api/geocode/json?address={address}&key={GEOCODING_API_KEY}"
#print(url)
response = requests.get(url)
response_json = response.json()
if response_json['status'] == 'OK':
lat = response_json['results'][0]['geometry']['location']['lat']
lon = response_json['results'][0]['geometry']['location']['lng']
return lat, lon
else:
logger.warning(f"Geocoding API error: {response_json['status']} for {address}")
return (None, None)
# Function to calculate travel time to location using the Google Maps Travel Time API
def get_travel_time_to_location(origin, destination, mode, time):
origin_str = f"{origin['latitude']},{origin['longitude']}"
destination_str = f"{destination['latitude']},{destination['longitude']}"
logger.debug(f"Converting {pformat(time)} for URL")
# Convert to Unix timestamp
time = int(time.timestamp())
url = f"https://maps.googleapis.com/maps/api/directions/json?origin={origin_str}&destination={destination_str}&{mode}_time={time}&key={TRAVEL_TIME_API_KEY}"
logger.debug(f"Calling: {url}")
response = requests.get(url)
response_json = response.json()
#logger.debug(f"Google said: {pformat(response_json)}")
if response_json['status'] == 'OK':
travel_duration = response_json['routes'][0]['legs'][0]['duration']['value']
return travel_duration
else:
logger.error(f"Failed to get travel time. Skipping event.")
return None
# Function to get events from Google Calendar
def get_google_calendar_events(credentials_path, calendar_ids):
creds = Credentials.from_authorized_user_file(credentials_path, SCOPES)
service = build('calendar', 'v3', credentials=creds)
now = datetime.datetime.utcnow().isoformat() + 'Z'
events_result = []
for calendar_id in calendar_ids:
try:
events = service.events().list(calendarId=calendar_id, timeMin=now,
maxResults=4, singleEvents=True,
orderBy='startTime').execute()
except Exception:
logger.error(f"Got exception when fetching events for calendar {calendar_id}")
logger.debug(f"Fetched {len(events.get('items'))} items for {calendar_id}")
#logger.debug(f"New items: {pformat(events.get('items'))}")
events_result.extend(events.get('items', []))
#logger.debug(f"Cumulative: {pformat(events_result)}")
logger.debug(f"Fetched {len(events_result)} events")
return events_result
# Parse Google's event times, where the TZ is in a separate field to the timestamp
def parse_datetime(dt_str, tz_str=None):
dt = parser.parse(dt_str)
if tz_str:
timezone = tz.gettz(tz_str)
dt = dt.astimezone(timezone)
elif dt.tzinfo is None or dt.tzinfo.utcoffset(dt) is None:
# Assume UTC if no timezone is provided
dt = dt.replace(tzinfo=tz.UTC)
return dt
# Function to filter and massage Google Calendar events
def filter_events(events):
filtered_events = []
logger.debug(f"Starting filter with {len(events)} items")
for event in events:
try:
event.pop("recurringEventId")
event.pop("etag")
event.pop("htmlLink")
event.pop("iCalUID")
event.pop("id")
event.pop("sequence")
except KeyError:
pass
# slogger.debug(f"Filtering: {pformat(event)}")
if 'description' in event and 'ha_no_precondition' in event['description']:
logger.debug(f"Skipped event {event['summary']} due to no_precondition instruction")
continue
if 'dateTime' not in event['start']:
logger.debug(f"Skipped event {event['summary']} due to missing start time")
continue
event['start_time'] = parse_datetime(event['start'].get('dateTime'), event['start'].get('timeZone'))
event['end_time'] = parse_datetime(event['end'].get('dateTime'), event['end'].get('timeZone'))
current_time = datetime.datetime.now(tz=tz.tzlocal())
if current_time < event['start_time']:
event['status'] = "future"
elif event['start_time'] <= current_time <= event['end_time']:
event['status'] = "in_progress"
else:
event['status'] = "finished"
if event['start_time'].time() <= datetime.time(7, 30) or 'location' not in event:
logger.debug(f"Skipped event {event['summary']} due to early start time")
continue
if 'location' not in event:
logger.debug(f"Skipped event {event['summary']} due to missing location)")
continue
filtered_events.append(event)
logger.debug(f"Ended filter with {len(filtered_events)} items")
return filtered_events
# Calculate the distance between two sets of coordinates
def haversine(coord1, coord2):
lat1, lon1 = map(float, coord1)
lat2, lon2 = map(float, coord2)
R = 6371000 # radius of Earth in meters
phi1 = math.radians(lat1)
phi2 = math.radians(lat2)
delta_phi = math.radians(lat2 - lat1)
delta_lambda = math.radians(lon2 - lon1)
a = math.sin(delta_phi / 2)**2 + math.cos(phi1) * math.cos(phi2) * math.sin(delta_lambda / 2)**2
c = 2 * math.atan2(math.sqrt(a), math.sqrt(1 - a))
distance = R * c # output distance in meters
return distance
# Remove redundant information from location strings
def clean_location(s):
logger.debug(f"Cleaning location {pformat(s)}")
s = s.replace('\n', ', ') # Remove line breaks
# Define patterns to remove
remove_patterns = [r'uk', r'united kingdom', r'england']
# This is a list of patterns that mark the end of the useful address in events close to you (where you don't need the town/county/etc part)
terminate_patterns = [r' regexp1', r'regexp2', r"regexp3"]
# Remove specified patterns (case-insensitive)
for pattern in remove_patterns:
s = re.sub(pattern, '', s, flags=re.IGNORECASE)
logger.debug(f"Stripped remove patterns: {pformat(s)}")
# Terminate string at specified patterns (case-insensitive)
for pattern in terminate_patterns:
matches = list(re.finditer(pattern, s, re.IGNORECASE))
if matches:
last_match = matches[-1]
s = s[:last_match.end()]
logger.debug(f"Truncating at {pattern}")
break
s = re.sub(r'\s+', ' ', s) # Consolidate spaces
s = re.sub(r', ,', ', ', s) # Clean up commas
return s.strip()
# Look at each event and work out what preconditioning time is appropriate for each one
def calculate_precondition_times(events, tesla_location):
arrival_time_offset = 5 # Time for end of journey (parking, etc). This ACCELLERATES activation of climate
exit_time_offset = 4 # Time to return to car at end of appt. This DELAYS activation of climate
preheat_timeout = 20 # Time we're prepared to let the preheating run for before assuming we made an error and cancelling
my_house = "My address. Make sure it geocodes" # Assumed destination for return trips
temp_delta = determine_tesla_temperature_delta()
for event in events:
preheat_time_required = 2 # base time for the car to warm up, modified below
concise_location=clean_location(event['location'])
logger.debug(f"Cleaned location: {concise_location}")
# Set some sensible defaults
event['next_journey_destination']=event['location'] # Default is that we drive out to where the event is
event['relevant_time'] = event['start_time']
event['mode']="arrival"
event['summary_string'] = f"home to {concise_location} for the start of {event['summary']}"
if 'description' in event and 'ha_no_linger' in event['description'] and event['status'] == "in_progress" :
# This is a 'drop off' event in progress, meaning the end of the event is effectively another arrival time
# to return there. The thing that needs changing is arriving at the end time, not the start.
event['relevant_time'] = event['end_time']
event['summary_string'] = f"home to {concise_location} for the end of {event['summary']}"
elif event['status'] == "in_progress":
# A normal event where we have remained at the location but need to drive home (assumption) afterwards.
# The end time needs to be used as the departure time (not arrival), and location needs to be home
event['next_journey_destination'] = my_house
event['mode']="departure"
event['relevant_time'] = event['end_time']
event['summary_string'] = f"{concise_location} to Home at end of {event['summary']}"
# Remaining scenario is normal, travel to beginning of appointment. Defaults apply
event['lat'], event['lon'] = get_lat_lon_from_address(event['next_journey_destination'])
if event['lat'] == None:
logger.debug(f"Abandoning {event['summary_string']} due to inoperable location")
continue
# If the car is already at the relevant destination then we don't need to handle this one
event['distance_to_destination'] = haversine((tesla_location['latitude'], tesla_location['longitude']), (event['lat'], event['lon']))
if event['distance_to_destination'] < 250:
logger.debug(f"The car is only {event['distance_to_destination']}m away from {event['next_journey_destination']}, so skipping.")
continue
event['travel_time'] = get_travel_time_to_location(tesla_location, {'latitude': event['lat'], 'longitude': event['lon']}, event['mode'], event['relevant_time'])
if event['travel_time'] == None:
logger.debug(f"Abandoning {event['summary_string']} due to unobtainable travel time.")
continue
# If we're driving to the location we need to account for travel time, but not if we're leaving it
if event['mode'] is 'departure':
departure_time = event['relevant_time'] + datetime.timedelta(minutes=exit_time_offset)
else:
departure_time = event['relevant_time'] - datetime.timedelta(seconds= (event['travel_time'] + arrival_time_offset*60) )
# Estimate a likely preheat time based on the delta between the inside of the car and the climate setpoint
logger.debug(f"Calculating preheat time. Delta:{temp_delta}, Default preheat:{preheat_time_required}, Arrival allowance: {arrival_time_offset}")
preheat_time_required = preheat_time_required + (temp_delta * 1)
# Calculate how long we're going to let the climate stay on for before shutting it off automatically if the car isn't used
event['preheat_timeout'] = preheat_timeout + preheat_time_required
# Calculate actual climate activation time-of-day, accounting for temperature
event['preheat_time'] = (departure_time - datetime.timedelta(minutes=preheat_time_required)).strftime('%Y-%m-%d %H:%M:%S')
logger.debug(f"Calculated values added: {pformat(event)}")
return events
# Function to update Home Assistant entities
def update_ha_entities(lat, lon, preheat_time, preheat_timeout, summary):
# Update latitude and longitude
attributes_dict = {
'editable': True,
'friendly_name': "Tesla next destination coordinates"
}
call_ha_api('states/input_text.tesla_next_journey_destination_coordinates', 'post', {
'state': f"{lat},{lon}",
'attributes' : attributes_dict
})
# Break the string in to its components for HA
dt_obj = datetime.datetime.strptime(preheat_time, '%Y-%m-%d %H:%M:%S')
attributes_dict = {
'has_date': True,
'has_time': True,
'editable': True,
'friendly_name': "Tesla next journey preheat time",
'year': dt_obj.year,
'month': dt_obj.month,
'day': dt_obj.day,
'hour': dt_obj.hour,
'minute': dt_obj.minute,
'second' : 0
}
call_ha_api('states/input_datetime.tesla_next_journey_preheat_time', 'post', {
'state': preheat_time,
'attributes': attributes_dict
})
# Update timeout
attributes_dict = {
'editable': True,
'friendly_name': "Tesla preheat timeout for this journey"
}
call_ha_api('states/input_number.tesla_next_journey_preheat_timeout_mins', 'post', {
'state': preheat_timeout,
'attributes' : attributes_dict
})
# Update address
attributes_dict = {
'editable': True,
'friendly_name': "Tesla next journey description"
}
call_ha_api('states/input_text.tesla_next_journey_desc', 'post', {
'state': summary,
'attributes' : attributes_dict
})
# Main logic
def main():
# Get current location of Tesla
tesla_location = get_tesla_location()
if tesla_location is None:
logger.error("Failed to retrieve Tesla location from Home Assistant.")
sys.exit(1) # Exits the program with an error status
# Retrieve events from Google Calendar
events = get_google_calendar_events(GOOGLE_API_CREDENTIALS, CALENDAR_IDS)
# Filter events
filtered_events = filter_events(events)
# Work out what travel time is appropriate for each event
filtered_events = calculate_precondition_times(filtered_events, tesla_location)
# Find the event with the most imminent departure time
earliest_event = None
for event in filtered_events:
#logger.info(f"Analysing {pformat(event)}")
if 'preheat_time' in event and event['preheat_time'] is not None:
if earliest_event is None or event['preheat_time'] < earliest_event['preheat_time']:
#logger.info("Found new earliest event")
earliest_event = event
# Update Home Assistant with the event's details
if earliest_event:
logger.info(f'Notified Home assistant to schedule {pformat(earliest_event)}')
update_ha_entities(
earliest_event['lat'],
earliest_event['lon'],
earliest_event['preheat_time'],
earliest_event['preheat_timeout'],
earliest_event['summary_string']
)
else:
logger.error(f"Failed to find a valid earliest event")
if __name__ == '__main__':
main()