AWS IoT Button : I am going to home button

Private Talk

I coded AWS IoT button lambda function called "I am going home" last year but I haven't release sorce code.
I was busy on 2019 due to changing my job.
Early of the year I was working as a computer engineer at so called "SIer" :P.
But at middle of the year I started my career as a car software engineer. Car is core industry in my country Japan and software such as connected technology becomes more important comporent on car. I am still enjoying my engineering life as car embedded software developper.

Let's back to a AWS Solusion

It takes about 1 hour commuting between my apartment and office by train. When I got on the return train to my home, always I email my wife about the time I will get home. It is simple task but it is a bit of work.
So I coded this AWS lambda function.

The folloings are full part of this Python function.
ImGoingHome.py

import boto3
import requests
import json
import math
from datetime import datetime, timedelta, timezone
from bs4 import BeautifulSoup 
import logging

RE = 6378.137
LANGUAGE = 'ja'
AWS_REGION='us-west-2'
AWS_ACCESS_KEY_ID=<YOUR AWS ACCESS KEY ID>
AWS_SECRET_ACESS_KEY=<YOUR AWS SECRET_ACCESS_KEY>
GEOCODE_API_KEY=<YOUR GEOCODE API KEY>

logger = logging.getLogger()
logger.setLevel(logging.INFO)

def check_email(ses, email):
    result = ses.get_identity_verification_attributes(Identities=[email])
    attr = result['VerificationAttributes']
    print(attr)
    if (email not in attr or attr[email]['VerificationStatus'] != 'Success'):
        ses.verify_email_identity(EmailAddress=email)
        return False
    return True

def get_coordinate(lat, lng):
    h = math.radians(lat)
    t = math.radians(lng)
    x = math.cos(h) * math.cos(t)
    y = math.cos(h) * math.sin(t)
    z = math.sin(h)
    return (x, y, z)

def calc_distance(base_coordinate, lat, lng):
    coordinate = get_coordinate(lat, lng)
    ip = base_coordinate[0]*coordinate[0] 
    ip += base_coordinate[1]*coordinate[1] 
    ip += base_coordinate[2]*coordinate[2]
    return RE * math.acos(ip)

def find_near_station(base_lat, base_lng):
    tuples = list()
    base_coordinate = get_coordinate(base_lat, base_lng)
    with open('./station_data.json') as f:
        json_dict = json.load(f)
        stations = json_dict['station']
        for r in stations:
            if 'lat' in r and 'lng' in r :
                d = calc_distance(base_coordinate, r['lat'], r['lng'])
                tp = (r['name'], str(d))
                tuples.append(tp)

        sorted_tuples = sorted(tuples, key=lambda t: t[1])
    return (sorted_tuples[0][0], sorted_tuples[0][1])

def get_transfer_navitime(station_from, station_to,year,month,day,hour,minute):

    t = station_from + ' ⇒ ' + station_to + ' : '

    url = 'https://www.navitime.co.jp/transfer/searchlist?'
    url += 'orvStationName=' + station_from + '&'
    url += 'dnvStationName=' + station_to + '&'
    url += 'thrStationName1=&'
    url += 'thrStationCode1=&'
    url += 'thrStationName2=&'
    url += 'thrStationCode2=&'
    url += 'thrStationName3=&'
    url += 'thrStationCode3=&'
    url += 'month=' + year + '%2F' + month +'&'
    url += 'day=' + day +'&' 
    url += 'hour=' + hour+ '&'
    url += 'minute=' + minute + '&'
    url += 'orvStationCode=&'
    url += 'dnvStationCode=&'
    url += 'basis=1&'
    url += 'from=view.transfer.top&'
    url += 'sort=0&'
    url += 'wspeed=100&'
    url += 'airplane=1&'
    url += 'sprexprs=1&'
    url += 'utrexprs=1&'
    url += 'othexprs=1&'
    url += 'mtrplbus=1&'
    url += 'intercitybus=1&'
    url += 'ferry=1&'
    url += 'ctl=020010&'
    url += 'atr=2&'
    url += 'init='
    r = requests.get(url)

    soup = BeautifulSoup(r.text,"html.parser")

    #element 'dt'
    dt = soup.find_all("dt")

    #all elements
    for tag in dt:
        try:
            #retrieve 'class'
            string_ = tag.get("class").pop(0)
            #in case that the class value is 'left'
            if string_ in "left":
                t += tag.string
                break
        except:
            #do nothing
            pass
    return t

def lambda_handler(event, context):
    logging.info('start')
    logging.info('Received event: ' + json.dumps(event))

    # subject & body
    subject = '帰宅のお知らせ'
    body = '----お知らせ----\n'

    # retrieve parameters
    attributes = event['placementInfo']['attributes']
    from_address = attributes['emailFrom']
    to_address = attributes['emailTo']
    destination = attributes['destination']
    s3_bucket_name = attributes['s3bucket']
    s3_object_key = attributes['s3key']
    #clickType = event['deviceEvent']['buttonClicked']['clickType']

    JST = timezone(timedelta(hours=+9), 'JST')
    currentTime = datetime.now(JST).strftime("%Y%m%d%H%M")
    year = currentTime[0:4]
    month = currentTime[4:6]
    day = currentTime[6:8]
    hour = currentTime[8:10]
    minute = currentTime[10:12]

    # create boto object
    ses = boto3.client('ses',
        region_name=AWS_REGION
#        ,aws_access_key_id=AWS_ACCESS_KEY_ID
#        ,aws_secret_access_key=AWS_SECRET_ACESS_KEY
    )
    s3 = boto3.client('s3',
        region_name=AWS_REGION
#        ,aws_access_key_id=AWS_ACCESS_KEY_ID
#        ,aws_secret_access_key=AWS_SECRET_ACESS_KEY
    )

    # retrieve device information (JSON) from S3
    s3_obj = s3.get_object(Bucket=s3_bucket_name, Key=s3_object_key)
    devinfo = json.loads(s3_obj['Body'].read().decode('utf-8'))

    # retrieve street address
    lat = devinfo['latitude']
    lng = devinfo['longitude']
    apikey = GEOCODE_API_KEY
    api_base_url = 'https://maps.googleapis.com/maps/api/geocode/json?language={0}&latlng={1},{2}&key={3}&sensor=false'
    headers = {'content-type': 'application/json'}
    api_url = api_base_url.format(LANGUAGE, str(lat), str(lng), apikey)
    r = requests.get(api_url, headers=headers)
    data = r.json()
    if 'results' in data and len(data['results']) > 0 and 'formatted_address' in data['results'][0]:
        for ret in data['results']:
            body += (ret['formatted_address'] + ' らへんにいます\n')
            break
    
    #get nearest stastion
    station = find_near_station(float(lat), float(lng))
    body += ('最寄駅は' + station[0] + 'です（現在地から駅までの距離は' + str(station[1])[0:4] + 'kmくらいです）\n')
    
    #transfer info
    station_from= station[0]
    station_to = destination
    body += ('着予定は 『' + get_transfer_navitime(station_from,station_to,year,month,day,hour,minute) + '』\n')
    
    body += ('これから帰ります\n')

    #email check
    if not check_email(ses, from_address):
        return 'From email is not verified'

    if not check_email(ses, to_address):
        return 'To email is not verified'

    # send email
    ses.send_email(Source=from_address,
                   Destination={'ToAddresses': [to_address]},
                   Message={'Subject': {'Data': subject}, 
                            'Body': {'Text': {'Data': body}}})
    logging.info('end')
    return 'OK'