Advertisement
Not a member of Pastebin yet?
Sign Up,
it unlocks many cool features!
- #!/usr/bin/python
- # Dependency: https://github.com/invisibleroads/socketIO-client
- # Install client first: sudo pip install -U socketIO-client
- # Disclaimer:
- # -I'm a Python novice
- #
- # Usage:
- # -Edit the car{} variable below, change the id's to your Automatic car id's
- # -Edit the googleMapsApiKey
- # -Start from (Linux) command-line to begin listening for websocket notifications: https://developer.automatic.com/api-reference/#real-time-events
- #
- # Notes:
- # -The notification occurs in the function: 'notifyNetwork' at the end of this file. This notification is unique to my network. It places popups on all the TV's and Desktops, flashes the lights, and causes Alexa to read the message outloud (all at once)
- # -There's a function in there that notifies my carpool partner that I've left the house in the morning, this probably doesn't apply to you
- # -There's a lot of code to account for the fact that (a) automatic sends notifications out of order, so I only show new ones, and (b) notifications often come late :(
- # -The logic for car starting/stopping = always notify, but the logic for location_updated is to only notify me if the car is on its way home, and then only repeat an update of ETA if the car has made 50% progress home since its last notification
- # -The two cars configured below are our Honda and Scion, but the 3rd car represents the Automatic app dashboard test events
- #
- # Video of this in action:
- # -(end of the video, last 15 seconds) --> https://www.youtube.com/watch?v=y09_YaduvEk
- #
- # Author: matt@farleyfamily.net
- #
- import logging
- logging.getLogger('socketIO-client').setLevel(logging.DEBUG)
- import sys
- import datetime
- import time
- import subprocess
- import json
- import requests
- import os
- from socketIO_client import SocketIO, LoggingNamespace
- """
- Initialize global vars ****************************************************
- """
- vehicleId = ""
- history = {}
- history_counter = 0
- history_limit = 400
- event = {}
- car = {}
- googleMapsApiKey = 'ABC123'
- # Tia's honda setup
- car['C_905zzzzzzzzzzzzz'] = {}
- car['C_905zzzzzzzzzzzzz']['name'] = 'Honda Minivan'
- car['C_905zzzzzzzzzzzzz']['previousMilesFromHome'] = 0.0
- car['C_905zzzzzzzzzzzzz']['currentMilesFromHome'] = 0.0
- # Matt's Scion setup
- car['C_d8daaaaaaaaaaaaa'] = {}
- car['C_d8daaaaaaaaaaaaa']['name'] = 'Scion tC'
- car['C_d8daaaaaaaaaaaaa']['previousMilesFromHome'] = 0.0
- car['C_d8daaaaaaaaaaaaa']['currentMilesFromHome'] = 0.0
- # Lambo
- car['C_40ebbbbbbbbbbbbb'] = {}
- car['C_40ebbbbbbbbbbbbb']['name'] = 'Lamborghini'
- car['C_40ebbbbbbbbbbbbb']['previousMilesFromHome'] = 0.0
- car['C_40ebbbbbbbbbbbbb']['currentMilesFromHome'] = 0.0
- """
- Triage websocket events, decide what to do ****************************************************
- """
- # *args can accept multiple arguments, coming through as args[0], args[1], etc
- def triage(*args):
- # Debug:
- print(args[0])
- # Any globals that will be modified must be called out
- global event
- global history
- global history_counter
- global vehicleId
- # Car ID to isolate tracking without cross-talk confusion
- vehicleId = args[0]['vehicle']['id']
- # Test for new event but occurs prior to an event we already received (thus can be disregarded, even though we missed it, tan-pis)
- # Will error on the first event since there's no last event
- try:
- newTimestamp = datetime.datetime.strptime(args[0]['location']['created_at'], '%Y-%m-%dT%H:%M:%S.%fZ');
- lastTimestamp = datetime.datetime.strptime(event[vehicleId]['location']['created_at'], '%Y-%m-%dT%H:%M:%S.%fZ');
- if newTimestamp < lastTimestamp:
- print("Skipping historic lagging event, we already received newer events: " + args[0]['type'])
- return
- except Exception, e:
- print("FYI - Exception testing newTimestamp vs Old Timestamp: " + str(e))
- pass
- # Capture all the json data in event
- event[vehicleId] = args[0];
- # Notify Carpool that I'm started my car (if it's early in the morning and we're turning the car on)
- if event[vehicleId]['type'] == 'ignition:on':
- if datetime.time(5,20) <= datetime.datetime.now().time() <= datetime.time(6,15):
- if datetime.datetime.today().weekday() < 5:
- notifyCarpool()
- # Test for duplicate
- if event[vehicleId]['id'] in history.values():
- print("Skipping duplicate: " + event[vehicleId]['type'])
- return
- # New notification, record it to history
- if history_counter > history_limit: history_counter = 0
- history[history_counter] = event[vehicleId]['id']
- history_counter += 1
- # Share what we're dealing with
- print("\n" + event[vehicleId]['type'] + " @ " + datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S"))
- # Google Location Geocode and Distance Matrix ETA
- ETA = getETA()
- location = getLocation()
- if event[vehicleId]['type'] == 'ignition:on':
- car[vehicleId]['previousMilesFromHome'] = 0.0
- car[vehicleId]['currentMilesFromHome'] = 0.0
- notifyNetwork(car[vehicleId]['name'] + " Started", "Ignition on at " + location + ", " + ETA)
- if event[vehicleId]['type'] == 'ignition:off':
- car[vehicleId]['previousMilesFromHome'] = 0.0
- car[vehicleId]['currentMilesFromHome'] = 0.0
- notifyNetwork(car[vehicleId]['name'] + " Parked", "Ignition off at " + location)
- if event[vehicleId]['type'] == 'location:updated':
- print('PreviousMiles = ' + str(car[vehicleId]['previousMilesFromHome']));
- print('CurrentMiles = ' + str(car[vehicleId]['currentMilesFromHome']));
- try:
- percentChange = ((car[vehicleId]['previousMilesFromHome'] - car[vehicleId]['currentMilesFromHome']) / car[vehicleId]['previousMilesFromHome']) * 100
- except Exception, e:
- # Divide by zero when car[vehicleId]['previousMilesFromHome'] is zero
- percentChange = 0.0
- car[vehicleId]['previousMilesFromHome'] = car[vehicleId]['currentMilesFromHome']
- # Only notify en route if there is a significant change in progress towards the house (or we're less than a mile away)
- if percentChange > 50.0 or car[vehicleId]['currentMilesFromHome'] == 0.0:
- car[vehicleId]['previousMilesFromHome'] = car[vehicleId]['currentMilesFromHome']
- notifyNetwork(car[vehicleId]['name'] + " En Route", "At " + location + ", " + ETA)
- else:
- print("Skipping due to lack of progress *towards* home; percent change = " + str(percentChange))
- return
- """
- Let Adam know I'm coming ****************************************************
- """
- def notifyCarpool():
- print("Texting Adam farley that we are leaving now...")
- # Notify Adam
- command = 'sms 1235551212 "Farley car started @ ' + datetime.datetime.now().strftime("%H:%M") + '. I assume they are coming for you. -Jarvis"'
- with open(os.devnull, 'wb') as devnull:
- subprocess.check_call(command, shell=True, stdout=devnull, stderr=subprocess.STDOUT)
- # Notify Matt
- command = 'sms 1235551111 "Farley car started @ ' + datetime.datetime.now().strftime("%H:%M") + '. I assume they are coming for you. -Jarvis"'
- with open(os.devnull, 'wb') as devnull:
- subprocess.check_call(command, shell=True, stdout=devnull, stderr=subprocess.STDOUT)
- return
- """
- Websocket Error ****************************************************
- """
- def on_error(*args):
- print('on_error', args)
- """
- Get ETA via Google Distance Matrix ****************************************************
- """
- def getETA():
- global vehicleId
- global googleMapsApiKey
- try:
- response = requests.get('https://maps.googleapis.com/maps/api/distancematrix/json?origins='+str(event[vehicleId]['location']['lat'])+','+str(event[vehicleId]['location']['lon'])+'&destinations=Houston,%20TX&key='+googleMapsApiKey+'&units=imperial')
- json_data = json.loads(response.text)
- text = json_data['rows'][0]['elements'][0]['duration']['text']
- seconds = json_data['rows'][0]['elements'][0]['duration']['value']
- timestamp = datetime.datetime.strptime(event[vehicleId]['location']['created_at'], '%Y-%m-%dT%H:%M:%S.%fZ')
- timestamp = timestamp - datetime.timedelta(hours=5)
- # Give ETA as event timestamp plus google travel time to home
- ETA = timestamp + datetime.timedelta(seconds=seconds)
- # Record last ETA Distance and this ETA Distance to help determine if we'll notify en route
- distance = str(json_data['rows'][0]['elements'][0]['distance']['text']).split()
- if distance[1] == 'mi':
- car[vehicleId]['currentMilesFromHome'] = float(distance[0]);
- else:
- car[vehicleId]['currentMilesFromHome'] = 0.0;
- return text + " / " + str(car[vehicleId]['currentMilesFromHome']) + " miles from home, ETA " + ETA.strftime("%I:%M %p")
- except Exception, e:
- print 'Location unknown, distance matrix calculation failed: ' + str(e)
- return 'unknown ETA'
- """
- Google Geocode Location ****************************************************
- """
- def getLocation():
- global vehicleId
- global googleMapsApiKey
- text = 'unknown location'
- try:
- response = requests.get('https://maps.googleapis.com/maps/api/geocode/json?latlng='+str(event[vehicleId]['location']['lat'])+','+str(event[vehicleId]['location']['lon'])+'&key='+googleMapsApiKey)
- json_data = json.loads(response.text)
- #print(json_data['results'][0]['address_components'])
- for component in json_data['results'][0]['address_components']:
- if component['types'][0] == 'route':
- text = component['long_name']
- except Exception, e:
- print('Location unknown, geocode failed: ' + str(e))
- pass
- return text
- """
- Notify Network ****************************************************
- """
- def notifyNetwork(header, message):
- global vehicleId
- # Notifying network
- # $1 = Icon
- # $2 = Prefix
- # $3 = Subject / Header
- # $4 = Message
- # $5 = Type / App
- command = '/mnt/documents/bin/notify_network /mnt/documents/Icons/' + event[vehicleId]['vehicle']['id'] + '.png '
- command += '"' + header + ' " '
- # If this event occurred more than a few minutes ago, then add a timestamp
- timestamp = datetime.datetime.strptime(event[vehicleId]['location']['created_at'], '%Y-%m-%dT%H:%M:%S.%fZ')
- timestamp = timestamp - datetime.timedelta(hours=5) # API sends them five hours later than CST
- tsNow = datetime.datetime.now()
- d1_ts = time.mktime(timestamp.timetuple())
- d2_ts = time.mktime(tsNow.timetuple())
- minutesAgo = int(d2_ts-d1_ts) / 60
- if minutesAgo > 4:
- command += '"- ' + timestamp.strftime("%I:%M %p") + '" '
- else:
- command += '"" '
- command += '"' + message + '" '
- command += '"automatic" '
- with open(os.devnull, 'wb') as devnull:
- subprocess.check_call(command, shell=True, stdout=devnull, stderr=subprocess.STDOUT)
- """
- Main Execution ****************************************************
- """
- io = SocketIO('https://stream.automatic.com', 443, LoggingNamespace, params={'token': 'clientid:zzzzzzzzzzzzzzzzzzzzz'})
- #io.on('trip:finished', triage)
- io.on('location:updated', triage)
- io.on('ignition:on', triage)
- io.on('ignition:off', triage)
- io.on('error', on_error)
- io.wait()
Advertisement
Add Comment
Please, Sign In to add comment
Advertisement