Athena, Can I get a Wake Up Call?
Tags: python, twilio, flask
Published: 2019-07-07 22:15:41

---
title: Athena, Can I get a Wake Up Call?
published: true
description: A walkthrough on how I use Twilio's API to create an alarm via Programmable Voice and Autopilot.
tags: python, Twilio, flask
---

Now that the [foundation](https://dev.to/itstayfay/setting-up-the-backend-for-athena-my-personal-assistant-4mb2) 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](#stack)
* [A Quick Note on Using Environment Variables in a Python App](#env)
* [Connecting to a Database](#database)
* [Affirmations](#affirmations)
* [Calling Time!](#call)
* [Am I Awake Yet?](#awake)

<a name = "stack"></a>
### Current Tech Stack
* [Github Repo](https://github.com/TaylorFacen/Athena)
* 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. 

<a name = "env"></a>
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`

```python
import os
from dotenv import load_dotenv

load_dotenv()

DATABASE_URL = os.environ['DATABASE_URL']
```

<a name = "database"></a>
### Connecting to a Database

[Code Checkpoint](https://github.com/TaylorFacen/Athena/tree/b53b899296960b7250bfad6cbcf7575ab12b3a4d)

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`
```python
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`
```python
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. 
![testing error](https://thepracticaldev.s3.amazonaws.com/i/xutoxk3vxggbck2k3h99.png)

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

<a name = "affirmations"></a>
### Affirmations

[Code Checkpoint](https://github.com/TaylorFacen/Athena/tree/d186940d35ea269bcabc332d22789251bea9ba8f)
Next up, let's get some affirmations in the database for Athena to pull from. I'll add an "affirmations" endpoint with some some CRU<del>D</del> (create, read, update, no delete) operations. 

The model for my affirmations table is simple; there's only two columns: affirmation_id, affirmation

```python
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. 

```python
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`
```python
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 CRU<del>D</del> operations to the affirmations table. After creating the affirmations endpoint and adding the response model, flask-restplus automatically updates the api's documentation.

![api documentation via swagger](https://thepracticaldev.s3.amazonaws.com/i/5jgzb6yey65x7080qlvb.png)

![creating an affirmation](https://thepracticaldev.s3.amazonaws.com/i/wduuz9fi347d727jswbg.png)

![getting an affirmation](https://thepracticaldev.s3.amazonaws.com/i/5woscowl8qywajbzctig.png)

![updating an affirmation](https://thepracticaldev.s3.amazonaws.com/i/z3qnqlqemjfrnlv81e95.png)

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.

```python
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."
]
```

<a name = "call"></a>
### Calling Time!
[Code Checkpoint](https://github.com/TaylorFacen/Athena/tree/97ce6696567ff9b823f1aeced7ab0ec36a811107)

Alright, the moment we've all been waiting for - Athena get's a voice! The workflow of the morning call is:
1. Every day at a specific time, Heroku Scheduler will tell my app to run a command that calls my phone
2. This call sends a request to my app with instructions on where to route my call (in this case, a "wake_up_call" task). 
3. 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
4. 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.
```python
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`

```python
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
```python
@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
```python
@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
```python
@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](https://rosettacode.org/wiki/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.


<a name = "awake"></a>
### Am I Awake Yet?
[Code Checkpoint](https://github.com/TaylorFacen/Athena/tree/cd76d524510bcf43ea1ea06b13bbebed9410c46c)

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`
```python
@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`
```python
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}
```

<a name = "daily"></a>
### 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`

![Heroku Scheduler Console](https://thepracticaldev.s3.amazonaws.com/i/8tm63on2zve8ofp6swiw.png)

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. 

[Athena v0.2](https://github.com/TaylorFacen/Athena/releases/tag/v0.2)