So I have a scenario where I need to be somewhere at a specific time and I decided today to throw this scenario at GPT to help me construct an app to send me a notification based on my current location to tell me that I need to leave at whatever time it tells me in order to reach my destination on time.
So I add this code to the apps.yaml
my_event:
module: calendar_travel_notifier
class: CalendarTravelNotifier
api_key: "GOOGLEMAPSAPIKEY"
device_tracker: "device_tracker.phone"
calendar_entity: "calendar.calendar_entity"
notification_device: "notification_device"
event_name: "Event Name From the Calendar Event"
arrival_mode: "end" #Do you want to be notified at the start or the end of the event
notification_offset_hours: 3
The fun part
import appdaemon.plugins.hass.hassapi as hass
import datetime
import requests
class CalendarTravelNotifier(hass.Hass):
def initialize(self):
# Retrieve inputs from apps.yaml
self.api_key = self.args.get("api_key")
self.device_tracker = self.args.get("device_tracker")
self.calendar_entity = self.args.get("calendar_entity")
self.notification_device = self.args.get("notification_device")
self.event_name = self.args.get("event_name")
self.arrival_mode = self.args.get("arrival_mode", "end") # Options: "start" or "end"
self.notification_offset_hours = self.args.get("notification_offset_hours", 3)
# Store the notification states to avoid duplicate alerts for both start and end times
self.notification_sent_start = False
self.notification_sent_end = False
# Schedule the check_event method to run every 15 minutes
self.run_every(self.check_event, self.datetime(), 60 * 15)
# Validate inputs
if not all([self.api_key, self.device_tracker, self.calendar_entity, self.notification_device, self.event_name]):
self.error("Missing required configuration parameters. Check apps.yaml.")
return
# Run the check every 5 minutes
self.run_every(self.check_event, self.datetime() + datetime.timedelta(seconds=5), 300)
self.log("CalendarTravelNotifier automation initialized")
def check_event(self, kwargs):
event_time_start, event_location_start = self.get_calendar_event_time_and_location(self.event_name, "start")
event_time_end, event_location_end = self.get_calendar_event_time_and_location(self.event_name, "end")
if event_time_start is None or event_time_end is None:
self.log("No matching event found in the calendar.")
return
# Determine the notify time based on arrival_mode
if self.arrival_mode == "start":
notify_time = event_time_start - datetime.timedelta(hours=self.notification_offset_hours)
destination = event_location_start
else:
notify_time = event_time_end - datetime.timedelta(hours=self.notification_offset_hours)
destination = event_location_end
# Only proceed if we have a valid destination (location)
if destination is None:
self.log(f"Event '{self.event_name}' found, but no location specified.")
return
# Get current time
now = self.datetime()
# Check if it's time to notify about leaving
if now >= notify_time:
if self.arrival_mode == "start" and not self.notification_sent_start:
self.send_notification("start", event_time_start, destination)
self.notification_sent_start = True
elif self.arrival_mode == "end" and not self.notification_sent_end:
self.send_notification("end", event_time_end, destination)
self.notification_sent_end = True
# Reset the notification flags after the event has ended
if now >= event_time_end:
self.log("Event has ended. Resetting notification flags.")
self.notification_sent_start = False # Ready for next start notification
self.notification_sent_end = False # Ready for next end notification
def get_calendar_event_time_and_location(self, event_name, mode):
"""Fetch the start or end time and location of a calendar event."""
calendar_state = self.get_state(self.calendar_entity, attribute="all")
if not calendar_state:
self.log(f"Calendar entity {self.calendar_entity} returned no data.", level="ERROR")
return None, None
attributes = calendar_state.get("attributes", {})
message = attributes.get("message", "")
if not message:
self.log(f"No 'message' attribute found for calendar entity {self.calendar_entity}.", level="WARNING")
return None, None
# Check if the message matches the event name
if event_name.lower() in message.lower():
if mode == "start" and "start_time" in attributes:
event_time = self.parse_datetime(attributes["start_time"])
event_location = attributes.get("location", None)
return event_time, event_location
elif mode == "end" and "end_time" in attributes:
event_time = self.parse_datetime(attributes["end_time"])
event_location = attributes.get("location", None)
return event_time, event_location
self.log(f"Event '{event_name}' not found in the 'message' attribute.", level="INFO")
return None, None
def send_notification(self, event_type, event_time, destination):
"""Calculate the travel time and send a notification."""
# Get current location (assumes you have a device tracker with 'latitude' and 'longitude' attributes)
latitude = self.get_state(self.device_tracker, attribute="latitude")
longitude = self.get_state(self.device_tracker, attribute="longitude")
if latitude is None or longitude is None:
self.log("Current location (latitude/longitude) not found.")
return
departure_time = self.calculate_departure_time(latitude, longitude, destination, event_time)
if departure_time:
# Send a notification to your mobile device
message = (
f"Leave by {departure_time.strftime('%I:%M %p')} to reach '{destination}' "
f"by {event_time.strftime('%I:%M %p')}."
)
self.call_service(
f"notify/{self.notification_device}",
title="Travel Notification",
message=message
)
self.log(f"Notification sent: {message}")
else:
self.log("Unable to calculate travel time.")
def calculate_departure_time(self, lat, lon, destination, event_time):
"""Calculate the departure time using Google Maps Distance Matrix API."""
url = "https://maps.googleapis.com/maps/api/distancematrix/json"
params = {
"origins": f"{lat},{lon}",
"destinations": destination,
"departure_time": "now",
"traffic_model": "best_guess",
"key": self.api_key,
}
response = requests.get(url, params=params)
if response.status_code == 200:
data = response.json()
travel_time_seconds = data["rows"][0]["elements"][0]["duration_in_traffic"]["value"]
travel_time = datetime.timedelta(seconds=travel_time_seconds)
departure_time = event_time - travel_time
return departure_time
else:
self.log(f"Error in Distance Matrix API: {response.text}")
return None
I know that HA has an integration already for working with google maps but that does a poll every 5 mins of course unless you create an automation to only do it when you tell it to do so.
But I didn’t really want an entity for how long it will take me to get there. The fun thing is that if this is a recurring event like it is in my case then it will just automatically fire off at the start/end of my event.
For my usecase its for picking up/dropping off my kids with their Mom and since I have a 2+ hour drive to where we exchange at, I wanted to make sure that wherever I’m at on that day I get notified with the current drive time.
Edit had to fix the attributes for the device tracker. That was my bad