Now that the foundation of my bot is set up, I can add my first (and probably my favorite) feature - daily wake up calls. It usually takes a lot to wake me up in the morning since I'm known for hitting the snooze button at least 10 times. I might be able to solve this by having Athena call me every morning, and in order to turn the alarm off, I'll have to repeat that day's affirmation. Seems like a positive and effective way of starting my day!
Contents
- Current Tech Stack
- A Quick Note on Using Environment Variables in a Python App
- Connecting to a Database
- Affirmations
- Calling Time!
- Am I Awake Yet?
Current Tech Stack
- Github Repo
- Heroku
- Python (Flask)
- Postgres Database
- Twilio
The two components that I'll be adding to Athena is a Postgres database and Twilio integration. I wish this was the part where I explain my deep thought process on why I chose SQL over NoSQL. To be completely honest, I decided to go the Postgres route for three main reasons:
- I'm very familiar with SQL
- The Flask ecosystem works well with Postgres
- I recognize that projects evolve over time. If something comes up in the future that warrants a more thorough investigation, I'll do it then. Sometimes it's better to make a decision and go with it until a change is necessary than to spend days laboring over a choice that might not even be that critical.
Integrating with Twilio was a no-brainer. Their API is easy to work with, and their documentation is extremely understandable, even for someone who doesn't have a lot of coding experience. I also appreciate that there's plenty of examples in various programming languages. I was able to use snippets from the Python examples to create almost all of the code used for the wake up call feature.
Alright, that's enough sales pitching. Let's get started.
If you're like me, you HATE editing your local environment variables in your bash_profile. Not only do I have to look up this process every time, I also have a few projects with conflicting variables (e.g. the database for one project won't be the same database for another). I alleviate this headache by using a .env file that contains all of the keys and values that I'll need for my project (of course, this isn't uploaded to my repo).
.env
DATABASE_URL=INSERT_DATABASE_URL_HERE
I then use the python-dotenv package to load all keys and values from the .env file. After this, I can access any environment variable from this file via os.environ!
config.py
import os
from dotenv import load_dotenv
load_dotenv()
DATABASE_URL = os.environ['DATABASE_URL']
Connecting to a Database
I decided to use two separate databases for my project - one for development and testing and the other one for production. This is because no matter what branch or environment code is sitting in, I want it to mimic as much of the real-life production code as possible. One alternative to this approach would be to have a database schema for development and one for testing. While this might work for small projects, I plan on having many tables grouped into schemas (e.g. tasks, budget, health, etc.).
Adding a database is easy. All I had to do was head over to the "Resources" tab on my application, search for "Postgres", and click "provision". That's it!. Now my app has a config variable called "DATABASE_URL" that can be used to access the database.
Next up, I installed Flask-SQLAlchemy, added my database's URL as a local environment variable, and updated my create_app function.
app.py
from flask import Flask
from flask_restplus import Api
from flask_sqlalchemy import SQLAlchemy
db = SQLAlchemy()
from .general import general_api
from config import DATABASE_URL
def create_app():
app = Flask(__name__)
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
app.config['SQLALCHEMY_DATABASE_URI'] = DATABASE_URL
api = Api(
title = "Athena",
version = "0.1",
description = "Athena the personal assistant"
)
# Add namespace
api.add_namespace(general_api, path = '/athena')
# Initialize app
api.init_app(app)
db.init_app(app)
return app
Let's go ahead and add a test to check to see that the app is connected to the database.
test_db.py
from athena import create_app, db
import pytest
@pytest.fixture
def app():
app = create_app()
return app
def test_db_connection(app):
with app.app_context():
result = db.session.execute('SELECT 1').fetchone()
assert result[0] == 1
Alright, let's get this pushed up...oops an error.
One thing to remember about using a Heroku pipeline is that you HAVE to make sure that the environment variables you use locally can also be found in the following locations
- Config Vars section in the app's settings
- The Config Vars section in the pipeline's settings
- the app.json file
After making these changes, everything ran smoothly
Affirmations
Code Checkpoint
Next up, let's get some affirmations in the database for Athena to pull from. I'll add an "affirmations" endpoint with some some CRUD (create, read, update, no delete) operations.
The model for my affirmations table is simple; there's only two columns: affirmation_id, affirmation
from athena import db
class Affirmation(db.Model):
"""Model for affirmations"""
__tablename__ = 'affirmations'
affirmation_id = db.Column(db.String, primary_key = True)
affirmation = db.Column(db.String, nullable = False)
def __repr__(self):
return '<Affirmation {}>'.format(self.affirmation)
def to_dict(self):
"""Converts object to dictionary"""
return {k: v for k, v in self.__dict__.items() if k[0] != '_'}
Adding the following in my app.py tells the app to create the affirmation table if it doesn't already exist.
with app.app_context():
from .models import Affirmation
db.create_all()
Call me paranoid, but I'm always worried that I might accidentally delete a table. So let's just add a test to verify that the affirmations table is there.
test_affirmations.py
from athena import create_app, db
import pytest
@pytest.fixture
def app():
app = create_app()
return app
def test_table_exists(app):
with app.app_context():
query = '''
SELECT
TABLE_NAME
FROM
INFORMATION_SCHEMA.TABLES
WHERE
TABLE_SCHEMA = 'public'
'''
tables_result = db.session.execute(query).fetchall()
tables = map(lambda row: row[0], tables_result)
assert 'affirmations' in tables
Alright, now that that's done, I can add the code to handle any CRUD operations to the affirmations table. After creating the affirmations endpoint and adding the response model, flask-restplus automatically updates the api's documentation.
I went ahead and added tests for each operation as well. After adding in a few affirmations, we can move on to integrating with Twilio.
affirmations = [
'You are exactly where you need to be.',
'I’m living in blessings and abundance in every area of my life.',
'I am gifted and resourceful. I have all the wisdom and power I need to realize my dream.',
'I appreciate every challenge and difficulty in my life. They are the best times for my growth.',
'I am born to be happy. Happiness is my life purpose and compass.',
'All the best opportunities are coming to me for my dream’s realization.',
'I give myself permission to do what is right for me.',
'I allow myself to be who I am without judgment.',
'I give myself the care and attention that I deserve.',
'My drive and ambition allow me to achieve my goals.',
'I am always headed in the right direction.',
'I trust myself to make the right decision.',
'I put my energy into things that matter to me.',
'I am learning valuable lessons from myself every day.',
'I am becoming closer to my true self every day.',
'I make a difference in the world by simply existing in it.',
'I am at peace with who I am as a person.',
"I am the beautiful combination of God's creativity and imagination."
]
Calling Time!
Alright, the moment we've all been waiting for - Athena get's a voice! The workflow of the morning call is:
- Every day at a specific time, Heroku Scheduler will tell my app to run a command that calls my phone
- This call sends a request to my app with instructions on where to route my call (in this case, a "wake_up_call" task).
- The "wake_up_call" task starts off with a Good Morning message, and then sends a request to my app to get instructions on which affirmation to say and how to validate it
- Athena waits for me to repeat the affirmation back and then checks to see if it mostly matches the original phrase (wooh NLP!)
Autopilot Set Up
The two Twilio services that will be used for this feature are Programmable Voice and Autopilot. Voice handles normal calling features, and Autopilot is Twilio's machine learning engine that's used to facilitate user <-> bot interactions. Most of the setup can be done within the console. However, I'm going to programmatically implement some of the setup process so that it's reproducible and tracked with version control.
To connect with Twilio, I installed the Twilio package and added the following to my .env file
ACCOUNT_SID=INSERT_TWILIO_ACCOUNT_SID
AUTH_TOKEN=INSERT_TWILIO_AUTH_TOKEN
TWILIO_NUMBER=INSERT_TWILIO_NUMBER
PHONE_NUMBER=INSERT_PERSONAL_NUMBER
ASSISTANT_SID=INSERT_ASSISTANT_SID
Now my config file looks like this.
import os
from dotenv import load_dotenv
load_dotenv()
# Database
DATABASE_URL = os.environ['DATABASE_URL']
# Twilio Credentials
ACCOUNT_SID = os.environ['ACCOUNT_SID']
AUTH_TOKEN = os.environ['AUTH_TOKEN']
TWILIO_NUMBER = os.environ['TWILIO_NUMBER']
PHONE_NUMBER = os.environ['PHONE_NUMBER']
ASSISTANT_SID = os.environ['ASSISTANT_SID']
The ACCOUNT_SID and AUTH_TOKEN came from my account console, I purchased a TWILIO_NUMBER with calling and messaging capabilities, PHONE_NUMBER is my personal number, and I created a Twilio Autopilot Assistant through the console and wrote down its ASSISTANT_SID. Just so that I don't forget, I'll go ahead and add all of my config variables across the two apps, test settings, and my app.json file.
I added a athena_setup folder that contains the code used to set up her stylesheet (e.g. voice style, failure messages, etc) and another to setup the wake up call task. I also added a model_build.py file that generates a new model build for the assistant. One thing I learned in this process (which hopefully will save you a few headaches) is that you HAVE to run a model build after creating a task or making significant changes to a task. If not, the assistant might not be able to parse any messages you say or text to your bot.
After adding the endpoints for the wake up call workflow, I can push up these changes to Twilio.
Wake Up Call Workflow
The overall set up was pretty straight forward. First, I added a command for initiating the call (requires the Flask_Script package).
manage.py
from flask_script import Manager
from twilio.rest import Client
from athena import create_app
from config import ACCOUNT_SID, AUTH_TOKEN, TWILIO_NUMBER, PHONE_NUMBER, API_BASE
app = create_app()
manager = Manager(app)
@manager.command
def initiate_wake_up_call():
client = Client(ACCOUNT_SID, AUTH_TOKEN)
call = client.calls.create(
url = '{}/assistant/route?target_task=wake_up_call'.format(API_BASE),
to = PHONE_NUMBER,
from_ = TWILIO_NUMBER
)
return call.sid
if __name__ == "__main__":
manager.run()
Then, I added three API calls to handle different aspects of the wake up call.
/route for routing the call to the correct assistant task
@api.route('/route')
class Router(Resource):
@api.expect(route_parser, validate = True)
def post(self):
args = route_parser.parse_args()
target_task = args.get('target_task', None)
response = VoiceResponse()
connect = Connect()
connect.autopilot(ASSISTANT_SID, TargetTask = target_task)
response.append(connect)
response = make_response(str(response))
response.headers['Content-type'] = 'text/html; charset=utf-8'
return response
wake_up_call/dialog for creating the right dialog with a randomly selected affirmation
@api.route('/wake_up_call/dialog')
class WakeUpCallDialog(Resource):
def post(self):
client = current_app.test_client()
resp = client.get('/athena/affirmations?random=True').json
affirmation = resp['affirmation']
affirmation_id = resp['affirmation_id']
return create_wake_up_call_actions(affirmation, affirmation_id)
def create_wake_up_call_actions(affirmation, affirmation_id):
webhook_url = "{}/assistant/wake_up_call/validate?affirmation_id={}".format(API_BASE, affirmation_id)
actions = {
"actions": [
{
"collect": {
"name": "affirmation_dialog",
"questions": [
{
"question": affirmation,
"name": "affirmation_respoinse",
"validate": {
"webhook": {
"url": webhook_url,
"method": "POST"
},
"on_success": {
"say": "Wooh! Let's get the day started."
}
}
}
],
"on_complete": {
"redirect": "task://hello_world"
}
}
}
]
}
return actions
/wake_up_call/validate for checking my response against the original affirmation to validate that I said it correctly
@api.route('/wake_up_call/validate')
class WakeUpCallValidate(Resource):
@api.expect(wake_up_call_validate_parser, validate = True)
def post(self):
args = wake_up_call_validate_parser.parse_args()
affirmation_id = args.get('affirmation_id')
affirmation_response = args.get('ValidateFieldAnswer')
# Get affirmation text
client = current_app.test_client()
resp = client.get('/athena/affirmations?affirmation_id=' + affirmation_id).json
affirmation = resp['affirmation']
return validate_wake_up_call_response(affirmation, affirmation_response)
def validate_wake_up_call_response(affirmation, affirmation_response):
def transform(text):
# Lower text
text_lower = text.lower()
# Strip out punctuation
text_alpha_num = text_lower.translate(str.maketrans('', '', string.punctuation))
return text_alpha_num
affirmation_response = transform(affirmation_response)
affirmation = transform(affirmation)
jaro_distance = jellyfish.jaro_distance(affirmation, affirmation_response)
if jaro_distance >= 0.75:
is_valid = True
else:
is_valid = False
return {"valid": is_valid, 'jaro_distance': jaro_distance}
In order to check my response against the original phrase, there's some initial text transformation that has to be applied. First, both texts are converted to lowercase, and then all punctuation is removed. Then, I use the jellyfish package to compute the Jaro distance between both strings. If the distance is greater than or equal to 0.75 (my personal threshold), then I'll count that as a match.
Am I Awake Yet?
Although I have the best intentions for waking up in the morning, sometimes my "sleepy" self acts against these intentions. So to make sure that I just don't pick up the phone and hang up, I'll make a few tweaks so that Athena calls every 15 minutes if the previous call wasn't successful. Doing this was pretty tricky. Although it's easy to get a list of previous calls, it's not possible to check previous dialogs to see if they're successful. My workaround is that when my '/validate' API says that my response is valid, it will give that call a feedback star count of 5. Then, in my command, I'll have the calling code in a while loop that only breaks if the previous call has a feedback star count of 5.
manage.py
@manager.command
def initiate_wake_up_call():
client = create_twilio_client()
call_complete = False
while not call_complete:
call = client.calls.create(
url = '{}/assistant/route?target_task=wake_up_call'.format(API_BASE),
to = PHONE_NUMBER,
from_ = TWILIO_NUMBER
)
call_sid = call.sid
call_status = 'in-progress'
while call_status != 'completed':
time.sleep(30)
last_call = client.calls(call_sid).fetch()
call_status = last_call.status
try:
feedback_score = last_call.feedback().fetch().quality_score
except:
feedback_score = 1
if feedback_score == 5:
call_complete = True
else:
# Snooze time
time.sleep(15 * 60)
return {'status': 'complete'}
wake_up_call.py
def validate_wake_up_call_response(affirmation, affirmation_response):
client = create_twilio_client()
def transform(text):
# Lower text
text_lower = text.lower()
# Strip out punctuation
text_alpha_num = text_lower.translate(str.maketrans('', '', string.punctuation))
return text_alpha_num
affirmation_response = transform(affirmation_response)
affirmation = transform(affirmation)
jaro_distance = jellyfish.jaro_distance(affirmation, affirmation_response)
# Get the last call (should be current call)
last_call = client.calls.list(limit = 1)[0]
if jaro_distance >= 0.75:
is_valid = True
feedback = client.calls(last_call.sid).feedback().create(quality_score = 5)
else:
is_valid = False
feedback = client.calls(last_call.sid).feedback().create(quality_score = 1)
return {"valid": is_valid, 'jaro_distance': jaro_distance}
Running the task every day
The last thing I need to do is to tell Heroku to run my task every day at a specific time. I'm going to be ambitious here and pick 5:00 AM EST (9:00 AM UTC) as my run time. Heroku makes this process really simple. All I have to do is add the Heroku Scheduler resource to my app and type the following.
python manage.py initiate_wake_up_call
Voila! That's it. After updating Athena's intro and version number and redeploying my task with my production API_BASE, I'm done!
Thanks for reading along! Stay tuned for more walkthroughs.
Top comments (0)